快捷導(dǎo)航

一文詳解JDK21中虛擬線程

更新時(shí)間：2023年10月03日 10:03:47 作者：離離原上草77

虛擬線程是JDK19中引入的,JDK21正式發(fā)布,本文主要介紹了JDK21中虛擬線程,具有一定的參考價(jià)值,感興趣的可以了解一下

支持服務(wù)端應(yīng)用程序以thread-per-request樣式編寫，并最大限度壓榨硬件性能。
兼容java.lang.Thread API，減少調(diào)用方代碼改動(dòng)。
兼容現(xiàn)有的JDK工具，用于故障排查、調(diào)整和分析。
不會(huì)替代線程的傳統(tǒng)實(shí)現(xiàn)，也不會(huì)靜默升級(jí)現(xiàn)有的線程模型
不會(huì)改變Java的并發(fā)編程模型
在Java語(yǔ)言或Java庫(kù)中提供新的數(shù)據(jù)并行結(jié)構(gòu)，不會(huì)動(dòng)搖Stream API的地位。

創(chuàng)建虛擬線程

虛擬線程并沒(méi)有引入獨(dú)立的API，而是選擇兼容Thread的API。這樣已有應(yīng)用程序只需要做最小的改動(dòng)就能使用協(xié)程帶來(lái)的好處。

Thread thread = Thread.ofVirtual().name("duke").unstarted(runnable);
Thread.startVirtualThread(Runnable)
Thread.isVirtual()

為了支持虛擬線程，Thread API被稍微修改，ThreadGroup不再支持

虛擬線程vs平臺(tái)線程

虛擬線程適合大量任務(wù)數(shù)，低CPU負(fù)載的任務(wù)。如果是CPU為瓶頸的應(yīng)用，使用協(xié)程不會(huì)有性能改進(jìn)。

虛擬線程的調(diào)度

Java虛擬線程采用M:N模式，即M個(gè)協(xié)程在N個(gè)線程中運(yùn)行。調(diào)度線程默認(rèn)和核心數(shù)一樣多?？梢酝ㄟ^(guò) system property jdk.virtualThreadScheduler.parallelism來(lái)配置。

虛擬線程的載體線程不是固定的，在生命其中可能運(yùn)行在不同的載體線程中。載體（carrier）線程和虛擬線程的堆棧信息是分離的，不會(huì)相互影響。

ThreadLocal和currentThread獲取的都是獨(dú)立的。在java代碼曾來(lái)說(shuō)虛擬線程后面的系統(tǒng)線程是不可見的。

虛擬線程的執(zhí)行

虛擬線程會(huì)在IO阻塞或者其他等待的時(shí)候掛起，在可用的時(shí)候恢復(fù)執(zhí)行。這不會(huì)阻塞載體線程,載體線程會(huì)去運(yùn)行新任務(wù)。受限于系統(tǒng)，個(gè)別的阻塞操作可能也會(huì)阻塞載體線程，這時(shí)候可能會(huì)臨時(shí)擴(kuò)大調(diào)度池的大小。

一些情況下會(huì)釘住（pinned）載體線程，當(dāng)同步方法/塊或者當(dāng)執(zhí)行本地/外部函數(shù)時(shí)。被釘住的虛擬線程在執(zhí)行很多操作的時(shí)候會(huì)阻塞背后的系統(tǒng)線程。高頻或者長(zhǎng)時(shí)間的synchronized使用java.util.concurrent.locks.ReentrantLock來(lái)代替，可以減少釘住。

JDK Flight Recorder可以查看被釘住的線程。-Djdk.tracePinnedThreads=full參數(shù)將能獲取到完整的釘住線程的堆棧信息。將來(lái)可能會(huì)改進(jìn)pinning inside synchronized的情況。

內(nèi)存使用和垃圾收集器

虛擬線程的堆棧分配在垃圾收集的堆上。棧在運(yùn)行時(shí)的增長(zhǎng)和縮減，所以占用的內(nèi)存少，非常的高效。

虛擬線程不是垃圾收集的根，不再使用的虛擬線程將會(huì)被回收掉。

由于虛擬線程通常比較多，Thread-local開銷就會(huì)比較大，需要謹(jǐn)慎使用。

已有應(yīng)用遷移到虛擬線程

只需要三步：

1.將普通線程的創(chuàng)建改成創(chuàng)建虛擬線程。

2.取消池化機(jī)制。因?yàn)樘摂M線程非常輕量級(jí)，不需要池化。

3.synchronized改為ReentrantLock，以減少被釘住。被釘住的虛擬線程容易阻塞背后的載體線程。

和async/await的比較

async/await比有棧協(xié)程在某些情況下更好用。例如請(qǐng)求c需要a和b作為參數(shù)的情況下，async/await比有棧協(xié)程實(shí)現(xiàn)起來(lái)更加簡(jiǎn)單直接：

a = requestA()
b = requestB()
c = await request(await a, await b)

例

寫一個(gè)簡(jiǎn)單的例子，開n個(gè)虛擬線程，每個(gè)線程sleep 1秒后累加1次，以模擬IO密集型操作。

    private static void runVirtualThread(int length) {
        AtomicInteger ai = new AtomicInteger();
        long start = System.currentTimeMillis();
        try (ExecutorService es = Executors.newVirtualThreadPerTaskExecutor()) {
            for (int i = 0; i < length; i++) {
                es.submit(() -> {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    if (ai.incrementAndGet() >= length) {
                        System.out.printf("duration=%d ms, done.%n", System.currentTimeMillis() - start);
                    }
                });
            }
        }
    }

然后再使用ScheduledExecutorService配合JMXBean打印JVM線程狀態(tài)

        ScheduledExecutorService se = Executors.newScheduledThreadPool(1);
        se.scheduleAtFixedRate(() -> {
            ThreadMXBean threadBean = ManagementFactory.getThreadMXBean();
            ThreadInfo[] threadInfo = threadBean.dumpAllThreads(false, false);
            System.out.printf("threadNumber=%d%n", threadInfo.length);
        }, 100, 1000, TimeUnit.MILLISECONDS);
        runVirtualThread(100000);
        // runThread(100000);
        Thread.sleep(100 * 1000);
        se.shutdownNow();

再寫一個(gè)使用傳統(tǒng)操作系統(tǒng)線程的對(duì)比程序

    private static void runThread(int length) {
        AtomicInteger ai = new AtomicInteger();
        long start = System.currentTimeMillis();
        try (ExecutorService es = Executors.newCachedThreadPool()) {
            for (int i = 0; i < length; i++) {
                es.submit(() -> {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    if (ai.incrementAndGet() >= length) {
                        System.out.printf("duration=%d ms, done.%n", System.currentTimeMillis() - start);
                    }
                });
            }
        }
    }