从平台线程到虚拟线程
要理解虚拟线程，首先要明白传统 Java 线程（平台线程）的局限性。平台线程 (Platform Thread) 是直接映射到操作系统内核线程 (OS Thread)。创建成本高（分配几 MB 栈内存）、上下文切换开销大。这导致一台机器很难支撑即几十上百万个并发连接。比如传统的 tomcat、jetty 等容器都使用了传统的现成模型。
- 当用户发起请求之后 ，tomcat 容器会为每个请求创建一个线程去处理这个请求；
- 比如这个线程要读取数据库，那么这个线程就会创建一个网络IO，在等待数据库响应的时间内，CPU是处于空闲状态的，但是这个线程并没有被释放；
- 在并发量非常大情况下，就会出现 CPU 使用率非常低，但占用的线程（平台线程的上限受制于内核数量）已经用完的情况。tomcat 容器使用线程池技术池化线程，只是减少了线程创建和销毁的开销，并不能提高可用线程数量的上限。
- 在过去的一段时间里，我们大多采用异步编程的方式解决这个问题，也就是采用非阻塞模型。但是在异步编程的代码里充满了回调的函数，学习成本很高，也不便于调试和代码的维护。
- 正是在这个背景之下，虚拟线程被提出来，用于提高传统的平台线程并发上限的限制。
虚拟线程初探
虚拟线程也许是 Java21最重大的革新。虚拟线程是以平台线程为载体的轻量级线程，它运行在用户空间内，由 JVM 进行管理和调度，它占用资源更少，同同时可以减少线程切换的开销。一个很好的比喻就是：平台线程就好比是我们家里平常用的插座，人生最痛苦的是 “插座用满了但是电却没用完”，虚拟线程就好比是插排，我们在线程上多节几个插排，就可以提高电器同时使用的数量了。
下面是一个简单的例子：
1
2
3
4
5
6
7
8
9
10
11
12
13
| public static void main(String[] args) throws InterruptedException {
Thread.ofPlatform().start(() -> {
System.out.println("平台线程开始跑");
});
Thread vt = Thread.ofVirtual().start(() -> {
System.out.println("虚拟线程开始跑");
});
vt.join();
}
|
ForkJoinPool
再继续介绍虚拟线程之前，我们先来回顾一下 ForkJoinPool。ForkJoinPool 是 Java 7 引入的一个 “专门为递归拆分任务而生的线程池”，它的核心目标是：把一个大任务拆成多个小任务并行算，算完再合并结果。简单来说，ForkJoinPool = 工作窃取（Work-Stealing）+ 分治思想（Divide & Conquer）。
- Fork：把大任务拆成小任务
- Join：等小任务算完，合并结果
- 工作窃取（Work-Stealing）：每个工作线程都有自己的双端任务队列，线程优先从队首取任务执行，如果自己的队列空了，去其他线程的队尾偷任务（Stealing），这就是为什么 ForkJoinPool 在 CPU 密集场景下比普通线程池快很多。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
|
class SumTask extends RecursiveTask<Long> {
private final int[] arr;
private final int start, end;
SumTask(int[] arr, int start, int end) {
this.arr = arr;
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
if (end - start <= 10) {
long sum = 0;
for (int i = start; i < end; i++) {
sum += arr[i];
}
return sum;
}
int mid = (start + end) / 2;
SumTask left = new SumTask(arr, start, mid);
SumTask right = new SumTask(arr, mid, end);
left.fork();
long rightResult = right.compute();
long leftResult = left.join();
return leftResult + rightResult;
}
public static void main(String[] args) {
int[] arr = IntStream.rangeClosed(1, 100).toArray();
ForkJoinPool pool = new ForkJoinPool();
Long result = pool.invoke(new SumTask(arr, 0, arr.length));
System.out.println(result);
}
}
class PrintTask extends RecursiveAction {
@Override
protected void compute() {
}
}
|
当然，实际开发中我们根本没必要自己造轮子，直接按如下止血法是等效的：
1
2
3
4
5
| public static void main(String[] args) throws IOException {
int[] array = IntStream.rangeClosed(1, 100).toArray();
long sum = Arrays.stream(array).parallel().mapToLong(i -> i).sum();
System.out.println(sum);
}
|
实际上，因为 parallel() 默认用的是 ForkJoinPool.commonPool()——JVM级别的全局池。如果你的计算任务涉及 I/O 阻塞 或 需要严格隔离线程资源，就不应该使用 parallelStream()，而应该退回到 ExecutorService / CompletableFuture。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
| import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;
public class Main {
public static void main(String[] args) {
int[] array = IntStream.rangeClosed(1, 100).toArray();
ThreadPoolExecutor executor = new ThreadPoolExecutor(
4,
8,
60L, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(100),
new NamedThreadFactory("array-sum-pool"),
new ThreadPoolExecutor.CallerRunsPolicy()
);
try {
int chunkSize = 25;
List<CompletableFuture<Long>> futures = new ArrayList<>();
for (int i = 0; i < array.length; i += chunkSize) {
int start = i;
int end = Math.min(array.length, i + chunkSize);
CompletableFuture<Long> future = CompletableFuture.supplyAsync(() -> {
long partialSum = 0;
for (int j = start; j < end; j++) {
partialSum += array[j];
}
return partialSum;
}, executor);
futures.add(future);
}
long totalSum = futures.stream()
.mapToLong(CompletableFuture::join)
.sum();
System.out.println("最终计算结果: " + totalSum);
} catch (Exception e) {
System.err.println("计算过程中发生异常: " + e.getMessage());
} finally {
shutdownAndAwaitTermination(executor);
}
}
static class NamedThreadFactory implements ThreadFactory {
private final String prefix;
private final AtomicInteger threadNumber = new AtomicInteger(1);
public NamedThreadFactory(String prefix) {
this.prefix = prefix;
}
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r, prefix + "-" + threadNumber.getAndIncrement());
t.setDaemon(false);
return t;
}
}
private static void shutdownAndAwaitTermination(ExecutorService pool) {
pool.shutdown();
try {
if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
pool.shutdownNow();
if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
System.err.println("线程池未能完全关闭");
}
}
} catch (InterruptedException ie) {
pool.shutdownNow();
Thread.currentThread().interrupt();
}
}
}
|
ForkJoinPool 是 CompletableFuture 以及 Parallel Stream 的默认线程池：
1
2
3
4
5
6
7
8
9
|
CompletableFuture.supplyAsync(() -> doWork());
ExecutorService myPool = Executors.newFixedThreadPool(4);
CompletableFuture.supplyAsync(() -> doWork(), myPool);
list.parallelStream().map(...).sum();
|
虚拟线程的工作原理
- 一个平台线程可以对应多个虚拟线程，平台线程作为一个载体，它是位于线程池中的（ForkJoinPool）。
- 每个平台线程都有一个任务列表，平台线程从任务列表中获取虚拟线程。
- 当虚拟线程启动的时候，会绑定到一个平台线程。
- 当虚拟线程处于阻塞状态的时候，它会从平台线程中进行卸载，这样这个平台线程可以继续运行其他的任务，甚至可以从其他平台线程的任务中获取任务（Work-Stealing）。
- 当虚拟线程回复运行之后，它会重新加入到某个平台线程的任务列表（不一定是原来的平台线程）。也就是说虚拟线程并不是和平台线程一一绑定的，她可能在block之后被重新调度到其他平台线程上。这样我们就可以使用少量的平台线程，去运行大量的虚拟线程。
案例一：
1
2
3
4
5
6
7
8
9
10
11
| public static void test01() {
try (ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor()) {
CompletableFuture<Integer>[] futures = new CompletableFuture[10];
for (int i = 0; i < 10; i++) {
int finalI = i;
futures[i] = CompletableFuture.supplyAsync(() -> getSomeNum(finalI), executorService);
}
CompletableFuture.allOf(futures).join();
}
System.out.println("execute done.");
}
|
案例二：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
| public static void main(String[] args) throws InterruptedException {
try (ExecutorService executorService =
Executors.newVirtualThreadPerTaskExecutor()) {
Set<Long> threadIds = ConcurrentHashMap.newKeySet();
Set<String> platformThreadNames = ConcurrentHashMap.newKeySet();
long start = System.currentTimeMillis();
IntStream.rangeClosed(1, 1_000_000).forEach(i -> {
executorService.submit(() -> {
try {
Thread.sleep(Duration.ofSeconds(1));
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
Thread thread = Thread.currentThread();
threadIds.add(thread.threadId());
platformThreadNames.add(thread.toString().split("@")[1]);
return i;
});
});
executorService.shutdown();
boolean finished = executorService.awaitTermination(2, TimeUnit.MINUTES);
System.out.println("Tasks finished: " + finished);
System.out.println("Thread Size: " + platformThreadNames.size());
System.out.println("Platform Thread Size: " + threadIds.size());
System.out.println("耗时：" + (System.currentTimeMillis() - start));
}
}
|
我们创建了100万个虚拟线程，它们平台线程只使用了8个，执行这100万个虚拟线程并发执行这个耗时1秒的任务，也只使用了 15 秒就执行完了。如果们要创建几百万个平台线程，这几乎是不可能的，因为一般的操作系统，它的平台线程数量撑死也就几万个上下。而我们的虚拟线程是丰富而且廉价的资源，单个JVM可以支持数百万个虚拟线程。
虚拟线程的使用场景
虚拟线程不是为了让代码跑得更快（CPU 计算速度不变），而是为了让系统在高并发、高 I/O 阻塞的场景下，能够承载更多的请求（只是增加了并发的规模上限，并没有增加并发的实际能力）。
最适合场景：
- Web 服务器：每个请求一个线程模式（如 Spring Boot）。
- 微服务调用：高频率、高延迟的 RPC 和 HTTP 调用。
- 数据库交互：频繁访问 DB 的应用。
- 通常我们不需要直接使用虚拟线程，像 tomcat、jetty、netty、spring boot 等都支持虚拟线程。
不适合场景：
- CPU 密集型任务：如加密、图像处理、深度学习运算。虚拟线程无法提高运算速度，反而会引入调度开销。
- 长时间锁定在 synchronized 块中：目前的 JVM 对 synchronized 处理已改进，但若在 synchronized 中进行阻塞操作，仍可能导致“固定（Pinning）”，即虚拟线程无法被卸载，从而降低并发度。
使用建议：
不要池化虚拟线程：虚拟线程的初衷就是“用完即丢”，不要像管理数据库连接池一样去管理它。不要创建 VirtualThreadPool。
监控 Pinning 问题：如果你的代码在执行 I/O 时没有卸载（Pinning），说明可能是在 synchronized 块中。尽量使用 ReentrantLock 替代 synchronized。
注意资源限制：虚拟线程虽然多，但数据库连接池、文件句柄等资源是有限的。即便有了虚拟线程，连接池依然需要配置（且通常配置不需要像以前那么大）。
如果你因为性能原因一直在维护极度复杂的异步代码（像 WebFlux），可以考虑在升级 JDK 21 后将其重构回简单的同步阻塞代码，逻辑将清晰得多。虚拟线程可以让你的代码回到了 “简单就是美” 的时代——你可以写“看起来像同步”的代码，却拥有 “异步” 的高性能。 对于 95% 以上的业务系统，我强烈推荐虚拟线程（同步编程风格），只有在极少数对极致吞吐量有严苛要求、且需要精细化控制背压（Backpressure）的特定场景下，才考虑使用响应式编程（异步编程风格）。
