前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >一文揭开 JDK21 虚拟线程的神秘面纱

一文揭开 JDK21 虚拟线程的神秘面纱

原创
作者头像
小江的学习日记
发布2024-07-21 14:15:28
130
发布2024-07-21 14:15:28

虚拟线程快速体验

环境:JDK21 + IDEA

代码语言:java
复制
public static void main(String[] args) {
    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
        IntStream.range(0, 10_000).forEach(i -> {
            executor.submit(() -> {
                Thread.sleep(Duration.ofSeconds(1));
                return i;
            });
        });
    } 
}

运行上面的代码看下执行时间,再试下 Executors.newFixedThreadPool(20) 和 Executors.newCachedThreadPool()

不出意外的话,会发现Executors.newVirtualThreadPerTaskExecutor()运行速度最快,Executors.newCachedThreadPool()运行时系统最卡顿,Executors.newFixedThreadPool(20) 最慢。

Executors.newCachedThreadPool()卡顿是因为一个任务创建一个Platform线程,占用了太多系统资源。

Executors.newFixedThreadPool(20)运行慢是因为只有20个并发去执行1万个任务

Executors.newVirtualThreadPerTaskExecutor()类似Executors.newCachedThreadPool(),但是创建的是虚拟线程,所以在获得高并发的同时也没有占用太多系统资源。

为什么引入虚拟线程

首先,我们来看看现在的Java线程是怎样的。

java.lang.Thread 这个类我相信大家都不陌生,代表Java中的最小并发单元,即一个线程。它是Java对底层的操作系统线程(OS Thread)的封装,为了区别于OS线程,我们称之为平台线程(Platform Thread)。当我们初始化一个Thread实例时,其实就是创建了一个Platform线程并将之与一个OS线程绑定(1:1)。

这种方式存在以下问题:

  1. OS线程是有限的,Platform线程的创建数量受限制于OS线程
  2. 因为绑定系统资源,因此线程的创建/销毁的代价都是昂贵的

这两个问题并非无解,比如,问题1的本质是垂直扩展到顶了,完全可以用水平扩展的方式解决,一台机器的OS线程不能满足需求,再增加一台便是;问题2可以通过池化技术来解决,既然线程的创建和销毁代价比较昂贵,那便将创建好的线程收集起来,推迟销毁的时机,尽量复用它。

JDK21则是在语言层面上的提供了一个替代方案,也就是本文要介绍的虚拟线程(virtual thread),熟悉linux的同学肯定知道系统线程和用户线程的区别,虚拟线程就像是JDK实现的“用户线程”,下面来重点介绍。

什么是虚拟线程

虚拟线程,可以看作是对Platform线程的轻量级封装,Platform线程和OS线程的关系是1:1,虚拟线程和Platform线程的关系则是M:N,且一般M要远远大于N。

可以直接看下虚拟线程的构造函数源码加深理解,坐标java.lang.VirtualThread#<init>。

虚拟线程实例化

代码语言:java
复制
final class VirtualThread extends BaseVirtualThread {
    VirtualThread(Executor scheduler, String name, int characteristics, Runnable task) {
        super(name, characteristics, /*bound*/ false);
        Objects.requireNonNull(task);

        // choose scheduler if not specified
        if (scheduler == null) {
            Thread parent = Thread.currentThread();
            if (parent instanceof VirtualThread vparent) {
                scheduler = vparent.scheduler;
            } else {
                scheduler = DEFAULT_SCHEDULER;
            }
        }

        this.scheduler = scheduler;
        this.cont = new VThreadContinuation(this, task);
        this.runContinuation = this::runContinuation;
    }
}

