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

2024-07-22
福建
本文字数：3296 字
阅读完需：约 11 分钟

环境：JDK21 + IDEA

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）。

这种方式存在以下问题：

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

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

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

什么是虚拟线程

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

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

虚拟线程实例化


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 的线程池，可以看到池子的几个配置参数。

最大 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 状态：
阻塞操作在 synchronized 代码块中（后续 JDK 可能优化这一点限制）
执行 native 方法时
Pinned 状态占用了 Platform 线程，无疑会影响性能，官方建议对于经常执行的 synchronized 代码块，最好使用 java.util.concurrent.locks.ReentrantLock 替代。如果不清楚自己代码里哪些地方使用到了 synchronized 代码块，在切换使用虚拟线程时，可以添加 JVM 参数 jdk.tracePinnedThreads 帮助排查。

总结

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

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

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

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

文章转载自：小江的学习日记
原文链接：https://www.cnblogs.com/whatever2code/p/18314433
体验地址：http://www.jnpfsoft.com/?from=infoq

发布于: 刚刚阅读数: 2

不在线第一只蜗牛

关注

还未添加个人签名 2023-06-19 加入

还未添加个人简介

发布

暂无评论

创作场景