如何通過Loom解決Java并發(fā)模型缺陷

作者 | Deepu K Sasidharan 譯者 | 張衛(wèi)濱 ?

本文最初發(fā)表于 okta 網(wǎng)站，經(jīng)原作者 Deepu K Sasidharan 授權(quán)由 InfoQ 中文站翻譯分享，未經(jīng)許可禁止轉(zhuǎn)載。

Java 19 已經(jīng)于日前發(fā)布，其中最引人注目的特性就要數(shù)虛擬線程了，本文介紹了 Loom 項目中虛擬線程和結(jié)構(gòu)化編程的基礎(chǔ)知識，并將其與操作系統(tǒng)線程進(jìn)行了對比分析。

Java 在其發(fā)展早期就具有良好的多線程和并發(fā)能力，能夠高效地利用多線程和多核 CPU。Java 開發(fā)工具包（Java Development Kit，JDK）1.1 對平臺線程（或操作系統(tǒng)（OS）線程）提供了基本的支持，JDK 1.5 提供了更多的實用工具和更新，以改善并發(fā)和多線程。JDK 8 帶來了異步編程支持和更多的并發(fā)改善。雖然在多個不同的版本中都進(jìn)行了改進(jìn)，但在過去三十多年中，除了基于操作系統(tǒng)的并發(fā)和多線程支持之外，Java 并沒有任何突破性的進(jìn)展。

盡管 Java 中的并發(fā)模型非常強(qiáng)大和靈活，但它并不是最易于使用的，而且開發(fā)人員的體驗也不是很好。這主要是因為它默認(rèn)使用的共享狀態(tài)并發(fā)模型。我們必須借助同步線程來避免數(shù)據(jù)競爭（data race）和線程阻塞這樣的問題。我曾經(jīng)在一篇名為“現(xiàn)代編程語言中的并發(fā)：Java”的博客文章中討論過 Java 并發(fā)問題。

1Loom 項目是什么？

Loom 項目致力于大幅減少編寫、維護(hù)和觀察高吞吐量并發(fā)應(yīng)用相關(guān)的工作，以最佳的方式利用現(xiàn)有的硬件。?

——Ron?Pressler（Loom?項目的技術(shù)負(fù)責(zé)人）

操作系統(tǒng)線程是 Java 并發(fā)模型的核心，圍繞它們有一個非常成熟的生態(tài)系統(tǒng)，但是它們也有一些缺點，如計算方式很昂貴。我們來看一下并發(fā)的兩個最常見使用場景，以及當(dāng)前的 Java 并發(fā)模型在這些場景下的缺點。

最常見的并發(fā)使用場景之一就是借助服務(wù)器在網(wǎng)絡(luò)上為請求提供服務(wù)。在這樣的場景中，首選的方法是“每個請求一個線程（thread-per-request）”模型，即由一個單獨的線程處理每個請求。這種系統(tǒng)的吞吐量可以用 Little 定律來計算，該定律指出，在一個穩(wěn)定的系統(tǒng)中，平均并發(fā)量（服務(wù)器并發(fā)處理的請求數(shù)）L 等于吞吐量（請求的平均速率）λ乘以延遲（處理每個請求的平均時間）W。基于此，我們可以得出，吞吐量等于平均并發(fā)除以延遲（λ = L/W）。

因此，在“每個請求一個線程”模型中，吞吐量將受到操作系統(tǒng)線程數(shù)量的限制，這取決于硬件上可用的物理核心 / 線程數(shù)。為了解決這個問題，我們必須使用共享線程池或異步并發(fā)，這兩種方法各有缺點。線程池有很多限制，如線程泄漏、死鎖、資源激增等。異步并發(fā)意味著必須要適應(yīng)更復(fù)雜的編程風(fēng)格，并謹(jǐn)慎處理數(shù)據(jù)競爭。它們還有可能出現(xiàn)內(nèi)存泄漏、線程鎖定等問題。

另一個常見的使用場景是并行處理或多線程，我們可能會把一個任務(wù)分成跨多個線程的子任務(wù)。此時，我們必須編寫避免數(shù)據(jù)損壞和數(shù)據(jù)競爭的解決方案。在有些情況下，當(dāng)執(zhí)行分布在多個線程上的并行任務(wù)時，還必須要確保線程同步。這種實現(xiàn)會非常脆弱，并且將大量的責(zé)任推給了開發(fā)人員，以確保沒有像線程泄露和取消延遲這樣的問題。

Loom 項目旨在通過引入兩個新特性來解決當(dāng)前并發(fā)模型中的這些問題，即虛擬線程（virtual thread）和結(jié)構(gòu)化并發(fā)（structured concurrency）。

2虛擬線程

Java 19 已經(jīng)于 2022 年 9 月 20 日發(fā)布，虛擬線程是其中的一項預(yù)覽功能。

