了解一下 Java21 的 VirtualThread

作者：袁世超

2024-01-17
北京
本文字数：2257 字
阅读完需：约 7 分钟

0、自上而下

从用户接口的角度，虚拟线程的使用方式即 ThreadPerTaskExecutor，只是这里为每个任务创建的线程是虚拟线程
从任务执行的角度，任务包装为 Continuation 后提交到 ForkJoinPool 执行
从调度的角度，Continuation 是“可挂起”的任务，执行->挂起->唤醒->执行，如此反复直至任务执行完毕

1、Continuation

按照 Ron Pressler 在《Loom项目提案》中的描述：

Continuation 是 Java 中的『协程』抽象，这是个有栈协程（相对于 Rust async 的无栈协程），但是这个栈不包含 JNI stack frames，换句话说就是无法在 JNI 调用中挂起，有个专有的名词 Pinned 描述这种情况，后面会介绍。

如果要在外部使用 Continuation 需要添加参数 --add-exports java.base/jdk.internal.vm=ALL-UNNAMED

import jdk.internal.vm.Continuation;import jdk.internal.vm.ContinuationScope;
public class ContinuationDemo {    public static void main(String[] args) {        ContinuationScope scope = new ContinuationScope("scope");        Continuation continuation = new Continuation(scope, () -> {            System.out.println("Running before yield");            Continuation.yield(scope);            System.out.println("Running after yield");        });
        System.out.println("First run");        continuation.run();        System.out.println("isDone: " + continuation.isDone());;
        System.out.println("Second run");        continuation.run();        System.out.println("isDone: " + continuation.isDone());;
        // java.lang.IllegalStateException: Continuation terminated        //continuation.run();    }}

复制代码

第 9 行的 yield 就会让 VM 记录当前的栈帧，让出线程；
第 18 行再次 run 的时候，VM 就会恢复栈帧，从上次让出的地方继续执行。

ContinuationScope 标识 Continuation 的作用域，该 API 支持层次结构的 Continuation

import jdk.internal.vm.Continuation;import jdk.internal.vm.ContinuationScope;
public class MultiContinuationDemo {    public static void main(String[] args) {        ContinuationScope scope = new ContinuationScope("scope");        ContinuationScope scope1 = new ContinuationScope("scope1");        ContinuationScope scope2 = new ContinuationScope("scope2");        Continuation child2 = new Continuation(scope2, () -> {            System.out.println("[Child2] Running before yield");            Continuation.yield(scope);            System.out.println("[Child2] Running after yield");        });        Continuation child1 = new Continuation(scope1, () -> {            System.out.println("[Child1] Running before child2");            child2.run();            System.out.println("[Child1] Running after child2");        });        Continuation continuation = new Continuation(scope, () -> {            System.out.println("[Main] Running before child1");            child1.run();            System.out.println("[Main] Running after child1");        });
        System.out.println("First run");        continuation.run();        System.out.println("isDone: " + continuation.isDone());;
        System.out.println("Second run");        continuation.run();        System.out.println("isDone: " + continuation.isDone());;
    }}

复制代码

但是 VirtualThread 创建 VThreadContinuation 时传入的 scope 是固定的，所以在虚拟线程的场景对 Continuation 的使用还是比较简单。

“栈帧”的记录与恢复逻辑是在 JVM 中实现的，Java 代码中只留了一个尾巴 StackChunk，这里不展开，详见《Continuations - Under the Covers》

2、VirtualThread

VirtualThread 对 Continuation 进一步封装，另外还声明了调度器的实现。

2.1、调度器

DEFAULT_SCHEDULER 是一个 ForkJoinPool，用于任务的执行。

UNPARKER 是一个调度线程池，用于唤醒定时的任务，例如 Thread.sleep(100)。

2.2、线程状态

VirtualThead 在 ForkJoinPool 的『平台线程』中开始执行，就会将自身状态设置为 RUNNING，遇到“挂起”的操作就会转为 PARKING 和 YIELDING 状态，然后让出『平台线程』。

YIELDING 和 PARKING 状态

“挂起”操作的入口主要有两个：

Thread.yield --> VirtualThread.tryYield --> yieldContinuation
LockSupport.park --> System.parkVirtualThread --> VirtualThread.park[Nanos] --> yieldContinuation