CountedCompleter通過pending count機(jī)制實(shí)現(xiàn)自定義任務(wù)完成條件，適用于分治算法；其核心是手動(dòng)管理等待計(jì)數(shù)，調(diào)用tryComplete()觸發(fā)onCompletion回調(diào)合并結(jié)果，比RecursiveTask更靈活但復(fù)雜。

在Java中，CountedCompleter 是 ForkJoinTask 的一個(gè)抽象子類，適用于需要自定義任務(wù)完成條件的場景，特別適合實(shí)現(xiàn)分治并行算法。它比普通的 RecursiveAction 或 RecursiveTask 更靈活，因?yàn)槟憧梢钥刂迫蝿?wù)何時(shí)才算“完成”——不是執(zhí)行完 compute 就結(jié)束，而是通過手動(dòng)調(diào)用 tryComplete() 或 complete(...) 來觸發(fā)完成邏輯。

理解 CountedCompleter 的核心機(jī)制

CountedCompleter 的關(guān)鍵在于“等待計(jì)數(shù)器”（pending count）。每個(gè)任務(wù)都有一個(gè)內(nèi)部計(jì)數(shù)器，表示它等待多少個(gè)子任務(wù)完成。當(dāng)一個(gè)任務(wù)被創(chuàng)建時(shí)，你可以設(shè)置它的 pending count。每當(dāng)一個(gè)子任務(wù)完成，計(jì)數(shù)器減一。當(dāng)計(jì)數(shù)器歸零時(shí)，系統(tǒng)自動(dòng)調(diào)用任務(wù)的 onCompletion(CountedCompleter) 方法，并向上游傳播完成狀態(tài)。

常用方法：

• compute()：主邏輯入口，通常用于拆分任務(wù)或執(zhí)行計(jì)算。
• tryComplete()：將當(dāng)前任務(wù)的 pending count 減一；如果歸零，則觸發(fā) onCompletion 并通知父任務(wù)。
• onCompletion(CountedCompleter)：任務(wù)完成時(shí)的回調(diào)，常用于合并結(jié)果。
• addToPendingCount(int delta)：增加等待的子任務(wù)數(shù)量。
• quietlyComplete()：強(qiáng)制完成任務(wù)而不觸發(fā)異常傳播。

使用 CountedCompleter 實(shí)現(xiàn)分治求和

下面以數(shù)組求和為例，展示如何用 CountedCompleter 實(shí)現(xiàn)并行分治：

立即學(xué)習(xí)“Java免費(fèi)學(xué)習(xí)筆記（深入）”；

import java.util.concurrent.*;
<p>public class SumTask extends CountedCompleter<Long> {
private final long[] array;
private final int lo, hi;
private Long result;</p><pre class='brush:java;toolbar:false;'>public SumTask(CountedCompleter<?> parent, long[] array, int lo, int hi) {
    super(parent);
    this.array = array;
    this.lo = lo;
    this.hi = hi;
}

@Override
public void compute() {
    if (hi - lo <= 1000) {
        // 小數(shù)據(jù)量直接計(jì)算
        long sum = 0;
        for (int i = lo; i < hi; i++) {
            sum += array[i];
        }
        result = sum;
        // 告知自己已完成，減少父任務(wù)的等待計(jì)數(shù)
        tryComplete();
    } else {
        // 拆分為兩個(gè)子任務(wù)
        int mid = (lo + hi) / 2;
        // 增加兩個(gè)待完成的子任務(wù)
        addToPendingCount(1); // 右半部分
        addToPendingCount(1); // 左半部分

        // 提交右半部分（作為新任務(wù)）
        new SumTask(this, array, mid, hi).fork();
        // 當(dāng)前任務(wù)處理左半部分
        new SumTask(this, array, lo, mid).fork(); // 也可復(fù)用當(dāng)前任務(wù)
    }
}

@Override
public void onCompletion(CountedCompleter<?> caller) {
    // 合并子任務(wù)結(jié)果
    SumTask left = (SumTask) getFirstChild();
    SumTask right = (SumTask) left.getNextSibling();

    long leftResult = left == null ? 0 : left.result;
    long rightResult = right == null ? 0 : right.result;

    result = leftResult + rightResult;
}

public Long getRawResult() {
    return result;
}

// 示例運(yùn)行
public static void main(String[] args) {
    long[] data = new long[100_000];
    Arrays.fill(data, 1);

    ForkJoinPool pool = new ForkJoinPool();
    SumTask task = new SumTask(null, data, 0, data.length);
    pool.invoke(task);

    System.out.println("Sum: " + task.getResult());
    pool.shutdown();
}

}

如知AI筆記

如知筆記——支持markdown的在線筆記，支持ai智能寫作、AI搜索，支持DeepseekR1滿血大模型

27

查看詳情

關(guān)鍵設(shè)計(jì)要點(diǎn)

使用 CountedCompleter 時(shí)需注意以下幾點(diǎn)：

• 顯式管理 pending count：每次創(chuàng)建子任務(wù)必須調(diào)用 addToPendingCount(1)，否則父任務(wù)不會(huì)等待它。
• 正確調(diào)用 tryComplete：葉子任務(wù)計(jì)算完成后應(yīng)調(diào)用 tryComplete() 觸發(fā)完成流程。
• 結(jié)果合并放在 onCompletion：不要在 compute 中直接合并，應(yīng)在 onCompletion 中處理，確保所有子任務(wù)已完成。
• 父子關(guān)系由構(gòu)造函數(shù)傳遞：子任務(wù)的 parent 應(yīng)傳入當(dāng)前任務(wù)，形成任務(wù)樹結(jié)構(gòu)。
• 避免內(nèi)存泄漏：長時(shí)間運(yùn)行的任務(wù)應(yīng)避免持有大對象引用，及時(shí)釋放中間結(jié)果。

適用場景與優(yōu)勢

CountedCompleter 特別適合以下情況：

• 任務(wù)依賴多個(gè)子任務(wù)完成才可繼續(xù)（如樹形結(jié)構(gòu)遍歷）。
• 需要?jiǎng)討B(tài)生成子任務(wù)數(shù)量（比如遞歸圖遍歷）。
• 希望更精細(xì)控制任務(wù)完成邏輯，而非簡單的 fork/join 模型。

相比 RecursiveTask，它提供了更大的靈活性，但代價(jià)是代碼復(fù)雜度上升。對于標(biāo)準(zhǔn)的分治問題（如歸并排序、矩陣乘法），若子任務(wù)數(shù)固定為2，RecursiveTask 更簡潔；但若子任務(wù)數(shù)可變或需要延遲完成判斷，CountedCompleter 是更好選擇。

基本上就這些。

以上就是如何在Java中使用CountedCompleter實(shí)現(xiàn)分治并行的詳細(xì)內(nèi)容，更多請關(guān)注php中文網(wǎng)其它相關(guān)文章！