全栈攻城狮

Appearance

ForkJoinPool 工作窃取线程池源码精读 ​

概述 ​

ForkJoinPool 是专为分治（Divide & Conquer）设计的线程池：任务可递归拆分（fork）、可等待子任务结果（join），每个工作线程持有一条双端队列（WorkQueue），自己任务用完还能从别的线程队尾"偷"任务——这就是工作窃取（Work Stealing）。parallelStream、CompletableFuture 的 commonPool 都建立在它之上。本文基于 OpenJDK 21 源码拆解核心机制。

一、WorkQueue[] 双端队列与窃取模型 ​

1.1 核心结构 ​

java
// java.util.concurrent.ForkJoinPool
volatile WorkQueue[] workQueues;      // 所有队列：工作线程队列 + 外部提交队列

// 内部类 WorkQueue
static final class WorkQueue {
    volatile int base;                // 队头索引（窃取方读取，从头拿）
    int top;                          // 队尾索引（所有者写入，从尾放/拿）
    ForkJoinTask<?>[] array;          // 环形数组存放任务
    final ForkJoinPool pool;          // 所属池
    final ForkJoinWorkerThread owner; // 队列所有者（null = 外部提交队列）
}

每个工作线程独占一条 WorkQueue，操作分区：

同一线程（owner）：
  push(task)   → array[top++]  写入队尾（新任务从尾进）
  pop()        → array[--top]  从队尾取（LIFO，最近的任务优先）

其他线程（窃取者）：
  poll()       → array[base]   base++  从队头取（FIFO，最老的任务）

LIFO + FIFO 的混用：所有者从队尾取（缓存友好、局部性好的大任务先处理）；窃取者从队头拿（最老、最可能被其他线程需要的小任务）——两条策略都是为了减少竞争、提高吞吐。

1.2 为什么叫"工作窃取" ​

单队列模型（如 ThreadPoolExecutor）：
  所有线程抢同一个队列 → 锁竞争激烈
  某个线程任务做完就空闲，即使其他队列排队

工作窃取模型（ForkJoinPool）：
  每个线程有自己的队列 → 无锁 push/pop（owner 独占操作）
  线程空闲 → 扫描别的队列 poll() 偷任务（CAS 更新 base）
  动态平衡：忙线程帮闲线程消化任务

二、任务提交：submit / execute ​

2.1 submit(ForkJoinTask) 的 external Push ​

java
public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
    if (task == null) throw new NullPointerException();
    externalPush(task);        // 外部提交：放入随机槽位的提交队列
    return task;
}

final void externalPush(ForkJoinTask<?> task) {
    WorkQueue[] ws; WorkQueue q; int m;
    int r = ThreadLocalRandom.getProbe();      // 线程探针（伪随机）
    // 队列数组未初始化 → 初始化（懒加载）
    if ((ws = workQueues) == null || (m = ws.length - 1) < 0)
        ws = initQueues();                     // 第一次提交才建队列数组
    // 命中一个提交队列槽位
    else if ((q = ws[m & r & SQMASK]) == null) {
        q = new WorkQueue(this, null);         // owner == null → 提交队列
        q.array = new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY];
        // CAS 把提交队列放到槽位
        if (U.compareAndSetReference(ws, (long)(m & r & SQMASK) * ASIZE + ABASE, null, q))
            q.array = new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY];
        else
            q = null;
    }
    if (q != null) {
        q.lockedPush(task);                    // 提交队列用锁保护（多线程共享）
        if (q.full) signalWork();              // 满则唤醒工作线程
        return;
    }
    // 槽位竞争失败 → 走 externalSubmit（扩容/重试）
    externalSubmit(task);
}

void signalWork() {
    // 工作线程不足 → 尝试新增一个工作线程
    // 没有空闲线程 → 把空闲线程从队列中唤醒
}

外部提交与工作线程内部提交的区别：外部提交队列 owner == null，多线程共享，入队需要 lockedPush（加锁 CAS）；工作线程自己的队列 owner 独占，push 无需锁。

2.2 execute(Runnable) 与 commonPool ​

