提升系統(tǒng)吞吐量，詳解JDK21虛擬線程，炸裂
環(huán)境：JDK21
1. 虛擬線程簡介
虛擬線程是輕量級的線程，可以減少編寫、維護和調試高吞吐量并發(fā)應用程序的工作量。線程是可以調度的最小處理單元。它與其他類似單元并發(fā)運行，而且在很大程度上是獨立運行的。它是java.lang.Thread的一個實例。線程有兩種，平臺線程和虛擬線程。
2. 什么是平臺線程
平臺線程被實現為操作系統(tǒng)(OS)線程的薄包裝器。平臺線程在其底層操作系統(tǒng)線程上運行Java代碼，平臺線程在平臺線程的整個生命周期中捕獲其操作系統(tǒng)線程。因此，可用平臺線程的數量受限于操作系統(tǒng)線程的數量。
平臺線程通常有一個比較大的線程堆棧和由操作系統(tǒng)維護的其他資源。它們適用于運行所有類型的任務，但可能是有限的資源。
3. 什么是虛擬線程
與平臺線程一樣，虛擬線程也是java.lang.thread的一個實例。然而，虛擬線程并沒有綁定到特定的操作系統(tǒng)線程。虛擬線程仍然在操作系統(tǒng)線程上運行代碼。但是，當虛擬線程中運行的代碼調用阻塞I/O操作時，Java運行時會掛起虛擬線程，直到可以恢復為止。與掛起的虛擬線程相關聯(lián)的OS線程現在可以自由地執(zhí)行其他虛擬線程的操作。
虛擬線程的實現方式與虛擬內存類似。為了模擬大量內存，操作系統(tǒng)將大量虛擬地址空間映射到有限的RAM。同樣，為了模擬大量線程，Java運行時將大量虛擬線程映射到少量操作系統(tǒng)線程。
與平臺線程不同，虛擬線程通常有一個淺調用堆棧，只執(zhí)行一個HTTP客戶端調用或一個JDBC查詢。盡管虛擬線程支持線程本地變量和可繼承的線程本地變量，但應該仔細考慮使用它們，因為單個JVM可能支持數百萬個虛擬線程。
虛擬線程適用于運行大部分時間被阻塞的任務，這些任務通常等待I/O操作完成。然而，它們并不適用于長時間運行的CPU密集型操作。
4. 為什么使用虛擬線程
在高吞吐量并發(fā)應用程序中使用虛擬線程，尤其是那些由大量并發(fā)任務組成、花費大量時間等待的應用程序。服務器應用程序是高吞吐量應用程序的示例，因為它們通常處理許多執(zhí)行阻塞I/O操作（如獲取資源）的客戶端請求。
虛擬線程不是更快的線程；它們運行代碼的速度并不比平臺線程快。它們的存在是為了提供規(guī)模（更高的吞吐量），而不是速度（更低的延遲）。
5. 創(chuàng)建虛擬線程
Thread和Thread.Builder APIs提供了創(chuàng)建平臺線程和虛擬線程的方法。java.util.concurrent.Executors類還定義了創(chuàng)建ExecutorService的方法，該方法為每個任務啟動一個新的虛擬線程。
5.1 Thread & Thread.Builder創(chuàng)建虛擬線程
調用Thread.ofVirtual()方法創(chuàng)建一個Thread.Builder實例，用于創(chuàng)建虛擬線程。如下示例：
Thread t= Thread.ofVirtual().start(() -> System.out.println("Hello")) ;
t.join() ;Thread.Builder接口允許創(chuàng)建具有公共線程屬性(如線程名稱)的線程。Thread.Builder.OfPlatform子接口創(chuàng)建平臺線程，而Thread.Builder.OfVirtual創(chuàng)建虛擬線程。
下面的示例使用Thread.Builder接口創(chuàng)建一個名為T-VM的虛擬線程，如下示例：
Thread.Builder builder = Thread.ofVirtual().name("T-VM") ;
Runnable task = () -> {
System.out.println("執(zhí)行任務") ;
} ;
Thread t = builder.start(task) ;
System.err.printf("線程名稱: %s%n", t.getName()) ;
t.join() ;輸出結果：
執(zhí)行任務
線程名稱T-VM下面的示例創(chuàng)建并啟動兩個具有Thread.Builder的虛擬線程：
Thread.Builder builder = Thread.ofVirtual().name("vm - worker - ", 0);
Runnable task = () -> {
System.out.printf("線程ID: %d%n", Thread.currentThread().threadId());
} ;
// 線程 "vm - worker - 0"
Thread t1 = builder.start(task) ;
t1.join();
System.out.println(t1.getName() + " terminated") ;
// 線程 "vm - worker - 1"
Thread t2 = builder.start(task) ;
t2.join() ;
System.out.println(t2.getName() + " terminated") ;輸出結果：
線程ID: 21
vm - worker - 0 terminated
線程ID: 24
vm - worker - 1 terminated以上是通過Thread.Builder創(chuàng)建虛擬線程的簡單示例。
5.2 Executors創(chuàng)建虛擬線程
Executors允許將線程管理和創(chuàng)建與應用程序的其余部分分離。
ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor() ;
// submit Runnable任務
Future<?> future = executor.submit(() -> System.out.println("Running thread")) ;
future.get() ;
System.out.println("Task completed") ;上面示例每當調用ExecutorService.submit(Runnable)時，都會創(chuàng)建一個新的虛擬線程并開始運行該任務。
6. 虛擬線程調度
操作系統(tǒng)在平臺線程運行時進行調度。然而，Java運行時會在虛擬線程運行時進行調度。當Java運行時調度虛擬線程時，它將虛擬線程分配或掛載到平臺線程上，然后操作系統(tǒng)像往常一樣調度該平臺線程。這個平臺線程稱為載體(carrier)。運行一些代碼后，虛擬線程可以從它的載體卸載。這通常發(fā)生在虛擬線程執(zhí)行阻塞I/O操作時。當一個虛擬線程從其宿主中卸載后，宿主就處于空閑狀態(tài)，這意味著Java運行時調度器可以在其上分配另一個虛擬線程。
虛擬線程被綁定到其宿主（平臺線程）時，在阻塞操作期間無法將其卸載。虛擬線程在以下情況下會被綁定：
- 虛擬線程在同步塊或方法內運行代碼
- 虛擬線程運行本機方法或外部函數
7. 虛擬線程應用指南
虛擬線程是由Java運行時而不是操作系統(tǒng)實現的Java線程。虛擬線程和傳統(tǒng)線程(平臺線程)的主要區(qū)別在于，可以很容易地在同一個Java進程中運行大量甚至數百萬個活動的虛擬線程。正是它們的數量賦予了虛擬線程強大的能力，通過允許服務器并發(fā)處理更多請求，它們可以更高效地運行以"thread-per-request"風格編寫的服務器應用程序，從而提高吞吐量，減少硬件浪費。
虛擬線程可以顯著提高以thread-per-request風格編寫的服務器的吞吐量，而不是延遲。在這種風格中，服務器在整個持續(xù)時間內使用一個線程來處理每個傳入的請求。它至少占用一個線程，因為在處理單個請求時，你可能希望使用更多線程并發(fā)地執(zhí)行某些任務。
阻塞平臺線程的代價是昂貴的，因為它會占用線程——這是一種相對稀缺的資源——而線程并沒有做很多有意義的工作。因為虛擬線程可能很多，所以阻塞它們是廉價的，也是值得鼓勵的。因此，應該使用簡單的同步風格并使用阻塞I/O API編寫代碼。
如下代碼以非阻塞異步風格編寫，不會從虛擬線程中獲得太多好處。
HttpClient client = ... ;
Executor pool = Executors.newVirtualThreadPerTaskExecutor() ;
CompletableFuture.supplyAsync(() -> {
HttpRequest request = HttpRequest.newBuilder(URI.create("http://localhost:8088/users/info")).build() ;
BodyHandler<String> bodyHandler = ... ;
try {
return client.send(request , bodyHandler) ;
}
}, pool)
.thenCompose(url -> getBodyAsync(url, HttpResponse.BodyHandlers.ofString()))
.thenApply(info::findImage)
.thenAccept(this::process)
.exceptionally(t -> { t.printStackTrace(); return null; });相反，以下代碼以同步風格編寫，并使用簡單的阻塞IO，將受益匪淺：
try {
String page = getBody(info.getUrl(), HttpResponse.BodyHandlers.ofString());
String imageUrl = info.findImage(page);
byte[] data = getBody(imageUrl, HttpResponse.BodyHandlers.ofByteArray());
info.setImageData(data);
process(info);
}8. 不要池化虛擬線程
關于虛擬線程最難理解的是，雖然它們與平臺線程具有相同的行為，但它們不應該表示相同的程序概念。
平臺線程很少，因此是一種寶貴的資源。寶貴的資源需要管理，管理平臺線程的最常見方法是使用線程池。接下來你需要回答的問題是，線程池中應該有多少個線程?
但是虛擬線程是很多的，因此每個線程不應該代表某種共享的、池化的資源，而應該代表一個任務。
將n個平臺線程轉換為n個虛擬線程幾乎沒有好處；相反，需要轉換的是任務。
為了將每個應用任務表示為一個線程，不要像下面的例子那樣使用共享線程池:
Future<ResultA> f1 = sharedThreadPoolExecutor.submit(task1);
Future<ResultB> f2 = sharedThreadPoolExecutor.submit(task2);
// ... use futures相反，使用虛擬線程
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
Future<ResultA> f1 = executor.submit(task1);
Future<ResultB> f2 = executor.submit(task2);
// ... use futures
}上面的代碼仍然使用ExecutorService，但從Executors.newVirtualThreadPerTaskExecutor()返回的代碼沒有使用線程池。相反，它為每個提交的任務創(chuàng)建一個新的虛擬線程。
此外，ExecutorService本身是輕量級的，我們可以像創(chuàng)建任何簡單對象一樣創(chuàng)建一個新的對象。這使得我們可以依賴新添加的ExecutorService#close方法和try-with-resources構造。close方法會在try塊結束時隱式調用，它會自動等待所有提交給ExecutorService的任務(即由ExecutorService生成的所有虛擬線程)結束。
對于扇出場景來說，這是一個特別有用的模式，在這種場景中，同時執(zhí)行多個對不同服務調用，如下面的示例所示:
void handle() throws Exception {
URL url1 = URI.create("http://www.pack.com").toURL() ;
URL url2 = URI.create("http://www.akf.com").toURL() ;
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
var future1 = executor.submit(() -> fetchURL(url1));
var future2 = executor.submit(() -> fetchURL(url2));
System.out.printf("result1: %s, result2: %s%n", future1.get(),future2.get()) ;
}
}
String fetchURL(URL url) throws IOException {
try (var in = url.openStream()) {
return new String(in.readAllBytes(), StandardCharsets.UTF_8);
}
}你應該創(chuàng)建一個新的虛擬線程，如上所示，即使是小的、短期的并發(fā)任務。
9. 避免長時間頻繁的Pinning
當前實現虛擬線程的一個限制是，在同步的塊或方法內部執(zhí)行阻塞操作會導致JDK的虛擬線程調度器阻塞一個操作系統(tǒng)線程，而在同步的塊或方法外部執(zhí)行阻塞操作則不會。這種情況稱為“Pinning”。如果阻塞操作持續(xù)時間長且頻繁，Pinning可能會對服務器的吞吐量產生負面影響。保護短期的操作，例如內存操作，或者使用同步塊或方法的不頻繁操作，應該不會有任何負面影響。
對于長時間又頻繁的地方應該使用ReentrantLock替換synchronized 。
synchronized(lockObj) {
frequentIO() ;
}
// 替換為
lock.lock();
try {
frequentIO() ;
} finally {
lock.unlock() ;
}