虛擬線程是輕量級的線程，它們不與操作系統(tǒng)線程綁定，而是由 JVM 來管理。它們適用于“每個請求一個線程”的編程風(fēng)格，同時沒有操作系統(tǒng)線程的限制。我們能夠創(chuàng)建數(shù)以百萬計的虛擬線程而不會影響吞吐。這與 Go 編程語言（Golang）的協(xié)程（如 goroutines）非常相似。

Java 19 中的虛擬線程新特性很易于使用。在這里，我將其與 Golang 的 goroutines 以及 Kotlin 的 coroutines 進(jìn)行了對比。

虛擬線程

Thread.startVirtualThread(() -> {
    System.out.println("Hello, Project Loom!");
});

Goroutine

go func() {
    println("Hello, Goroutines!")
}()

Kotlin coroutine

runBlocking {
    launch {
        println("Hello, Kotlin coroutines!")
    }
}

冷知識：在 JDK 1.1 之前，Java 曾經(jīng)支持過綠色線程（又稱虛擬線程），但該功能在 JDK 1.1 中移除了，因為當(dāng)時該實現(xiàn)并沒有比平臺線程更好。

虛擬線程的新實現(xiàn)是在 JVM 中完成的，它將多個虛擬線程映射為一個或多個操作系統(tǒng)線程，開發(fā)人員可以按需使用虛擬線程或平臺線程。這種虛擬線程實現(xiàn)還有如下幾個注意事項：

在代碼、運行時、調(diào)試器和剖析器（profiler）中，它是一個 Thread。

它是一個 Java 實體，并不是對原生線程的封裝。

創(chuàng)建和阻塞它們是代價低廉的操作。

它們不應(yīng)該放到池中。

虛擬線程使用了一個基于任務(wù)竊?。╳ork-stealing）的 ForkJoinPool 調(diào)度器。

可以將可插拔的調(diào)度器用于異步編程中。

虛擬線程會有自己的棧內(nèi)存。

虛擬線程的 API 與平臺線程非常相似，因此更容易使用或移植。

我們看幾個展示虛擬線程威力的樣例。

線程的總數(shù)量

首先，我們看一下在一臺機(jī)器上可以創(chuàng)建多少個平臺線程和虛擬線程。我的機(jī)器是英特爾酷睿 i9-11900H 處理器，8 個核心、16 個線程、64GB 內(nèi)存，運行的操作系統(tǒng)是 Fedora 36。

平臺線程

var counter = new AtomicInteger();
while (true) {
    new Thread(() -> {
        int count = counter.incrementAndGet();
        System.out.println("Thread count = " + count);
        LockSupport.park();
    }).start();
}

在我的機(jī)器上，在創(chuàng)建 32,539 個平臺線程后代碼就崩潰了。

虛擬線程

var counter = new AtomicInteger();
while (true) {
    Thread.startVirtualThread(() -> {
        int count = counter.incrementAndGet();
        System.out.println("Thread count = " + count);
        LockSupport.park();
    });
}

在我的機(jī)器上，進(jìn)程在創(chuàng)建 14,625,956 個虛擬線程后被掛起，但沒有崩潰，隨著內(nèi)存逐漸可用，它一直在緩慢進(jìn)行。你可能想知道為什么會出現(xiàn)這種情況。這是因為被 park 的虛擬線程會被垃圾回收，JVM 能夠創(chuàng)建更多的虛擬線程并將其分配給底層的平臺線程。

任務(wù)吞吐量

我們嘗試使用平臺線程來運行 100,000 個任務(wù)。

try (var executor = Executors.newThreadPerTaskExecutor(Executors.defaultThreadFactory())) {
    IntStream.range(0, 100_000).forEach(i -> executor.submit(() -> {
        Thread.sleep(Duration.ofSeconds(1));
        System.out.println(i);
        return i;
    }));
}

在這里，我們使用了帶有默認(rèn)線程工廠的 newThreadPerTaskExecutor 方法，因此使用了一個線程組。運行這段代碼并計時，我得到了如下的結(jié)果。當(dāng)使用 Executors.newCachedThreadPool() 線程池時，我得到了更好的性能。

# 'newThreadPerTaskExecutor' with 'defaultThreadFactory'
0:18.77 real,   18.15 s user,   7.19 s sys,     135% 3891pu,    0 amem,         743584 mmem
# 'newCachedThreadPool' with 'defaultThreadFactory'
0:11.52?real,???13.21?s?user,???4.91?s?sys,?????157%?6019pu,????0?amem,?????????2215972?mmem

看著還不錯。現(xiàn)在，讓我們用虛擬線程完成相同的任務(wù)。

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    IntStream.range(0, 100_000).forEach(i -> executor.submit(() -> {
        Thread.sleep(Duration.ofSeconds(1));
        System.out.println(i);
        return i;
    }));
}