java
public void execute(Runnable task) {
    if (task == null) throw new NullPointerException();
    ForkJoinTask<?> job;
    if (task instanceof ForkJoinTask<?> fjt)    // ForkJoinTask 直接提交
        job = fjt;
    else
        job = new ForkJoinTask.RunnableExecuteAction(task);  // 包装成任务
    externalPush(job);
}

// 全局共享池：CompletableFuture 默认异步执行器
public static ForkJoinPool commonPool() { ... }

ForkJoinPool.commonPool() 是全局共享的工作窃取池（并行度 = 核数 - 1，守护线程），parallelStream 与 CompletableFuture 默认都走它。

三、fork / join 分治流程 ​

3.1 fork() 的 push ​

java
// ForkJoinTask.fork
public final ForkJoinTask<V> fork() {
    Thread t;
    // 当前线程是池内工作线程 → push 到自己队列（无锁，owner 独占）
    if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
        ((ForkJoinWorkerThread) t).workQueue.push(this);
    else
        ForkJoinPool.common.externalPush(this);   // 外部线程 → 提交队列
    return this;
}

// WorkQueue.push：owner 独占操作，无锁
final void push(ForkJoinTask<?> task) {
    ForkJoinTask<?>[] a; int s = top;
    if ((a = array) != null) {              // 队列已初始化
        int m = a.length - 1;
        if ((s & m) == ((s = top + 1) & m)) // 队满（top 撞上 base）
            growArray();                    // 扩容：数组翻倍并迁移任务
        a[s & m] = task;                    // 写入队尾
        top = s;                            // top 递增（volatile 写入）
    }
}

fork() 只是把任务推入当前线程的队列尾部，不立即执行——子任务等待父任务的 join() 或队列扫描时被消费。

3.2 join() 的 doJoin ​

java
public final V join() {
    int s;
    if ((s = status) < 0)                    // ① 已完成（status 非负 = 未完成）
        return getRawResult();               //    直接返回结果
    if (Thread.currentThread() instanceof ForkJoinWorkerThread wt)
        return wt.pool.awaitJoin(wt.workQueue, this, 0L);  // ② 池内线程等待
    else
        return externalAwaitDone();          // ③ 外部线程阻塞等待
}

池内线程的 awaitJoin 不是干等，而是边等边干活：

awaitJoin 的辅助策略：
  1. 循环检查任务状态（status < 0 → 完成）
  2. 从自己队列中 pop 任务执行（doExec）
  3. 从其他队列 scan 窃取任务执行
  4. 都空 → 阻塞（unpark 后被唤醒继续）
java
// doExec：真正执行任务主体
final int doExec() {
    int s; boolean completed;
    if ((s = status) >= 0) {
        try {
            completed = exec();              // 调用 compute()（RecursiveTask/RecursiveAction）
        } catch (Throwable rex) {
            s = exceptionalCompletion(rex);  // 异常 → 记录到 status
            completed = false;
        }
        ...
    }
    return s;
}

join 的核心思想：等待期间不空闲，把当前线程变成"执行者"去消费队列里的任务（包括别人的），这就是高吞吐的来源。

3.3 compute() 的分治递归 ​

java
// 典型用法：RecursiveTask 求和
public class SumTask extends RecursiveTask<Long> {
    static final int THRESHOLD = 10_000;       // 阈值：足够小就直算
    final long[] array; final int lo, hi;

    protected Long compute() {
        if (hi - lo <= THRESHOLD) {            // ① 小任务 → 直接计算
            long sum = 0;
            for (int i = lo; i < hi; i++) sum += array[i];
            return sum;
        }
        int mid = (lo + hi) >>> 1;
        SumTask left  = new SumTask(array, lo, mid);
        SumTask right = new SumTask(array, mid, hi);
        left.fork();                           // ② 左半部分 fork（入队）
        long rightResult = right.compute();    // ③ 右半部分当前线程直接算
        return left.join() + rightResult;      // ④ 等待左半结果
    }
}
分治执行过程（4 核机器，数组 80000 个元素）：
  根任务 compute → fork 左 → 算右 → join 左
  左任务被某个空闲线程窃取 → 再 fork/join 递归
  每个线程都在切自己的子问题 → 动态负载均衡
  THRESHOLD 控制叶子任务大小 → 避免过度拆分（任务开销 > 计算收益）

四、WorkQueue.scan() 的工作窃取 ​

