您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

Java 併發進階：Fork/Join 框架原理與實戰

編輯：Java 相關知識

前言

Fork/Join 框架是 Java 7 引入的並行任務框架，核心思想是「分治」：將大任務遞歸拆分為子任務並行執行，再合併結果。它採用工作竊取算法最大化 CPU 利用率，是並行排序、大數據聚合、遞歸計算的利器。本文涵蓋核心概念、工作竊取機制、API 使用、異常處理及源碼級實現原理。

一、核心概念：Fork 與 Join

Fork：將大任務拆分為多個獨立子任務，提交到線程池並行執行。
Join：等待所有子任務完成，匯總結果得到最終答案。

Fork/Join 運行流程圖

二、工作竊取算法

工作竊取流程圖

機制

每個工作線程維護一個雙端隊列。
線程從自己隊列頭部取任務執行（pollFirst）。
自身隊列空時，隨機選擇其他線程隊列尾部竊取任務（pollLast）。

優缺點

優點	缺點
充分利用 CPU 核心，減少全局隊列競爭	雙端隊列單元素時仍有競爭
遞歸任務自然負載均衡	額外系統資源（多線程、多隊列）
適合計算密集型遞歸任務	不適合 I/O 密集、長阻塞任務

三、核心類體系

類/介面	角色	關鍵方法
`ForkJoinTask<V>`	任務基類	`fork()`、`join()`、`invoke()`、`compute()`
`RecursiveAction`	無返回結果任務	繼承實現 `compute()`
`RecursiveTask<V>`	有返回結果任務	繼承實現 `compute()` 返回 V
`ForkJoinPool`	執行器	`invoke(task)`、`submit(task)`、`execute(task)`

四、實戰範例：計算 1+2+3+4

任務定義（閾值=2）

static class SumTask extends RecursiveTask {
    private final int start, end;
    SumTask(int start, int end) { this.start = start; this.end = end; }

    @Override protected Integer compute() {
        if (end - start <= 2) { // 足夠小，直接計算
            int sum = 0;
            for (int i = start; i <= end; i++) sum += i;
            return sum;
        }
        int mid = (start + end) >>> 1;
        SumTask left = new SumTask(start, mid);
        SumTask right = new SumTask(mid + 1, end);
        left.fork();              // 異步提交左半
        int rightResult = right.compute(); // 當前線程執行右半
        int leftResult = left.join();      // 等待左半結果
        return leftResult + rightResult;
    }
}

提交執行

ForkJoinPool pool = new ForkJoinPool();
int result = pool.invoke(new SumTask(1, 4)); // 返回 10
pool.shutdown();

五、異常處理機制

子任務異常不會自動傳播到主線程，需主動檢查：

ForkJoinTask task = pool.submit(new SumTask(1, 100));
task.join(); // 等待完成
if (task.isCompletedAbnormally()) {
    Throwable ex = task.getException(); // 獲取異常
    ex.printStackTrace();
}

getException() 返回 Throwable：取消返回 CancellationException，異常返回實際異常，未完成/無異常返回 null。

六、源碼級實現原理

1. ForkJoinPool 結構

ForkJoinTask[] 任務陣列（工作隊列存儲）
ForkJoinWorkerThread[] 工作線程陣列

2. fork() 異步提交流程

public final ForkJoinTask fork() {
    ((ForkJoinWorkerThread) Thread.currentThread()).pushTask(this);
    return this;
}

// ForkJoinWorkerThread.pushTask
final void pushTask(ForkJoinTask t) {
    ForkJoinTask[] q = queue;
    int s = queueTop, m = q.length - 1;
    long u = (((s & m) << ASHIFT) + ABASE); // 計算陣列索引
    UNSAFE.putOrderedObject(q, u, t);      // CAS 入隊
    queueTop = s + 1;
    if ((s - queueBase) <= 2) pool.signalWork(); // 隊列較短，喚醒/創建線程
    else if (s == m) growQueue(); // 擴容
}

3. join() 同步等待流程

public final V join() {
    if (doJoin() != NORMAL) return reportResult();
    return getRawResult();
}

private int doJoin() {
    Thread t = Thread.currentThread();
    if (t instanceof ForkJoinWorkerThread) {
        ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
        if (status < 0) return status; // 已完成
        if (w.unpushTask(this)) {      // 嘗試從自己隊列彈出（已執行過？）
            try { completed = exec(); } // 執行 compute()
            catch (Throwable rex) { return setExceptionalCompletion(rex); }
            if (completed) return setCompletion(NORMAL);
        }
        return w.joinTask(this); // 幫助執行其他任務或等待
    }
    return externalAwaitDone(); // 外部線程（非 ForkJoinWorkerThread）純等待
}

關鍵點：join() 不會單純阻塞，工作線程會嘗試幫助執行其他任務（helpQuiesce），避免線程空閒，這是 Fork/Join 高效的核心。

4. 任務狀態機

狀態常量	含義
NORMAL (0)	正常完成
CANCELLED (1)	被取消
SIGNAL (2)	信號狀態（內部使用）
EXCEPTIONAL (3)	拋出異常

七、最佳實踐與避坑

閾值設定：太大 → 並行度不足；太小 → 任務創建開銷大。經驗值：任務執行時間 1-10ms。
避免阻塞操作：compute() 中不要做 I/O、鎖等待、睡眠，否則會拖垮整個池。

使用 invokeAll 簡化：

invokeAll(left, right); // 等同於 left.fork(); right.compute(); left.join();

共用池：JDK 8+ 提供 ForkJoinPool.commonPool()，避免重複創建線程池。
結果合併無副作用：join() 後合併結果應為純函數操作，避免共享狀態競爭。

八、總結

Fork/Join 以工作竊取雙端隊列實現任務動態負載均衡，以幫助執行機制消除 join() 等待時的線程空閒。它將「分治遞歸」映射為「任務拆分-並行執行-結果合併」的並行模式，極大降低了並行遞歸程式的編寫門檻。掌握 RecursiveTask.compute() 拆分邏輯、fork/join 語義、doJoin() 幫助執行流程，能讓你在面對大規模計算任務（並行排序、MapReduce 風格聚合、樹狀結構遍歷）時，寫出既簡潔又高效的並行代碼。