承接的问题
卷一第 11 章讲了线程池的 ctl 与 Worker 生命周期，那里解决的是"任务太多、线程太少"。工作窃取要解决的是另一个问题：任务已经拆成了很多份，怎么让闲下来的线程去帮别人干活，同时又不让所有人挤在同一端抢。
旧教材讲这一段时，通常会画出"每个线程一个双端队列，自己从头部取、小偷从尾部偷"，然后贴出 scan(WorkQueue) 和 steal() 两个方法。这两个方法在 JDK 25 的源码里已经不存在了：本机 ForkJoinPool.java（4422 行）中 scan(WorkQueue 出现 0 次，steal( 作为方法名 0 次，pollAtMost 0 次。现在承担"找活干"职责的是 pollScan(boolean submissionsOnly)，承担"偷一个"职责的是 WorkQueue.poll()。
五个字段，一个环形数组
先看 JDK 25 里 WorkQueue 的字段。这不是从旧文章抄的，是本机 javap -p 与 src.zip 对得上的版本：
// JDK 25 源码摘录，已删减
static final class WorkQueue {
final ForkJoinWorkerThread owner; // null if shared
ForkJoinTask<?>[] array; // the queued tasks; power of 2 size
int base; // index of next slot for poll
final int config; // mode bits
@jdk.internal.vm.annotation.Contended("w")
int top; // index of next slot for push
@jdk.internal.vm.annotation.Contended("w")
volatile int phase; // versioned active status
@jdk.internal.vm.annotation.Contended("w")
int stackPred; // pool stack (ctl) predecessor link
@jdk.internal.vm.annotation.Contended("w")
volatile int source; // source queue id (or DROPPED)
@jdk.internal.vm.annotation.Contended("w")
int nsteals; // number of steals from other queues
@jdk.internal.vm.annotation.Contended("w")
volatile int parking; // nonzero if parked in awaitWork
}
五个字段的职责可以直接从注释读出来：
| 字段 | 语义 | 谁写 |
|---|---|---|
array | 任务槽数组，长度是 2 的幂，下标用 cap - 1 掩码 | owner 扩容时换新数组 |
base | 下一个可被"取走"的槽位，只增不减 | 任何非 owner 的窃取者，以及 owner 在 FIFO 模式下 |
top | 下一个可被"压入"的槽位，只增不减 | 只有 owner |
config | 模式位：FIFO = 1 << 0、CLEAR_TLS = 1 << 1、PRESET_SIZE = 1 << 2 | 构造时定死，final |
phase | 低 16 位是队列 id，第 16 位是 IDLE，其余位做版本号 | seqlock 的两端 |
两个下标都不回绕，只是用 (cap - 1) & index 映射到数组槽位。于是"队列长度"就是 top - base，queueSize() 的实现只有一行 Math.max(top - base, 0)。top 和 phase 都标了 @Contended，因为它们被不同线程高频读写，和 base 挤在同一条缓存行上会互相打脸——这是 12.2 要展开的竞争问题的一半。
owner 为 null 的队列是"提交队列"（shared queue）。池里的 queues 数组把两类队列按奇偶分开：worker 队列在奇数下标，提交队列在偶数下标。源码注释给的理由很直接：扫描时两类队列要按相近概率采样，放在同一个数组里最简单；而奇偶又能让扫描者不看内容就知道类型。pollScan(true) 正是靠这个约定，把偶数下标一跳一跳地扫。
javap -p 对本机 WorkQueue 的字段输出（节选）：
final class java.util.concurrent.ForkJoinPool$WorkQueue {
final java.util.concurrent.ForkJoinWorkerThread owner;
java.util.concurrent.ForkJoinTask<?>[] array;
int base;
final int config;
int top;
volatile int phase;
private static final jdk.internal.misc.Unsafe U;
private static final long PHASE, BASE, TOP, ARRAY; // 原文逐行列出，此处合并
// stackPred / source / nsteals / parking 的 javap 行此处省略
}
owner 走 nextLocalTask，窃取者走 poll
同一个队列，两端各有入口：
// JDK 25 源码摘录，已删减
private ForkJoinTask<?> nextLocalTask(int fifo) {
ForkJoinTask<?> t = null;
ForkJoinTask<?>[] a = array;
int b = base, p = top, cap;
if (p - b > 0 && a != null && (cap = a.length) > 0) {
for (int m = cap - 1, s, nb;;) {
if (fifo == 0 || (nb = b + 1) == p) {
if ((t = (ForkJoinTask<?>)U.getAndSetReference(
a, slotOffset(m & (s = p - 1)), null)) != null)
updateTop(s); // else lost race for only task
break;
}
if ((t = (ForkJoinTask<?>)U.getAndSetReference(
a, slotOffset(m & b), null)) != null) {
updateBase(nb);
break;
}
while (b == (b = U.getIntAcquire(this, BASE)))
Thread.onSpinWait(); // spin to reduce memory traffic
if (p - b <= 0)
break;
}
}
return t;
}
fifo 来自 config & FIFO，只有一个 bit，所以 nextLocalTask() 无参版本就是 nextLocalTask(config & FIFO)。这段代码里有三条容易读漏的分支：
第一，fifo == 0（默认的 LIFO 模式）时走 top - 1 那一格，用 getAndSetReference 一次拿走。先压后取，局部性好，最近 fork 的子任务大概率还在缓存里。
第二，fifo != 0（asyncMode = true）时走 base 那一格，也就是 owner 自己从队头取，变成 FIFO。这样队列里排最久的任务先执行，适合"事件流"而不是"分治树"。
第三，(nb = b + 1) == p 这个条件：队列里只剩一个任务时，base 和 top 两端指向同一格，这时 owner 必须从 top - 1 取，否则会和窃取者抢同一格。这是两端设计的边界，不是优化。
窃取者没有 nextLocalTask，它用 poll()：
// JDK 25 源码摘录，已删减
/**
* Polls for a task. Used only by non-owners.
*/
final ForkJoinTask<?> poll() {
for (int pb = -1, b; ; pb = b) { // track progress
ForkJoinTask<?> t; int cap, nb; long k; ForkJoinTask<?>[] a;
if ((a = array) == null || (cap = a.length) <= 0)
break;
t = (ForkJoinTask<?>)U.getReferenceAcquire(
a, k = slotOffset((cap - 1) & (b = base)));
Object u = U.getReference( // next slot
a, slotOffset((cap - 1) & (nb = b + 1)));
if (base != b) // inconsistent
;
else if (t == null) {
if (u == null && top - b <= 0)
break; // empty
if (pb == b)
Thread.onSpinWait(); // stalled
}
else if (U.compareAndSetReference(a, k, t, null)) {
updateBase(nb);
return t;
}
}
return null;
}
注释写的是 “Used only by non-owners”。要补一句准确的：runWorker 的顶层循环并没有调用 poll()，它把同样形状的"读 base 槽 → CAS 抢走 → 推进 base"内联进了自己的扫描循环；poll() 本身只被 pollScan 调用，服务的是外部路径（pollSubmission()、drainTasksTo()）；helpJoin 另有自己内联的轮询循环。读源码时不要因为 runWorker 里没有 q.poll() 就以为这个方法没人用。
扫描：随机起点与 submissionsOnly
pollScan 只有二十来行，但把"避免系统性不公平"这件事做完了：
// JDK 25 源码摘录，已删减
/**
* Scans for and returns a polled task, if available. Used only
* for untracked polls. Begins scan at a random index to avoid
* systematic unfairness.
*
* @param submissionsOnly if true, only scan submission queues
*/
private ForkJoinTask<?> pollScan(boolean submissionsOnly) {
if ((runState & STOP) == 0L) {
WorkQueue[] qs; int n; WorkQueue q; ForkJoinTask<?> t;
int r = ThreadLocalRandom.nextSecondarySeed();
if (submissionsOnly) // even indices only
r &= ~1;
int step = (submissionsOnly) ? 2 : 1;
if ((qs = queues) != null && (n = qs.length) > 0) {
for (int i = n; i > 0; i -= step, r += step) {
if ((q = qs[r & (n - 1)]) != null &&
(t = q.poll()) != null)
return t;
}
}
}
return null;
}
三件事：
- 随机起点：
ThreadLocalRandom.nextSecondarySeed()取一个种子，r & (n - 1)落到某个队列。如果所有人都从下标 0 开始扫，下标 0 的队列会被反复抢，后面的队列永远没人看。 submissionsOnly只扫偶数下标：r &= ~1把最低位清零，步长 2。这样扫到的必是提交队列（owner == null），用于pollSubmission()这种只想拿"用户提交、还没开始跑"的任务的场景。- 扫一整圈：
i从n减到 0，覆盖全部槽位；r每轮加步长，靠掩码回绕。遇到null槽或poll()返回null就继续下一个。
和旧版对照，JDK 8 的 ForkJoinPool.scan(WorkQueue) 把扫描、失败重试和 worker 激活揉在一个方法里。JDK 25 把它拆成两处：worker 自己的扫描在 runWorker 里内联并带 rescan 标记，外部扫描收进 pollScan。名字和切分变了，随机化起点、失败就换目标这两个原则没变。
getSurplusQueuedTaskCount：为什么能减少窃取
ForkJoinTask.getSurplusQueuedTaskCount() 是公开 API，池里那个同名方法是包私有的：
// JDK 25 源码摘录，已删减
static int getSurplusQueuedTaskCount() {
Thread t; ForkJoinWorkerThread wt; ForkJoinPool pool; WorkQueue q;
if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) &&
(pool = (wt = (ForkJoinWorkerThread)t).pool) != null &&
(q = wt.workQueue) != null) {
int n = q.top - q.base;
int p = pool.parallelism;
int a = (short)(pool.ctl >>> RC_SHIFT);
return n - (a > (p >>>= 1) ? 0 :
a > (p >>>= 1) ? 1 :
a > (p >>>= 1) ? 2 :
a > (p >>>= 1) ? 4 :
8);
}
return 0;
}
它返回的是"我本地队列里比’合理水位’多出来的任务数"。n 是本地队列长度，后面那串三元表达式按活跃线程数 a 相对并行度 p 的比例，给出一个 0/1/2/4/8 的折扣：活跃线程越少，允许自己多囤一点；活跃线程接近并行度时，折扣是 0，多一个任务就算"盈余"。
它的用途不是统计，而是让 fork 的一方在队列已经够深时停止 fork、改为就地执行。源码注释把理由写得很清楚：稳态下如果每个线程都维持 2 个盈余任务，那么"队列长度"本身就是个够用的估计；但在 ramp-up / ramp-down 这类非稳态里会严重误判，所以再用空闲线程数做一次修正。
这件事直接决定窃取次数。本机用 64 个根任务、每个根 fork 256 个子任务（共 16384 个叶子），对比"无条件 fork"和"盈余达到阈值就就地执行"：
surplusLimit=-1 forked=16384 inline= 0 steals= 5377 ms=63
surplusLimit= 0 forked= 105 inline=16279 steals= 168 ms=47
surplusLimit= 2 forked= 5784 inline=10600 steals= 5777 ms=45
surplusLimit= 4 forked= 6907 inline= 9477 steals= 6844 ms=46
（本机 Temurin 25.0.5+8，10 核，load average 82.6；单次运行、随调度波动、不作基准。）
读法：阈值 0 时，只要本地队列非空就不再 fork，16384 个叶子只剩 105 个真正入队，窃取次数从 5377 掉到 168。代价是并行度下降——16279 个叶子是在 fork 它的那个线程里串行跑完的。阈值 2 和 4 的窃取次数反而回到 5000 以上，因为队列里始终有货可偷。这组数说明 getSurplusQueuedTaskCount 不是一个"调优旋钮"，而是一个在任务粒度和调度开销之间选点的开关：设 0 最省调度，但把并行度压到接近 1。
外部提交与内部 fork 的差别
同一批 16 个任务，从主线程 submit 和从 worker 内部 fork，走的是两条路。这个程序可以直接跑：
import java.util.concurrent.*;
public class WhoRuns {
public static void main(String[] a) throws Exception {
ForkJoinPool p = new ForkJoinPool(4);
for (int i = 0; i < 16; i++) { int id = i;
p.submit(() -> System.out.println("submit-" + id + "@" + Thread.currentThread().getName())); }
Thread.sleep(300); // 让 A 组先跑完
p.submit(() -> { for (int i = 0; i < 16; i++) { int id = i;
new RecursiveAction() { protected void compute() {
System.out.println("fork-" + id + "@" + Thread.currentThread().getName()); } }.fork(); } }).join();
Thread.sleep(300);
p.shutdown();
}
}
本机一次运行：
submit-0@ForkJoinPool-1-worker-1
submit-2@ForkJoinPool-1-worker-3
...
fork-15@ForkJoinPool-1-worker-4
fork-1@ForkJoinPool-1-worker-1
fork-0@ForkJoinPool-1-worker-3
fork-2@ForkJoinPool-1-worker-2
...
fork-12@ForkJoinPool-1-worker-4
差别在两端：
- 外部提交进提交队列（偶数下标，
owner == null），worker 从base端取；队列内部是 FIFO，但四个 worker 并发地取，打印顺序会被调度打散——上面submit-0、submit-2、submit-1就是。 - 内部 fork 进自己的 worker 队列（奇数下标），
fork-15第一个跑：owner 从top端弹，后进先出；worker-1、worker-2、worker-3从base端偷走1、0、2，worker-4继续弹14、13、12。两端在同一个数组上相向而行，这就是 12.2 的撞车点。
同一程序连跑三次，行的顺序不完全相同，但"最后 fork 的先被 owner 弹出、先 fork 的被别人从 base 偷走"这条规律每次都成立。外部提交也会被"帮助执行"：主线程调用 parallelStream 时自己会参与跑任务，12.3 会给数据。
在 libraryd-conc 里落地
月末盘点要为一个分馆的 40 万条库存记录生成"按分类聚合的可借/在借/预约"快照。这是典型的可拆分聚合：每条记录独立计算，结果按分类合并。用 RecursiveTask 实现，拆到叶子后直接累加，不在叶子里做 I/O：
package com.plume.conc.core;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
/** 分馆库存快照：按分类聚合，纯内存计算，I/O 在调用方完成。 */
public final class InventorySnapshot {
public record Row(String isbn, String category, int onShelf, int onLoan, int reserved) {}
private static final class Sum extends RecursiveTask<Map<String, int[]>> {
private final List<Row> rows;
private final int lo, hi;
private final int threshold;
private final int surplusLimit;
Sum(List<Row> rows, int lo, int hi, int threshold, int surplusLimit) {
this.rows = rows; this.lo = lo; this.hi = hi;
this.threshold = threshold; this.surplusLimit = surplusLimit;
}
@Override
protected Map<String, int[]> compute() {
if (hi - lo <= threshold) {
Map<String, int[]> acc = new HashMap<>();
for (int i = lo; i < hi; i++) {
Row r = rows.get(i);
int[] v = acc.computeIfAbsent(r.category(), k -> new int[3]);
v[0] += r.onShelf();
v[1] += r.onLoan();
v[2] += r.reserved();
}
return acc;
}
int mid = (lo + hi) >>> 1;
Sum left = new Sum(rows, lo, mid, threshold, surplusLimit);
Sum right = new Sum(rows, mid, hi, threshold, surplusLimit);
// 本地队列已经够深时不再 fork，直接在当前线程跑左半，减少窃取
if (surplusLimit < 0
|| ForkJoinTask.getSurplusQueuedTaskCount() < surplusLimit) {
left.fork();
Map<String, int[]> acc = right.compute();
merge(acc, left.join());
return acc;
}
Map<String, int[]> acc = left.compute();
merge(acc, right.compute());
return acc;
}
private static void merge(Map<String, int[]> into, Map<String, int[]> from) {
from.forEach((k, v) -> {
int[] t = into.computeIfAbsent(k, x -> new int[3]);
t[0] += v[0]; t[1] += v[1]; t[2] += v[2];
});
}
}
public static Map<String, int[]> snapshot(List<Row> rows, int threshold, int surplusLimit) {
return ForkJoinPool.commonPool()
.invoke(new Sum(rows, 0, rows.size(), threshold, surplusLimit));
}
}
三个决策值得说明。第一，叶子只做内存累加，不在叶子里查数据库或写文件——12.3 会用数据说明为什么阻塞任务不能放进 commonPool。第二，每层用 HashMap 而不是共享的 ConcurrentHashMap，合并时只在一个线程里做，避免每条记录都去抢同一个 CounterCell。第三，surplusLimit 是显式参数，上面那组实验说明它直接决定窃取次数与并行度的平衡点。
这个设计的代价
第一，两端队列的内存布局很贵。top、phase、stackPred、source、nsteals、parking 六个字段共用一个 @Contended("w") 分组，整组与 base、array 隔开，代价是填充；一个 WorkQueue 对象在 32 GB 的机器上可能占掉几百字节，队列数组本身另算。并行度 9 的池子，队列数组初始是 64 格，扩容按 2 倍走。对小任务来说，分配队列的成本可能比任务本身还高。
第二，外部提交要抢锁。submissionQueue 用 tryLockPhase() 在提交队列上做自旋锁，抢不到就换一个下标。提交线程多的时候，这一步的失败率不低——12.2 会给出 phase 的原始值。
常见错误与边界
把 getSurplusQueuedTaskCount() 当性能指标打日志。 它的返回值依赖当前线程是不是 ForkJoinWorkerThread，外部线程调用恒为 0。在监控里打这个值，看到的全是 0，然后误判"没有积压"。
在 compute() 里做阻塞 I/O。 fork 出去的任务如果卡在 socket 上，worker 不会自动补偿（只有 ManagedBlocker 才会触发 tryCompensate）。并行度 9 的池子会被 9 个阻塞任务占死。
按旧版方法名搜索。 搜 scan(、steal、pollAtMost 在 JDK 25 里会一无所获，然后得出"实现没变"的错误结论。正确入口是 pollScan、WorkQueue.poll、nextLocalTask。
小结
WorkQueue 是一个用 base/top 两个只增下标描述的双端队列：owner 从 top 压和弹（或按 config & FIFO 从 base 取），窃取者从 base 用 CAS 抢。pollScan 负责"去哪儿找活"，随机起点避免系统性不公平，submissionsOnly 用奇偶下标把提交队列单独筛出来。getSurplusQueuedTaskCount 则是一个反压信号：队列够深就停止 fork，用并行度换调度开销，本机数据里它能把 16384 次 fork 压到 105 次、窃取从 5377 次压到 168 次。
阅读导航：上一节 11.3 DelayedWorkQueue ｜ 下一节 12.2 双向队列与 steal 竞争
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。