運行這段代碼并計時，我得到了如下結(jié)果：

0:02.62?real,???
6.83?s?user,??
??1.46?s?sys,?
????316%?14840pu,??
?0?amem,??????
???350268?mmem

這比基于平臺線程的線程池要好得多。當(dāng)然，這些都是很簡單的使用場景，線程池和虛擬線程的實現(xiàn)都可以進(jìn)一步優(yōu)化以獲得更好的性能，但這不是這篇文章的重點。

用同樣的代碼運行 Java Microbenchmark Harness（JMH），得到的結(jié)果如下?？梢钥吹剑摂M線程的性能比平臺線程要好很多。

# Throughput
Benchmark                             Mode  Cnt  Score   Error  Units
LoomBenchmark.platformThreadPerTask  thrpt    5  0.362 ± 0.079  ops/s
LoomBenchmark.platformThreadPool     thrpt    5  0.528 ± 0.067  ops/s
LoomBenchmark.virtualThreadPerTask   thrpt    5  1.843 ± 0.093  ops/s


# Average time
Benchmark                             Mode  Cnt  Score   Error  Units
LoomBenchmark.platformThreadPerTask   avgt    5  5.600 ± 0.768   s/op
LoomBenchmark.platformThreadPool      avgt    5  3.887 ± 0.717   s/op
LoomBenchmark.virtualThreadPerTask????avgt????5??1.098?±?0.020???s/op

你可以在 GitHub 上找到該基準(zhǔn)測試的源代碼。如下是其他幾個有價值的虛擬線程基準(zhǔn)測試：

在 GitHub 上，Elliot Barlas 使用 ApacheBench 做的一個有趣的基準(zhǔn)測試。

Alexander Zakusylo 在 Medium 上使用 Akka actors 的基準(zhǔn)測試。

在 GitHub 上，Colin Cachia 做的 I/O 和非 I/O 任務(wù)的 JMH 基準(zhǔn)測試。

3結(jié)構(gòu)化并發(fā)

結(jié)構(gòu)化并發(fā)是 Java 19 中的一個孵化功能。

結(jié)構(gòu)化并發(fā)的目的是簡化多線程和并行編程。它將在不同線程中運行的多個任務(wù)視為一個工作單元，簡化了錯誤處理和任務(wù)取消，同時提高了可靠性和可觀測性。這有助于避免線程泄漏和取消延遲等問題。作為一個孵化功能，在穩(wěn)定過程中可能會經(jīng)歷進(jìn)一步的變更。

我們考慮如下這個使用 java.util.concurrent.ExecutorService 的樣例。

void handleOrder() throws ExecutionException, InterruptedException {
    try (var esvc = new ScheduledThreadPoolExecutor(8)) {
        Future inventory = esvc.submit(() -> updateInventory());
        Future order = esvc.submit(() -> updateOrder());


        int theInventory = inventory.get();   // Join updateInventory
        int theOrder = order.get();           // Join updateOrder


        System.out.println("Inventory " + theInventory + " updated for order " + theOrder);
    }
}

我們希望 updateInventory() 和 updateOrder() 這兩個子任務(wù)能夠并發(fā)執(zhí)行。每一個任務(wù)都可以獨立地成功或失敗。理想情況下，如果任何一個子任務(wù)失敗，handleOrder() 方法都應(yīng)該失敗。然而，如果某個子任務(wù)發(fā)生失敗的話，事情就會變得難以預(yù)料。

設(shè)想一下，updateInventory() 失敗并拋出了一個異常。那么，handleOrder() 方法在調(diào)用 invent.get() 時將會拋出異常。到目前為止，還沒有什么大問題，但 updateOrder() 呢？因為它在自己的線程上運行，所以它可能會成功完成。但是現(xiàn)在我們就有了一個庫存和訂單不匹配的問題。假設(shè) updateOrder() 是一個代價高昂的操作。在這種情況下，我們白白浪費了資源，不得不編寫某種防護(hù)邏輯來撤銷對訂單所做的更新，因為我們的整體操作已經(jīng)失敗。

假設(shè) updateInventory() 是一個代價高昂的長時間運行操作，而 updateOrder() 拋出一個錯誤。即便 updateOrder() 拋出了錯誤，handleOrder() 任務(wù)依然會在 inventory.get() 方法上阻塞。理想情況下，我們希望 handleOrder() 任務(wù)在 updateOrder() 發(fā)生故障時取消 updateInventory()，這樣就不會浪費時間了。

如果執(zhí)行 handleOrder() 的線程被中斷，那么中斷不會被傳播到子任務(wù)中。在這種情況下，updateInventory() 和 updateOrder() 會泄露并繼續(xù)在后臺運行。