空闲工作线程执行 scan() 寻找可窃取任务：

java
// ForkJoinWorkerThread.run 主循环 → scan()
private int scan(WorkQueue w, WorkQueue[] ws, int n, int r) {
    WorkQueue q; ForkJoinTask<?>[] a; int b, k;
    int m = n - 1;                             // 数组掩码
    // 从随机起始槽位开始，环形扫描所有队列
    for (int j = (n << 2) | r, i = 0; i < n; ++i) {
        if ((q = ws[k = (j + i) & m]) != null) {   // 命中一个队列
            if ((a = q.array) != null && (b = q.base) != q.top) {  // 队列非空
                if (U.compareAndSetInt(q, QBASE, b, b + 1)) {      // CAS 抢 base
                    ForkJoinTask<?> t = a[b & (a.length - 1)];
                    if (q.top - b <= 1) ...                       // 只剩一个任务
                    return (t != null) ? t : 0;                    // 偷到任务
                }
            }
        }
    }
    return 0;   // 没偷到 → 尝试休眠
}
scan 关键点：
  起始槽位由线程探针 r 决定（伪随机）→ 多个窃取者分散，避免抢同一队列
  CAS base 抢任务 → 多窃取者并发安全
  偷到任务后调 doExec 执行（执行的是"别人的子任务"）
  所有队列都空 → 线程进入休眠（等 signalWork 唤醒）

五、ForkJoinWorkerThread 注册与生命周期 ​

java
// java.util.concurrent.ForkJoinWorkerThread
public class ForkJoinWorkerThread extends Thread {
    final ForkJoinPool pool;      // 所属池
    final ForkJoinPool.WorkQueue workQueue;   // 自己的队列
}

创建与注册流程：

java
// ForkJoinPool 增加线程（addWorker 类似机制）
private boolean createWorker() {
    ForkJoinWorkerThreadFactory fac = factory;
    ...
    ForkJoinWorkerThread wt = fac.newThread(this);   // 工厂创建线程（可自定义线程名等）
    // 线程启动后执行 registerWorker 注册
}

// ForkJoinWorkerThread.run
public void run() {
    if (workQueue.array == null) {     // 首次运行 → 注册
        WorkQueue w;
        try {
            w = pool.registerWorker(this);   // ① 注册：分配队列槽位
        } catch (Throwable ex) { ... }
        ...
    }
    try {
        pool.runWorker(w);             // ② 进入工作循环（scan 任务 + 执行）
    } catch (Throwable ex) { ... }
    ...
}

// ForkJoinPool.registerWorker
final WorkQueue registerWorker(ForkJoinWorkerThread wt) {
    // 从 workQueues 数组中找一个空槽位
    // workQueues[slot] = workQueue；并把 owner 指向该线程
}
线程生命周期：
  创建：工厂 newThread → run() 启动
  注册：registerWorker 占槽位（workQueues[slot] = wq）
  工作：runWorker 主循环 → scan() 偷任务 / 执行自己队列任务
  休眠：队列全空 → 阻塞等待（池按需唤醒）
  终止：空闲超时（keepAlive）或被池关闭 → deregisterWorker 清理槽位

六、ManagedBlocker 补偿线程 ​

池内线程执行 compute() 时如果调用了阻塞 API（如 BlockingQueue.take、Lock.lock），会占着线程不干活——若所有线程都阻塞，池就"饿死"了。ManagedBlocker 让池临时补偿一个线程替代阻塞者：

java
// ForkJoinPool.managedBlock
public static void managedBlock(ManagedBlocker blocker) throws InterruptedException {
    ForkJoinPool p = null; Thread t = Thread.currentThread();
    if (t instanceof ForkJoinWorkerThread wt) p = wt.pool;
    if (p == null || !p.tryCompensate(p.workQueues, wt))   // 补偿失败/非池内线程
        return;                                            // → 普通阻塞（无补偿）
    try {
        do { } while (!blocker.isReleasable() && !blocker.block());   // 阻塞等待
    } finally {
        U.getAndAddInt(p, CTL, COMPENSATION);   // 归还补偿名额
    }
}

