ForkJoinPool 深度解析
理解ForkJoinPool的工作窃取算法、RecursiveTask/RecursiveAction的实现机制，以及在并行流和虚拟线程场景下的最佳实践。
1. 设计哲学与核心概念
1.1 为什么需要ForkJoinPool
传统的线程池（如ThreadPoolExecutor）采用固定数量的工作线程，每条任务队列。当某个线程的任务队列为空时，它只能等待——这造成了CPU资源的浪费。ForkJoinPool的工作窃取（Work-Stealing）算法解决了这个问题。
1.2 Fork/Join 计算模型
ForkJoinPool基于"分治"（Divide and Conquer）思想设计：
// Fork: 将大任务拆分为多个小任务
// Join: 等待所有子任务完成，合并结果
Result compute() {
Result left, right;
if (isLargeProblem(data)) {
// Fork: 拆分为子任务
left = fork(() -> leftTask(data));
right = fork(() -> rightTask(data));
// Join: 等待并合并结果
return merge(left.join(), right.join());
} else {
// 直接计算（不再拆分）
return solveDirectly(data);
}
}1.3 双端队列结构
每个Worker线程拥有自己的双端队列（Deque）：LIFO用于自己执行任务，FIFO用于被其他线程窃取。这确保了本地性（cache-friendly）和负载均衡的平衡。
// 每个Worker的Deque支持三种操作
// 1. Push (LIFO) - 自己的任务入栈
push(task); // 只能自己操作
// 2. Pop (LIFO) - 自己的任务出栈
pop(); // 只能自己操作，用于获取任务
// 3. Take (FIFO) - 窃取别人的任务
take(); // 其他线程可以偷取队列头部
// 示意图
// 队列头部(Take) ← [Task-A][Task-B][Task-C] ← 队列尾部(Push/Pop)
// ↑ ↑
// 被窃取 自己的push/pop工作窃取的效率取决于任务粒度。如果任务太大，窃取带来的收益有限；如果任务太小，管理开销反而会抵消收益。通常，任务应该能花费至少100μs才值得Fork。
2. 核心实现机制
2.1 Worker线程状态机
ForkJoinPool的Worker线程有四种状态，在执行过程中不断转换：
// ForkJoinPool Worker 状态
enum WorkerStatus {
ACTIVE, // 执行任务中
INACTIVE, // 等待新任务（队列为空）
BLOCKED, // 执行join()时阻塞
TERMINATED // 线程结束
}
// 状态转换流程
ACTIVE → (队列空) → INACTIVE → (被窃取) → ACTIVE
ACTIVE → (调用join) → BLOCKED → (任务完成) → ACTIVE2.2 工作窃取算法伪代码
当Worker的本地队列为空时，它会进入"窃取模式"：
void runWorker() {
while (!isTerminated()) {
ForkJoinTask<?> task;
// 1. 先尝试从本地队列获取（LIFO）
task = localQueue.pop();
// 2. 本地队列为空，尝试窃取（FIFO）
if (task == null) {
task = stealFromOther();
}
// 3. 都没有，进入休眠等待
if (task == null) {
waitForWork();
continue;
}
// 4. 执行任务
task.exec();
}
}
ForkJoinTask<?> stealFromOther() {
// 随机选择一个victim线程
int victimIndex = random(poolSize);
Worker victim = workers[victimIndex];
// 从victim队列头部窃取（FIFO，保持公平）
return victim.deque.take();
}2.3 CommonPool
JDK 8引入的ForkJoinPool.commonPool()是全局共享的池，被Java并行流（Parallel Streams）默认使用：
// 并行流默认使用commonPool
List<String> results = dataList
.parallelStream()
.map(this::process)
.collect(Collectors.toList());
// 获取commonPool
ForkJoinPool common = ForkJoinPool.commonPool();
// commonPool默认并行度 = CPU核心数 - 1
// 可通过系统属性配置
// -Djava.util.concurrent.ForkJoinPool.common.parallelism=8
// 提交任务到commonPool
ForkJoinPool.commonPool().submit(() -> {
// your task
});3. RecursiveTask 与 RecursiveAction
3.1 有返回值的任务：RecursiveTask
RecursiveTask<V>用于有返回值的递归任务，例如并行求和：
class SumTask extends RecursiveTask<Long> {
private static final long THRESHOLD = 10_000;
private final long[] array;
private final int start, end;
SumTask(long[] array, int start, int end) {
this.array = array;
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
int length = end - start;
// 小于阈值，直接计算
if (length <= THRESHOLD) {
long sum = 0;
for (int i = start; i < end; i++) {
sum += array[i];
}
return sum;
}
// 拆分任务
int mid = start + length / 2;
SumTask left = new SumTask(array, start, mid);
SumTask right = new SumTask(array, mid, end);
// Fork: 异步执行左半部分
left.fork();
// 直接计算右半部分（当前线程）
Long rightResult = right.compute();
// Join: 获取左半部分结果
Long leftResult = left.join();
return leftResult + rightResult;
}
}
// 使用
ForkJoinPool pool = ForkJoinPool.commonPool();
Long result = pool.invoke(new SumTask(data, 0, data.length));3.2 无返回值的任务：RecursiveAction
RecursiveAction用于无返回值的递归任务，例如并行数组排序：
class SortTask extends RecursiveAction {
private static final int THRESHOLD = 1000;
private int[] array;
private int low, high;
SortTask(int[] array, int low, int high) {
this.array = array;
this.low = low;
this.high = high;
}
@Override
protected void compute() {
int length = high - low;
if (length <= THRESHOLD) {
Arrays.sort(array, low, high);
return;
}
int mid = low + length / 2;
SortTask left = new SortTask(array, low, mid);
SortTask right = new SortTask(array, mid, high);
// Fork both tasks asynchronously
forkAll(left, right);
// forkAll会等待所有任务完成（隐式join）
}
}不要在RecursiveTask中混用fork()和compute()的顺序。先fork再join的顺序可能导致深度递归时线程耗尽。正确做法是：总是尝试用当前线程执行一个子任务（compute），只fork另一个。
4. ForkJoinPool 与虚拟线程
4.1 虚拟线程的默认carrier
JDK 21的虚拟线程默认使用ForkJoinPool作为carrier thread pool。这意味着当虚拟线程调用阻塞操作时，carrier thread可以执行其他虚拟线程的任务。
// 虚拟线程的调度器
// 默认使用 ForkJoinPool.commonPool() 作为carrier
// 创建虚拟线程时，它会被调度到carrier上
try ( ExecutorService executor =
Executors.newVirtualThreadPerTaskExecutor()) {
Future<String> f = executor.submit(() -> {
// 虚拟线程执行时，可能被调度到
// ForkJoinPool.common() 的worker线程上
return blockingCall();
});
}4.2 在虚拟线程中使用ForkJoinPool
需要特别注意：在虚拟线程中直接使用ForkJoinPool可能不是最佳选择。因为虚拟线程的"阻塞-调度"机制与ForkJoinPool的工作窃取模型不同，混用可能导致性能问题。
// 虚拟线程中调用并行流
List<String> results = List.of("a", "b", "c")
.parallelStream() // 使用commonPool
.map(this::blockingTransform) // 阻塞操作
.toList();
// 问题：并行流的commonPool worker是平台线程
// 阻塞操作会占用平台线程，降低并行度
// 更好的做法：在虚拟线程中顺序执行，或使用专用池
ExecutorService vtPool = Executors.newVirtualThreadPerTaskExecutor();
List<String> results = List.of("a", "b", "c")
.stream() // 顺序流
.map(s -> vtPool.submit(() -> this.blockingTransform(s)))
.map(Future::join)
.toList();5. 性能调优与最佳实践
5.1 任务粒度控制
任务粒度是ForkJoinPool性能的关键。太粗会导致负载不均，太细会带来调度开销。
// 阈值设置原则
// 1. 计算密集型任务：阈值 = 总数据量 / (4 * CPU核心数)
// 2. 混合任务（IO+计算）：阈值 = 1000 ~ 10000
// 示例：并行排序
class ParallelMergeSort<T extends Comparable<T>> extends RecursiveAction {
// 动态计算阈值
private int computeThreshold() {
Runtime rt = Runtime.getRuntime();
int cpus = rt.availableProcessors();
return Math.max(1000, array.length / (cpus * 4));
}
// 或使用系统属性
// -Dforkjoin.threshold=10000
}5.2 线程池配置
| 场景 | 并行度 | 说明 |
|---|---|---|
| 计算密集型 | CPU核心数 | 太多线程会导致上下文切换 |
| 混合IO型 | CPU核心数 × 2 | 线程可在IO时让出CPU |
| 并行流（默认） | CPU核心数 - 1 | 留一个给主线程 |
| 虚拟线程carrier | 动态 | 按需创建/销毁 |
ForkJoinPool vs ThreadPoolExecutor的选择：计算密集、可以分解为相似粒度的任务，用ForkJoinPool；异步IO、任务差异大、用Future管理依赖，用ThreadPoolExecutor。