java 虚拟线程学习

虚拟线程作为 jdk 19+ 最重要的特性之一, 足以对所有 java 开发者构成巨大的吸引力; 我们即便无法很快在生产环境中升级到最新版本的 jdk, 但也应该足够重视, 尽早学习, 跟上时代节奏, 避免被职场淘汰;

虚拟线程相关的 jvm 配置参数

# ForkJoinPool 默认并发度 / 平台线程 (物理 OS 线程) 数
-Djdk.virtualThreadScheduler.parallelism=128
# 调度器的最大并发度
-Djdk.virtualThreadScheduler.maxPoolSize=2048
# 最少运行的线程数
-Djdk.virtualThreadScheduler.minRunnable=32

# 追踪被 pin 住的虚拟线程: full 表示全量采样 / short 表示部分采样
-Djdk.tracePinnedThreads=full/short

虚拟线程的使用

方式一: 直接构造 ThreadFactory 并调用其 newThread 方法

public static final ThreadFactory newVirtualThreadFactory(final String prefix) {
    return Thread.ofVirtual()
        .name(prefix + "-")
        .uncaughtExceptionHandler((t, e) -> {
            log.error("Uncaught exception, thread:[{}]", t, e);
        })
        .factory();
}

final ThreadFactory vTreadFactory = newVirtualThreadFactory("myThreadPrefix-");
final Thread vThread = vThreadFactory.newThread(() -> {......});
vThread.start();
if (vThread.getState() == Thread.State.TERMINATED) {
    throw new IllegalStateException("thread is terminated");
}

方式二: 使用 Executors 封装的工具方法

// 默认无线程名, 为空字符串
var executor = Executors.newVirtualThreadPerTaskExecutor();
// 指定线程名前缀及起始计数, 线程名: myThreadPrefix-${i}
var executor = Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("myThreadPrefix-", 1).factory());

虚拟线程的火焰图

使用虚拟线程的注意事项

池化问题

传统的 java 线程与 OS 线程是一一对应的, 创建代价较为昂贵, 在 Thread-Per-Request 编程模式下, 请求数远超 OS 所支持的最大线程数, 因此我们通常会使用线程池 (ThreadPoolExecutor) 来避免不必要的线程创建和销毁; 与此同时, ThreadPoolExecutor 内部存在较为复杂的状态管理, 并引入阻塞队列 workQueue, 在特定状态下引导 command 排队挂起/唤醒, 这是管理池化资源不得不引入的复杂性;
而虚拟线程本质上只是一个轻量级的 java 对象, 创建和回收的成本很低, 在这种情况下如果使用复杂的 ThreadPoolExecutor 管理 virtual thread, 显得有些小题大做, 因此 java 官方不推荐对虚拟线程做池化缓存;
但虚拟线程作为 java 对象毕竟还是占用有限的内存资源的, 不可能无限创建, 对此 java 官方推荐直接使用 java.util.concurrent.Semaphore (信号量) 来限制 virtual thread 的总量; 同 ThreadPoolExecutor 中的 Worker 类似, Semaphore 也是基于 AbstractQueuedSynchronizer 的同步器, 但 Semaphore 只关心资源总量的限制, 没有 ThreadPoolExecutor 其他的复杂管理逻辑, 使用效率更高;

FJP 虚拟线程回收速度

在 jdk 21 下, 当使用 virtual thread 的应用流量卸掉后, ForkJoinPool 线程回缩得很慢, 每次 keepAlive 时间后只缩小 1 个 virtual thread, 官方回复 jdk 22 修复: JDK-8319662;

pinning 问题

受限于操作系统或 virtual thread 的实现, 有一些阻塞操作是无法被 carrier 线程卸载的, 即虚拟线程会被禁锢在了 carrier 线程上, 致使 carrier 线程无法调度其他虚拟线程, 我们应当尽可能避免这种情况的发生, 例如:

synchronized 代码块内的阻塞操作, 比如 synchronized 里去调用 Thread#sleep;
- 典型 bad case: guava cache 使用了 synchronized, 所以不能在使用了 guava cache 的业务应用里直接使用 virtual thread, 需要先替换为 caffeine cache;
- java 官方对虚拟线程未来支持 synchronized 的态度: 不够重视, 预计很难在 jdk 23 之前支持;
Object#wait() 方法的执行;
native 方法的执行;
外部函数的执行;
大部分对文件系统的操作;

类加载 (致命弱点)

1
2
3

-Djdk.virtualThreadScheduler.parallelism=1
-Djdk.virtualThreadScheduler.maxPoolSize=1
-Djdk.virtualThreadScheduler.minRunnable=1

public class VTTest {
    static final CountDownLatch countDownLatch = new CountDownLatch(1);
    static final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();

    public static void main(String[] args) throws IOException {
        executor.execute(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            countDownLatch.countDown();
        });

        executor.execute(InnerSleepClass::hello);
        System.in.read();
    }

    private static class InnerSleepClass {
        static {
            try {
                countDownLatch.await();
                System.out.println("exit count down.");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        static void hello() {
            System.out.println("hello");
        }
    }
}

当 OS 线程因为这些特殊的阻塞操作无法卸载虚拟线程时, 调度器 (ForkJoinPool) 会临时增大并发度, 可通过该选项来调整最大并发度:

1	-Djdk.virtualThreadScheduler.maxPoolSize=2048

JDK 也提供了工具帮助我们发现这些无法被卸载的阻塞操作:

1
2
3

# full 表示全量采样
# short 表示部分采样
-Djdk.tracePinnedThreads=full/short

采样举例:

Thread[#22,ForkJoinPool-1-worker-1,5,CarrierThreads]
    java.base/java.lang.VirtualThread$VThreadContinuation.onPinned(VirtualThread.java:185)
    java.base/jdk.internal.vm.Continuation.onPinned0(Continuation.java:393)
    java.base/java.lang.VirtualThread.parkNanos(VirtualThread.java:634)
    java.base/java.lang.VirtualThread.sleepNanos(VirtualThread.java:806)
    java.base/java.lang.Thread.sleep(Thread.java:594)
    org.example.Main.lambda$executeTest$0(Main.java:35) <== monitors:1
    java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)
    java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
    java.base/java.lang.VirtualThread.run(VirtualThread.java:314)

参考链接

ForkJoinPool trims worker threads too slowly