// 使用示例：池内阻塞安全写法
ForkJoinPool.managedBlock(new ManagedBlocker() {
    public boolean block() throws InterruptedException {
        queue.take(); return true;
    }
    public boolean isReleasable() { return false; }
});
补偿机制：
  池内线程要阻塞 → tryCompensate：若池未满则启动一个临时线程顶替
  阻塞者恢复后 → 归还补偿名额（临时线程空闲超时后被回收）
  效果：池的"活跃工作线程数"不因阻塞减少 → 避免死锁与饥饿
  对比：ThreadPoolExecutor 阻塞会占线程（无补偿），只能靠 maxPoolSize 兜底

七、CountedCompleter 完成回调 ​

CountedCompleter 是另一种任务类型：无返回值，用计数器跟踪子任务完成度，全部完成后触发一次 onCompletion() 回调。适合树形遍历、聚合场景：

java
// java.util.concurrent.CountedCompleter
public abstract class CountedCompleter<T> extends ForkJoinTask<T> {
    final CountedCompleter<?> completer;   // 父任务（完成后通知它）
    volatile int pending;                  // 待完成计数（子任务数）

    protected void onCompletion(CountedCompleter<?> caller) { }   // 回调钩子（空实现）
}

// 核心：tryComplete 递减计数
public final void tryComplete() {
    CountedCompleter<?> a = this, s = a;
    for (;;) {
        int p = s.pending;
        if (p == 0) {                        // 计数为 0 → 本节点完成
            s.onCompletion(s);               // ① 回调
            if ((a = s.completer) == null) { // ② 无父 → 整个树完成
                s.quietlyComplete();
                return;
            }
            s = a;                           // ③ 沿父链继续
        } else if (U.compareAndSetInt(s, PENDING, p, p - 1)) {
            return;                          // ④ 还有子任务 → 计数减 1 等待
        }
    }
}
java
// 使用示例：数组求和（CountedCompleter 版本）
class SumCC extends CountedCompleter<Long> {
    final long[] array; final int lo, hi;
    long sum;
    SumCC(SumCC parent, long[] array, int lo, int hi) {
        super(parent); this.array = array; this.lo = lo; this.hi = hi;
    }
    public void compute() {
        if (hi - lo <= THRESHOLD) {
            for (int i = lo; i < hi; i++) sum += array[i];
            tryComplete();                    // 叶子：标记完成
        } else {
            int mid = (lo + hi) >>> 1;
            new SumCC(this, array, lo, mid).fork();   // 子任务挂到父的 pending
            new SumCC(this, array, mid, hi).fork();
        }
    }
    public void onCompletion(CountedCompleter<?> caller) {
        // 所有子任务完成 → 汇总（把子任务的和加到自己）
        if (caller != this) sum = ((SumCC) caller).sum + sum;
    }
}
CountedCompleter 要点：
  pending：子任务计数，fork 子任务时父的 pending 递增
  tryComplete：pending 减到 0 → onCompletion 回调 → 沿 completer 链上溯
  onCompletion：汇总/清理逻辑（子任务结果在此合并）
  对比 RecursiveTask.join：不需要阻塞等待，靠计数驱动，适合异步回调式汇总

八、实现要点 ​

ForkJoinPool 核心：
  WorkQueue[] 双端队列：owner 从尾 push/pop（LIFO），窃取者从尾 poll（FIFO）
  工作窃取：空闲线程 scan 随机起始槽位环形扫队列，CAS base 抢任务
  fork：入队不执行；join：doJoin 等待期间边等边干活
  分治：THRESHOLD 阈值控制叶子任务大小，compute 递归拆分
  提交：externalPush 进提交队列（owner==null，lockedPush），signalWork 唤醒
  线程注册：registerWorker 占槽位，runWorker 主循环 scan
  ManagedBlocker：阻塞时补偿临时线程，避免池饥饿
  CountedCompleter：pending 计数 + onCompletion 回调，无阻塞汇总

常见陷阱：
  递归拆分过深 → 任务对象爆炸（合理设置 THRESHOLD）
  池内线程阻塞（锁/队列 take）→ 用 managedBlock 补偿
  parallelStream 共用 commonPool → 避免耗时任务占满并行度
  join 顺序错误 → 先 fork 左再 join 左（先 fork 的先 join 避免栈溢出）
  countDownLatch 等在池内 → 死锁风险（补偿机制不覆盖所有情况）