承接的问题
12.1 把 WorkQueue 拆成了五个字段：base、top、array、config、phase。结构清楚了，问题才刚开始：owner 从 top 端弹，窃取者从 base 端抢，两个方向在同一个环形数组上相向而行，中间只隔着一个"队列长度"。当队列里只剩一个任务时，两端指向同一格，谁赢？
这一节回答三个问题。第一，多个窃取者同时抢 base 那一格，失败会怎样、失败多少次。第二，phase 为什么既像锁又像版本号，外部提交线程和窃取者靠它怎么错开。第三，nsteals 记的是什么，getStealCount() 为什么在池子不安静时会低估。
先给结论：WorkQueue.poll() 是唯一面向非 owner 的取任务入口，它不做锁，只做一次 compareAndSetReference；撞车失败的一方不会排队，而是重读 base 再试，并且用 pb == b 判断"进度停住了"才调 Thread.onSpinWait()。phase 的低 16 位是队列 id，第 16 位是 IDLE，其余位是版本号；tryLockPhase() 是"检查 IDLE 位并用一次 CAS 把它翻掉"，unlockPhase() 是"把 IDLE 加回去"——加一次，版本号也顺带加一。
poll() 的字节码：一次 acquire 读，一次 CAS
先确认入口。JDK 25 里 WorkQueue 没有任何 steal 方法，非 owner 取任务走 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;
}
本机 javap -p -c 对这段的字节码里，关键调用是：
46: invokevirtual #106 // Method jdk/internal/misc/Unsafe.getReferenceAcquire:(Ljava/lang/Object;J)Ljava/lang/Object;
72: invokevirtual #139 // Method jdk/internal/misc/Unsafe.getReference:(Ljava/lang/Object;J)Ljava/lang/Object;
114: invokestatic #130 // Method java/lang/Thread.onSpinWait:()V
129: invokevirtual #146 // Method jdk/internal/misc/Unsafe.compareAndSetReference:(Ljava/lang/Object;JLjava/lang/Object;Ljava/lang/Object;)Z
138: invokevirtual #123 // Method updateBase:(I)V
（本机 Temurin 25.0.5+8，javap -p -c 'java.util.concurrent.ForkJoinPool$WorkQueue' 摘录。）
三处内存语义值得单独指出：
- 取槽用
getReferenceAcquire，保证读到任务引用之后，任务内部的字段（status、结果）也可见。 - 读下一格用普通的
getReference，因为这里只需要"猜"队列是否真的空了，不需要 acquire。 - 抢槽用
compareAndSetReference，成功后updateBase是putIntVolatile，让其它窃取者看到base前进。
注意失败分支是空的 if (base != b) ;——这一行不是笔误，是"别人已经推进了 base，我这轮读到的槽位信息作废，直接重来"。真正的空队列判断放在 t == null 分支里：只有当前格为空、下一格也为空、且 top - b <= 0 三个条件同时成立才退出。为什么看两格？源码注释解释过：非空队列不会连续出现两个空槽（除了扩容和位移这两种可被二次检查识别的瞬态），所以"两格都空"是比"top == base“更省内存流量的近似。
两端为什么会在同一格上撞车
owner 那端的入口是 12.1 里的 nextLocalTask(int fifo)。LIFO 模式下它取 top - 1，FIFO 模式下取 base。撞车发生在三种情形：
情形一：队列只剩一个任务。 top - 1 == base。owner 取 top - 1，窃取者取 base，两者指向同一格。nextLocalTask 里的 (nb = b + 1) == p 分支就是为这个情形准备的：只剩一个任务时，owner 必须从 top - 1 取，而不是从 base 取，否则两个线程会 CAS 同一格。即便这样，双方仍然可能同时 CAS，谁先成功谁拿到，失败的一方看到的是 null，走"重读"路径。
情形二：窃取者之间的争夺。 四个 worker 都空闲，都从各自的扫描循环里选中了同一个队列，都读到了同一个 base。四个线程同时 CAS 同一格，只有一个成功，另外三个的 compareAndSetReference 返回 false，poll() 继续下一轮。这不是错误，是设计预期：源码注释说得很直白，“如果一次窃取以这些方式失败，扫描者会换一个目标再试”。
情形三：owner 正在扩容。 growArray 从 top 往下逐格 getAndSetReference 搬到新数组，再 updateArray 换掉引用。窃取者此时可能读到旧数组的槽位；getReferenceAcquire 拿到的是旧数组里的任务，CAS 成功后照样执行——任务不会丢，但 base 的推进发生在旧数组上，新数组那边要靠 owner 重新对齐。这就是为什么 array 字段要用 getAndSetReference 换引用：换的那一刻是 full fence，之前的所有搬运对读者可见。
pb == b 这个停滞判断是本节最容易被略过的一行。pb 是上一轮的 base，b 是本轮的 base；两者相等意味着"我转了一圈，base 一点没动”。这时调 Thread.onSpinWait()，在 x86 上是 PAUSE，在 ARM 上是 YIELD，目的是降低自旋对内存子系统的压力。注意它只在 t == null（槽位暂时为空）时才判断，CAS 失败的分支不调——CAS 失败说明有人在动，不算停滞。
phase：一个字段兼任锁和版本号
phase 的注释只有三个词：“versioned active status”。它的位分配是：
bit 31 .. 17 : 版本计数（每次加 IDLE 就进位）
bit 16 : IDLE 位（1 = 空闲/可加锁，0 = 活跃/已加锁）
bit 15 .. 0 : 队列 id（worker 队列是奇数下标，提交队列是 EXTERNAL_ID_MASK 内的偶数）
锁的两个方法只有八行：
// JDK 25 源码摘录，已删减
final void unlockPhase() {
U.getAndAddInt(this, PHASE, IDLE);
}
final boolean tryLockPhase() { // seqlock acquire
int p;
return (((p = phase) & IDLE) != 0 &&
U.compareAndSetInt(this, PHASE, p, p + IDLE));
}
IDLE = 1 << 16。所以 p + IDLE 做的是"把第 16 位加 1"：如果这一位本来是 1，加完变 0 并向第 17 位进位——锁住了，同时版本号加一。unlockPhase 反过来加一次 IDLE：第 16 位从 0 变 1，版本号再加一。
这不是普通的互斥锁，是 seqlock 的写法：读者可以先读 phase，做完事再读一次，两次版本号相同就说明期间没人改过。JDK 25 里 phase 的读者主要是 deactivate、awaitWork、signalWork 这条唤醒链，它们靠版本号判断"这个 worker 的状态在我读的过程中有没有翻过"。
外部提交线程拿的是同一种锁。submissionQueue(int r, boolean rejectOnShutdown) 在探测到的队列上 tryLockPhase()，抢不到就 ThreadLocalRandom.advanceProbe(r) 换一个下标再试；push(task, pool, internal) 在 internal == false 时最后调 unlockPhase() 放锁：
// JDK 25 源码摘录，已删减
final void push(ForkJoinTask<?> task, ForkJoinPool pool, boolean internal) {
int s = top, b = base, m, cap, room; ForkJoinTask<?>[] a;
if ((a = array) != null && (cap = a.length) > 0 && // else disabled
task != null) {
int pk = task.noUserHelp() + 1; // prev slot offset
if ((room = (m = cap - 1) - (s - b)) >= 0) {
top = s + 1;
long pos = slotOffset(m & s);
if (!internal)
U.putReference(a, pos, task); // inside lock
else
U.getAndSetReference(a, pos, task); // fully fenced
if (room == 0) // resize
growArray(a, cap, s);
}
if (!internal)
unlockPhase();
if (room < 0)
throw new RejectedExecutionException("Queue capacity exceeded");
if ((room == 0 || U.getReferenceAcquire(a, slotOffset(m & (s - pk))) == null) &&
pool != null)
pool.signalWork(); // may have appeared empty
}
}
internal 这个参数是 owner 和外部提交者的分界线：owner 往自己的队列里压任务不需要锁（只有它能写 top），所以走 getAndSetReference 的 full fence 版本；外部提交者必须先拿到 phase 锁，用普通 putReference 写槽位。同一个方法，两种内存序，靠一个布尔参数切换——这是 JDK 25 相对旧版"两套队列类"的一个明显简化。
phase 的真实值可以用反射看到。本机跑一个小探针（--add-opens java.base/java.util.concurrent=ALL-UNNAMED），在一个 ForkJoinPool(2) 跑完一个空任务后打印：
queues.length=16
idx=5 phase=0x3c6f0005 idBits=5 IDLE=true
idx=8 phase=0x000139b8 idBits=14776 IDLE=true
seqlock idx=8 before=0x000139b8 locked=true during=0x000239b8 after=0x000339b8
读法：
idx=5是 worker 队列（奇数下标），低 16 位0x0005正好等于下标 5；高位的0x3c6f是版本计数。idx=8是提交队列（偶数下标），低 16 位0x39b8 = 14776，落在EXTERNAL_ID_MASK = 0x3ffe范围内，不是数组下标。- seqlock 那行最关键：
before = 0x000139b8，tryLockPhase()返回true后变成0x000239b8（加了0x10000，IDLE 位清零），unlockPhase()后是0x000339b8（再加一次，IDLE 位回到 1）。一次加锁+解锁，版本号正好前进 2。探针只读不改业务状态，但它动的是内部字段，只能用于学习和诊断。
nsteals：统计的是"跨线程执行的次数"
runWorker 的扫描循环里只有一行统计：
// JDK 25 源码摘录，已删减
boolean propagate;
int nb = q.base = b + 1, prevSrc = src;
w.nsteals = ++nsteals;
w.source = src = j; // volatile
rescan = true;
nsteals 是 worker 本地计数器，每次成功从别的队列取走一个任务就加一。它不区分这个任务是从提交队列拿的还是从别的 worker 队列偷的——只要不是自己队列里的，就算一次 steal。
getStealCount() 把它汇总起来：
// JDK 25 源码摘录，已删减
public long getStealCount() {
long count = stealCount;
WorkQueue[] qs; WorkQueue q;
if ((qs = queues) != null) {
for (int i = 1; i < qs.length; i += 2) {
if ((q = qs[i]) != null)
count += (long)q.nsteals & 0xffffffffL;
}
}
return count;
}
两个细节。第一，只遍历奇数下标——提交队列没有 nsteals，它们的任务是"被拿走"而不是"去偷"。第二，stealCount 是池级累加器：worker 退出时（deregisterWorker）把自己的 nsteals 加进去，所以池子跑过一段时间后，计数不会因为线程销毁而丢失。
javadoc 自己承认了精度问题：“The reported value underestimates the actual total number of steals when the pool is not quiescent.” 原因是 w.nsteals 是普通 int（不是 volatile），读者可能读到旧值；而且 worker 正在执行的窃取还没写回。把它当趋势指标可以，当精确账本不行。
争抢同一个 base 的失败率
JDK 没有暴露"CAS 失败次数"的计数器，nsteals 只记成功次数。要观察失败率，只能按 poll() 的形状复刻一个队列。下面这个模型把"读槽 → CAS 抢槽 → 推进 base"简化成"先 CAS 推进 base 序号 → 再读槽"：争抢的是同一个序号，失败率的量级可比，但不是逐行等价：
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerArray;
import java.util.concurrent.atomic.AtomicLong;
public final class StealContention {
static final class Ring {
final AtomicIntegerArray slots;
final int mask, top;
final AtomicInteger base = new AtomicInteger();
final AtomicLong attempts = new AtomicLong();
final AtomicLong failures = new AtomicLong();
final AtomicLong stolen = new AtomicLong();
Ring(int tasks) {
this.top = tasks;
this.mask = Integer.highestOneBit(tasks) * 2 - 1;
this.slots = new AtomicIntegerArray(mask + 1);
for (int i = 0; i < tasks; i++) slots.set(i & mask, i + 1);
}
int poll() {
for (;;) {
attempts.incrementAndGet();
int b = base.get();
if (b >= top) return -1;
if (base.compareAndSet(b, b + 1)) {
stolen.incrementAndGet();
return slots.getAndSet(b & mask, 0) - 1;
}
failures.incrementAndGet();
}
}
}
public static void main(String[] args) throws Exception {
int tasks = 200_000;
for (int thieves : new int[]{1, 2, 4, 8, 16}) {
Ring ring = new Ring(tasks);
AtomicLong consumed = new AtomicLong();
Thread[] ts = new Thread[thieves];
for (int i = 0; i < thieves; i++) {
ts[i] = Thread.ofPlatform().start(() -> {
while (consumed.get() < tasks) {
if (ring.poll() >= 0) consumed.incrementAndGet();
}
});
}
for (Thread t : ts) t.join();
System.out.printf("thieves=%2d attempts=%d failures=%d fail/attempt=%.3f%n",
thieves, ring.attempts.get(), ring.failures.get(),
ring.failures.get() / (double) ring.attempts.get());
}
}
}
本机输出：
thieves= 1 attempts=200000 failures=0 fail/attempt=0.000
thieves= 2 attempts=200001 failures=0 fail/attempt=0.000
thieves= 4 attempts=211543 failures=11540 fail/attempt=0.055
thieves= 8 attempts=430148 failures=230141 fail/attempt=0.535
thieves=16 attempts=375202 failures=142936 fail/attempt=0.381
（本机 Temurin 25.0.5+8，10 核，load average 82.6；单次运行、随调度波动、不作基准。）
1 到 8 个窃取者时失败率从 0 爬到 0.535：抢同一个序号的人越多，撞车越多，这一段方向稳定。16 个窃取者不能当规律读——同一份程序连跑三次，这一行给出 0.141、0.468、0.381，方向都不一致。10 核机器上开 16 个线程、load average 八十多，线程被调度得太散，撞在同一瞬间的概率随机波动；这不是"线程多了竞争反而小"。
真实池子上的对应指标是 getStealCount()。同样 64 个根任务、每个根 fork 256 个叶子（16384 个），只改并行度：
par= 2 forked=16384 steals= 3421 ms=248
par= 4 forked=16384 steals= 4393 ms=96
par= 8 forked=16384 steals= 7281 ms=48
par=16 forked=16384 steals= 6841 ms=47
并行度从 2 加到 8，窃取次数从 3421 涨到 7281——每多一个线程，就多一个"去别人队列里找活"的人；但耗时从 248ms 掉到 48ms。这正是工作窃取的账本：窃取次数不是越低越好，它买到的是并行度。12.1 里 surplusLimit=0 把窃取压到 168 次、耗时 47ms，那是另一个方向的极端——叶子被就地串行执行，窃取少了，并行也没了。16 个线程时窃取次数反而降到 6841，同样是本机高负载下的调度偏差。
在 libraryd-conc 里落地
报表预聚合的提交端有一个不明显的竞争点：如果 32 个 HTTP 请求线程各自调用 pool.submit(...)，它们会一起去抢提交队列的 phase 锁。submissionQueue 的处理方式是"抢不到就换下标"，所以请求线程越多，创建的提交队列越多，queues 数组越大，扫描成本跟着涨。
一个可控的做法是：固定一组提交线程，让它们各自预热 ThreadLocalRandom 探针，把提交压力摊到固定几个队列上：
package com.plume.conc.core;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.atomic.LongAdder;
/** 报表预聚合的提交端：固定提交线程数，避免每个请求线程各建一个提交队列。 */
public final class ReportAggregator implements AutoCloseable {
private final ForkJoinPool pool;
private final LongAdder submitted = new LongAdder();
public ReportAggregator(int parallelism) {
this.pool = new ForkJoinPool(parallelism);
}
/** 预聚合：调用方保证 rows 已在内存，I/O 在提交前完成。 */
public ForkJoinTask<Map<String, int[]>> submit(List<InventorySnapshot.Row> rows, int threshold) {
submitted.increment();
return pool.submit(() -> aggregate(rows, threshold));
}
private static Map<String, int[]> aggregate(List<InventorySnapshot.Row> rows, int threshold) {
return InventorySnapshot.snapshot(rows, threshold, 2);
}
/** 观察指标：窃取次数与提交次数一起看，才知道池子是饿还是挤。 */
public String stats() {
return "submitted=" + submitted.sum()
+ " steals=" + pool.getStealCount()
+ " queued=" + pool.getQueuedTaskCount()
+ " parallelism=" + pool.getParallelism();
}
@Override
public void close() {
pool.shutdown();
}
}
三点取舍：
第一，用专用池而不是 commonPool。commonPool 的并行度是"核数 - 1"，并且被全 JVM 的 parallelStream 共享；报表预聚合一旦把 9 个 worker 占满，同一个 JVM 里任何 parallelStream 都会排队（12.3 有数据）。专用池把这种耦合切断。
第二，stats() 里把 steals 和 queued 放在一起。只看 steals 会误判：steals 高可能是任务粒度太细，也可能是并行度不够；queued 长期大于 0 才是真的堆积。
第三，submit 返回 ForkJoinTask 而不是 Future，调用方可以 join()，也可以拿到原始任务做 tryUnfork。代价是调用方必须知道这是 ForkJoinTask，接口不如 ExecutorService 通用。
这个设计的代价
缓存行是稀缺资源，但 @Contended 不是万能的。 top、phase、stackPred、source、nsteals、parking 六个字段都标了 @Contended("w")。按注解的语义，同一个 group tag 的字段只保证整组与其它组隔离，组内字段之间并不保证互不共享缓存行；只有不带 tag 的 @Contended 才是每个字段独立成组。所以这六个字段是"抱团避开 base 和 array"，不是"六条独立缓存行"。隔离本身要付填充的代价，并行度 16 的池子光队列对象的空间开销就不可忽略。
窃取没有公平性保证。 pollScan 从随机下标开始，runWorker 用 xorshift 生成步长，两者都不保证"每个队列被扫到的概率相等"。源码注释只说"approximately the same probability"。如果你需要严格公平（比如按租户隔离计算资源），工作窃取给不了。
getStealCount() 不是实时指标。 它遍历 queues 数组读 nsteals，本身要 O(队列数) 的时间；池子不安静时还会低估。把它接进监控要按分钟级采样，不要每秒拉一次。
什么时候不要用工作窃取。 任务之间需要共享可变状态、或者必须按提交顺序执行时，ForkJoinPool 的双端队列会让执行顺序不可预测；任务里要做阻塞 I/O 时，worker 不会自动补偿，应该换成虚拟线程或专门的 I/O 池；任务粒度小到纳秒级时，fork/join 的簿记成本远大于计算本身。
常见错误与边界
以为 tryLockPhase() 是普通锁，可以长期持有。 它是 seqlock 的写端，持锁期间 phase 的 IDLE 位为 0，唤醒链上的读者会认为这个队列"活跃"。外部提交者持锁时间必须是一个 push 的长度，不能在里面做 I/O。
用 nsteals 判断"负载均衡好不好"。 它只记成功次数，不记失败次数，也不记任务大小。两个池子 steals 相同，可能一个是 1000 个小任务，一个是 10 个大任务。
把 phase 的低 16 位当数组下标。 只有 worker 队列的 phase & SMASK 等于下标；提交队列的低 16 位是 EXTERNAL_ID_MASK 范围内的 id，和 queues 下标没有一一对应关系。
假设 base 单调且与 top 的差就是精确长度。 base 是 volatile 写，读到旧值很正常；queueSize() 的注释写着"approximate"，源码里还有 Math.max(top - base, 0) 兜底负数。
小结
WorkQueue 的两端不是对称的：owner 从 top 弹，只写自己的 top，不需要锁；窃取者从 base 抢，用一次 compareAndSetReference 决定归属，失败就重读。两端在"只剩一个任务"时指向同一格，靠 (nb = b + 1) == p 这个分支错开。phase 是这套协作的中枢：低 16 位标识队列，第 16 位是 IDLE，其余位在每次加锁/解锁时各进一位，所以它既是提交队列的自旋锁，也是唤醒链的版本号。nsteals 只在成功时加一，getStealCount() 把奇数下标队列的计数和池级的 stealCount 相加，池子不安静时会低估。
本机的复刻实验显示，窃取者从 1 涨到 8，CAS 失败率从 0 涨到 0.544；真实池子上并行度从 2 涨到 8，窃取次数从 3421 涨到 7281，耗时从 248ms 降到 48ms。窃取次数是买并行度的价钱，不是越低越好。
下一节离开调度器，看任务是怎么被拆出来的：Spliterator 的三个契约、AbstractTask 的循环式拆分，以及为什么 parallelStream 里做 I/O 会把整个 commonPool 拖住。
阅读导航：上一节 12.1 工作窃取队列与任务扫描 ｜ 下一节 12.3 并行流的拆分策略
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。