探索 Java 并发如何从 Java 8 的增强发展到 Java 21 的虚拟线程，从而实现轻量级、可扩展且高效的多线程处理。
引言
并发编程仍然是构建可扩展、响应式 Java 应用程序的关键部分。多年来，Java 持续增强了其多线程编程能力。本文回顾了从 Java 8 到 Java 21 并发的演进，重点介绍了重要的改进以及 Java 21 中引入的具有重大影响的虚拟线程。
从 Java 8 开始，并发 API 出现了显著的增强，例如原子变量、并发映射以及集成 lambda 表达式以实现更具表现力的并行编程。
Java 8 引入的关键改进包括：
线程与执行器
同步与锁
原子变量与 ConcurrentMap
Java 21 于 2023 年底发布，带来了虚拟线程这一重大演进，从根本上改变了 Java 应用程序处理大量并发任务的方式。虚拟线程为服务器应用程序提供了更高的可扩展性，同时保持了熟悉的"每个请求一个线程"的编程模型。
或许，Java 21 中最重要的特性就是虚拟线程。在 Java 21 中，Java 的基本并发模型保持不变，Stream API 仍然是并行处理大型数据集的首选方式。随着虚拟线程的引入，并发 API 现在能提供更好的性能。在当今的微服务和可扩展服务器应用领域，线程数量必须增长以满足需求。虚拟线程的主要目标是使服务器应用程序能够实现高可扩展性，同时仍使用简单的"每个请求一个线程"模型。
虚拟线程
在 Java 21 之前，JDK 的线程实现使用的是操作系统线程的薄包装器。然而，操作系统线程代价高昂：
虚拟线程的目标是打破 Java 线程与操作系统线程之间的 1:1 关系。虚拟线程应用了类似于虚拟内存的概念。正如虚拟内存将大的地址空间映射到较小的物理内存一样，虚拟线程允许运行时通过将它们映射到少量操作系统线程来制造拥有许多线程的假象。
平台线程是操作系统线程的薄包装器。而虚拟线程并不绑定到任何特定的操作系统线程。虚拟线程可以执行平台线程可以运行的任何代码。这是一个主要优势——现有的 Java 代码通常无需修改或仅需少量修改即可在虚拟线程上运行。虚拟线程由平台线程承载，这些平台线程仍然由操作系统调度。
例如，您可以像这样创建一个使用虚拟线程的执行器：
ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
复制代码
对比示例
虚拟线程仅在主动执行 CPU 密集型任务时才消耗操作系统线程。虚拟线程在其生命周期内可以在不同的载体线程上挂载或卸载。
通常，当虚拟线程遇到阻塞操作时，它会自行卸载。一旦该阻塞任务完成，虚拟线程通过挂载到任何可用的载体线程上来恢复执行。这种挂载和卸载过程频繁且透明地发生——不会阻塞操作系统线程。
示例 — 源代码
Example01CachedThreadPool.java
在此示例中，使用缓存线程池创建了一个执行器：
var executor = Executors.newCachedThreadPool()
复制代码
package threads;
import java.time.Duration;import java.util.concurrent.Executors;import java.util.stream.IntStream;
/** * * @author Milan Karajovic <milan.karajovic.rs@gmail.com> * */
public class Example01CachedThreadPool {
public void executeTasks(final int NUMBER_OF_TASKS) {
final int BLOCKING_CALL = 1; System.out.println("Number of tasks which executed using 'newCachedThreadPool()' " + NUMBER_OF_TASKS + " tasks each.");
long startTime = System.currentTimeMillis();
try (var executor = Executors.newCachedThreadPool()) {
IntStream.range(0, NUMBER_OF_TASKS).forEach(i -> { executor.submit(() -> { // 模拟阻塞调用 Thread.sleep(Duration.ofSeconds(BLOCKING_CALL)); return i; }); });
} catch (Exception e) { throw new RuntimeException(e); }
long endTime = System.currentTimeMillis(); System.out.println("For executing " + NUMBER_OF_TASKS + " tasks duration is: " + (endTime - startTime) + " ms"); }
}
复制代码
package threads;
import org.junit.jupiter.api.MethodOrderer;import org.junit.jupiter.api.Order;import org.junit.jupiter.api.Test;import org.junit.jupiter.api.TestMethodOrder;
/** * * @author Milan Karajovic <milan.karajovic.rs@gmail.com> * */
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)public class Example01CachedThreadPoolTest {
@Test @Order(1) public void test_1000_tasks() { Example01CachedThreadPool example01CachedThreadPool = new Example01CachedThreadPool(); example01CachedThreadPool.executeTasks(1000); }
@Test @Order(2) public void test_10_000_tasks() { Example01CachedThreadPool example01CachedThreadPool = new Example01CachedThreadPool(); example01CachedThreadPool.executeTasks(10_000); }
@Test @Order(3) public void test_100_000_tasks() { Example01CachedThreadPool example01CachedThreadPool = new Example01CachedThreadPool(); example01CachedThreadPool.executeTasks(100_000); }
@Test @Order(4) public void test_1_000_000_tasks() { Example01CachedThreadPool example01CachedThreadPool = new Example01CachedThreadPool(); example01CachedThreadPool.executeTasks(1_000_000); }
}
复制代码
我 PC 上的测试结果：
Example02FixedThreadPool.java
使用固定线程池创建执行器：
var executor = Executors.newFixedThreadPool(500)
复制代码
package threads;
import java.time.Duration;import java.util.concurrent.Executors;import java.util.stream.IntStream;
/** * * @author Milan Karajovic <milan.karajovic.rs@gmail.com> * */
public class Example02FixedThreadPool {
public void executeTasks(final int NUMBER_OF_TASKS) {
final int BLOCKING_CALL = 1; System.out.println("Number of tasks which executed using 'newFixedThreadPool(500)' " + NUMBER_OF_TASKS + " tasks each.");
long startTime = System.currentTimeMillis();
try (var executor = Executors.newFixedThreadPool(500)) {
IntStream.range(0, NUMBER_OF_TASKS).forEach(i -> { executor.submit(() -> { // 模拟阻塞调用 Thread.sleep(Duration.ofSeconds(BLOCKING_CALL)); return i; }); });
} catch (Exception e) { throw new RuntimeException(e); }
long endTime = System.currentTimeMillis(); System.out.println("For executing " + NUMBER_OF_TASKS + " tasks duration is: " + (endTime - startTime) + " ms"); }
}
复制代码
package threads;
import org.junit.jupiter.api.MethodOrderer;import org.junit.jupiter.api.Order;import org.junit.jupiter.api.Test;import org.junit.jupiter.api.TestMethodOrder;
/** * * @author Milan Karajovic <milan.karajovic.rs@gmail.com> * */
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)public class Example02FixedThreadPoolTest {
@Test @Order(1) public void test_1000_tasks() { Example02FixedThreadPool example02FixedThreadPool = new Example02FixedThreadPool(); example02FixedThreadPool.executeTasks(1000); }
@Test @Order(2) public void test_10_000_tasks() { Example02FixedThreadPool example02FixedThreadPool = new Example02FixedThreadPool(); example02FixedThreadPool.executeTasks(10_000); }
@Test @Order(3) public void test_100_000_tasks() { Example02FixedThreadPool example02FixedThreadPool = new Example02FixedThreadPool(); example02FixedThreadPool.executeTasks(100_000); }
@Test @Order(4) public void test_1_000_000_tasks() { Example02FixedThreadPool example02FixedThreadPool = new Example02FixedThreadPool(); example02FixedThreadPool.executeTasks(1_000_000); }
}
复制代码
我 PC 上的测试结果：
Example03VirtualThread.java
使用虚拟线程每任务执行器创建执行器：
var executor = Executors.newVirtualThreadPerTaskExecutor()
复制代码
package threads;
import java.time.Duration;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.stream.IntStream;
/** * * @author Milan Karajovic <milan.karajovic.rs@gmail.com> * */
public class Example03VirtualThread {
public void executeTasks(final int NUMBER_OF_TASKS) {
final int BLOCKING_CALL = 1; System.out.println("Number of tasks which executed using 'newVirtualThreadPerTaskExecutor()' " + NUMBER_OF_TASKS + " tasks each.");
long startTime = System.currentTimeMillis();
try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
IntStream.range(0, NUMBER_OF_TASKS).forEach(i -> { executor.submit(() -> { // 模拟阻塞调用 Thread.sleep(Duration.ofSeconds(BLOCKING_CALL)); return i; }); });
} catch (Exception e) { throw new RuntimeException(e); }
long endTime = System.currentTimeMillis(); System.out.println("For executing " + NUMBER_OF_TASKS + " tasks duration is: " + (endTime - startTime) + " ms"); }
}
复制代码
package threads;
import org.junit.jupiter.api.MethodOrderer;import org.junit.jupiter.api.Order;import org.junit.jupiter.api.Test;import org.junit.jupiter.api.TestMethodOrder;
/** * * @author Milan Karajovic <milan.karajovic.rs@gmail.com> * */
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)public class Example03VirtualThreadTest {
@Test @Order(1) public void test_1000_tasks() { Example03VirtualThread example03VirtualThread = new Example03VirtualThread(); example03VirtualThread.executeTasks(1000); }
@Test @Order(2) public void test_10_000_tasks() { Example03VirtualThread example03VirtualThread = new Example03VirtualThread(); example03VirtualThread.executeTasks(10_000); }
@Test @Order(3) public void test_100_000_tasks() { Example03VirtualThread example03VirtualThread = new Example03VirtualThread(); example03VirtualThread.executeTasks(100_000); }
@Test @Order(4) public void test_1_000_000_tasks() { Example03VirtualThread example03VirtualThread = new Example03VirtualThread(); example03VirtualThread.executeTasks(1_000_000); }
@Test @Order(5) public void test_2_000_000_tasks() { Example03VirtualThread example03VirtualThread = new Example03VirtualThread(); example03VirtualThread.executeTasks(2_000_000); }
}
复制代码
我 PC 上的测试结果：
结论
您可以清楚地看到用于处理所有 NUMBER_OF_TASKS 的不同执行器实现之间的执行时间差异。值得尝试不同的 NUMBER_OF_TASKS 值以观察性能变化。
虚拟线程的优势在处理大量任务时变得尤其明显。当 NUMBER_OF_TASKS 设置为较高的数值时——例如 1,000,000——性能差距是显著的。如下表所示，虚拟线程在处理大量任务时效率要高得多：
我确信，在澄清这一点之后，如果您的应用程序使用并发 API 处理大量任务，您会认真考虑迁移到 Java 21 并利用虚拟线程。在许多情况下，这种转变可以显著提高应用程序的性能和可扩展性。
源代码：GitHub Repository – Comparing Threads in Java 21
【注】本文译自：Java 21 Virtual Threads vs Cached and Fixed Threads
评论