private static ForkJoinPool createDefaultScheduler() {
        ForkJoinWorkerThreadFactory factory = pool -> {
            PrivilegedAction<ForkJoinWorkerThread> pa = () -> new CarrierThread(pool);
            return AccessController.doPrivileged(pa);
        };
        PrivilegedAction<ForkJoinPool> pa = () -> {
            int parallelism, maxPoolSize, minRunnable;
            String parallelismValue = System.getProperty("jdk.virtualThreadScheduler.parallelism");
            String maxPoolSizeValue = System.getProperty("jdk.virtualThreadScheduler.maxPoolSize");
            String minRunnableValue = System.getProperty("jdk.virtualThreadScheduler.minRunnable");
            if (parallelismValue != null) {
                parallelism = Integer.parseInt(parallelismValue);
            } else {
                parallelism = Runtime.getRuntime().availableProcessors();
            }
            if (maxPoolSizeValue != null) {
                maxPoolSize = Integer.parseInt(maxPoolSizeValue);
                parallelism = Integer.min(parallelism, maxPoolSize);
            } else {
                maxPoolSize = Integer.max(parallelism, 256);
            }
            if (minRunnableValue != null) {
                minRunnable = Integer.parseInt(minRunnableValue);
            } else {
                minRunnable = Integer.max(parallelism / 2, 1);
            }
            Thread.UncaughtExceptionHandler handler = (t, e) -> { };
            boolean asyncMode = true; // FIFO
            return new ForkJoinPool(parallelism, factory, handler, asyncMode,
                         0, maxPoolSize, minRunnable, pool -> true, 30, SECONDS);
        };
        return AccessController.doPrivileged(pa);
    }

可以看到,创建虚拟线程的时候,使用了一个默认的调度器(ForkJoinPool),也就是Platform的线程池,可以看到池子的几个配置参数。

  1. 最大Platform线程数:默认为系统核心数,最大为256,可以通过jdk.virtualThreadScheduler.maxPoolSize设置

这个时候,爱思考的同学可能就要问了,既然默认的最大Platform线程数为系统核心数,岂不是大大限制了并发能力?是不是要主动设置一个较大值?

答案是不需要,因为JDK在线程池的基础上实现了调度的功能。当虚拟线程启动时,调度器会将虚拟线程mount到Platform线程,此时该Platform线程被称为这个虚拟线程的carrier;当线程运行遇到IO操作需要等待时,调度器又会将虚拟现场unmount,把Platform线程释放出来给其他虚拟线程使用,不占用CPU时间。因此,对于非CPU密集的应用,很少的Platform线程就能支持大量的虚拟线程来执行任务。事实上,对于CPU密集的应用,虚拟线程并不会带来多大的提升。虚拟线程真正的应用场景是生存周期短、调用栈浅的任务,如一次http请求、一次JDBC查询。

需要明确的是,操作系统真正能同时运算的线程数也就只有逻辑CPU数,多出来的线程只能等待系统的调度获得CPU时间。

虚拟线程状态

虚拟线程状态流转图
虚拟线程状态流转图

可以看出,虚拟线程相较原先的线程状态,多了Parked、Unparked、Pinned等状态

  • Parked:就是前面说的mount
  • Unparked:就是前面说的unmount
  • Pinned:虚拟线程阻塞时,正常会unmount,但是在一些特殊场景下,不能unmount,此时就会进入Pinned状态:
  1. 阻塞操作在 synchronized 代码块中(后续JDK可能优化这一点限制)
  2. 执行 native 方法时

Pinned状态占用了Platform线程,无疑会影响性能,官方建议对于经常执行的 synchronized 代码块,最好使用java.util.concurrent.locks.ReentrantLock 替代。如果不清楚自己代码里哪些地方使用到了 synchronized 代码块,在切换使用虚拟线程时,可以添加JVM参数jdk.tracePinnedThreads帮助排查。

总结

虚拟线程特别适用如下场景:有大量的并发任务需要执行,且任务是非CPU密集的。

虚拟线程使用上和普通的线程没有太大区别,甚至因为内置了调度逻辑和线程池,可以让开发人员不用再考虑线程池的大小、拒绝策略等,尤其给框架开发者提供了新的优化思路。

对于已经使用了reactive技术的如webFlux框架,没必要再切换到虚拟线程,两者性能相当。

对于web容器如tomcat来说,本身已经使用reactor、nio等技术优化吞吐量,在小的并发数场景下,没必要切换虚拟线程,提升不大。

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 虚拟线程快速体验
  • 为什么引入虚拟线程
  • 什么是虚拟线程
    • 虚拟线程实例化
      • 虚拟线程状态
      • 总结
      相关产品与服务
      容器服务
      腾讯云容器服务(Tencent Kubernetes Engine, TKE)基于原生 kubernetes 提供以容器为核心的、高度可扩展的高性能容器管理服务,覆盖 Serverless、边缘计算、分布式云等多种业务部署场景,业内首创单个集群兼容多种计算节点的容器资源管理模式。同时产品作为云原生 Finops 领先布道者,主导开源项目Crane,全面助力客户实现资源优化、成本控制。
      领券
      问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档