對于這些場景，我們必須小心翼翼地編寫變通方案和故障防護(hù)措施，把所有的職責(zé)推到了開發(fā)人員身上。

我們可以使用下面的代碼，用結(jié)構(gòu)化并發(fā)實現(xiàn)同樣的功能。

void handleOrder() throws ExecutionException, InterruptedException {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        Future inventory = scope.fork(() -> updateInventory());
        Future order = scope.fork(() -> updateOrder());


        scope.join();           // Join both forks
        scope.throwIfFailed();  // ... and propagate errors


        // Here, both forks have succeeded, so compose their results
        System.out.println("Inventory " + inventory.resultNow() + " updated for order " + order.resultNow());
    }
}

與之前使用 ExecutorService 的樣例不同，我們現(xiàn)在使用 StructuredTaskScope 來實現(xiàn)同樣的結(jié)果，并將子任務(wù)的生命周期限制在詞法的作用域內(nèi)，在本例中，也就是 try-with-resources 語句體內(nèi)。這段代碼更易讀，而且意圖也很清楚。StructuredTaskScope 還自動確保以下行為：

基于短路的錯誤處理：如果 updateInventory() 或 updateOrder() 失敗，另一個將被取消，除非它已經(jīng)完成。這是由 ShutdownOnFailure() 實現(xiàn)的取消策略來管理的，我們還可以使用其他策略。

取消傳播：如果運行 handleOrder() 的線程在調(diào)用 join() 之前或調(diào)用過程中被中斷的話，當(dāng)該線程退出作用域時，兩個分支（fork）都會被自動取消。

可觀察性：線程轉(zhuǎn)儲文件將清楚地顯示任務(wù)層次，運行 updateInventory() 和 updateOrder() 的線程被顯示為作用域的子線程。

4Loom 項目狀況

Loom 項目開始于 2017 年，經(jīng)歷了許多變化和提議。虛擬線程最初被稱為 fibers，但后來為了避免混淆而重新進(jìn)行了命名。如今隨著 Java 19 的發(fā)布，該項目已經(jīng)交付了上文討論的兩個功能。其中一個是預(yù)覽狀態(tài)，另一個是孵化狀態(tài)。因此，這些特性的穩(wěn)定化之路應(yīng)該會更加清晰。

5這對普通的 Java 開發(fā)人員意味著什么？

當(dāng)這些特性生產(chǎn)環(huán)境就緒時，應(yīng)該不會對普通的 Java 開發(fā)人員產(chǎn)生太大的影響，因為這些開發(fā)人員可能正在使用某些庫來處理并發(fā)的場景。但是，在一些比較罕見的場景中，比如你可能進(jìn)行了大量的多線程操作但是沒有使用庫，那么這些特性就是很有價值的了。虛擬線程可以毫不費力地替代你現(xiàn)在使用的線程池。根據(jù)現(xiàn)有的基準(zhǔn)測試，在大多數(shù)情況下它們都能提高性能和可擴(kuò)展性。結(jié)構(gòu)化并發(fā)有助于簡化多線程或并行處理，使其能加健壯，更易于維護(hù)。

6這對 Java 庫開發(fā)人員意味著什么？

當(dāng)這些特性生產(chǎn)環(huán)境就緒時，對于使用線程或并行的庫和框架來說，將是一件大事。庫作者能夠?qū)崿F(xiàn)巨大的性能和可擴(kuò)展性提升，同時簡化代碼庫，使其更易維護(hù)。大多數(shù)使用線程池和平臺線程的 Java 項目都能夠從切換至虛擬線程的過程中受益，候選項目包括 Tomcat、Undertow 和 Netty 這樣的 Java 服務(wù)器軟件，以及 Spring 和 Micronaut 這樣的 Web 框架。我預(yù)計大多數(shù) Java web 技術(shù)都將從線程池遷移到虛擬線程。Java web 技術(shù)和新興的反應(yīng)式編程庫，如 RxJava 和 Akka，也可以有效地使用結(jié)構(gòu)化并發(fā)。但這并不意味著虛擬線程將成為所有問題的解決方案，異步和反應(yīng)式編程仍然有其適用場景和收益。

了解更多關(guān)于 Java、多線程和 Loom 項目的信息：

On the Performance of User-Mode Threads and Coroutines

https://inside.java/2020/08/07/loom-performance/

State of Loom

http://cr.openjdk.java.net/~rpressler/loom/loom/sol1_part1.html

Project Loom: Modern Scalable Concurrency for the Java Platform

https://www.youtube.com/watch?v=EO9oMiL1fFo

Thinking About Massive Throughput? Meet Virtual Threads！

https://foojay.io/today/thinking-about-massive-throughput-meet-virtual-threads/

Does Java 18 finally have a better alternative to JNI?

https://developer.okta.com/blog/2022/04/08/state-of-ffi-java

OAuth for Java Developers