Java中如何使用CountedCompleter处理ForkJoin任务-java教程-PHP中文网

Java中如何使用CountedCompleter处理ForkJoin任务

P粉602998670

发布： 2025-11-03 22:25:32

原创

947人浏览过

CountedCompleter是ForkJoinTask的抽象子类，通过维护待完成计数器控制任务完成时机，适用于需协调多个子任务的场景。其核心在于pending count机制：调用addToPendingCount增加计数，tryComplete减少计数并传播完成状态，当计数归零时触发onCompletion方法执行汇总操作。示例中并行求和任务在拆分时增加计数，子任务完成后自动递减，最终在onCompletion中合并结果。该机制灵活支持树形处理、图遍历等复杂依赖结构，但需确保计数管理正确，避免提前完成或死锁。

java中如何使用countedcompleter处理forkjoin任务

在Java中，CountedCompleter 是 ForkJoinPool 提供的一个特殊类型的 ForkJoinTask，它允许你自定义任务的完成条件，而不仅仅依赖于任务自身的执行结束。这在需要多个子任务协同完成、或某个任务需等待其他任务完成后才触发回调时非常有用。

什么是CountedCompleter？

CountedCompleter 是 ForkJoinTask 的抽象子类，它引入了一个“待完成计数器”（pending count）。每当一个子任务启动或某个依赖任务被注册，你可以手动增加这个计数器。只有当计数器归零时，任务的 onCompletion 方法才会被调用，表示该任务真正完成。

与普通的 RecursiveAction 或 RecursiveTask 相比，CountedCompleter 更灵活，适用于树形结构处理、图遍历、流水线聚合等场景。

核心方法说明

你需要重点关注以下几个方法：

立即学习“Java免费学习笔记（深入）”；

compute()：主逻辑入口，通常在这里拆分任务或执行实际工作。
onCompletion(CountedCompleter>)：当 pending count 变为0时调用，用于汇总结果或触发后续动作。
tryComplete()：将当前任务的 pending count 减1，并向父任务传播完成状态。
addToPendingCount(int)：增加待完成计数，常用于 fork 子任务前。

使用示例：计算数组和（带回调）

下面是一个使用 CountedCompleter 实现并行求和的例子，展示如何控制任务完成时机。

AppMall应用商店

AI应用商店，提供即时交付、按需付费的人工智能应用服务

查看详情

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

public void compute() {
    if (hi - lo <= 1000) {
        // 小数据直接计算
        long sum = 0;
        for (int i = lo; i < hi; i++) {
            sum += array[i];
        }
        result = sum;
        tryComplete(); // 减少 pending count 并通知父任务
    } else {
        int mid = (lo + hi) / 2;
        addToPendingCount(1); // 增加一个待完成子任务
        new SumCompleter(this, array, lo, mid).fork();
        new SumCompleter(this, array, mid, hi).fork();
        tryComplete(); // 当前任务本身也算完成（结构已拆分）
    }
}

public void onCompletion(CountedCompleter<?> caller) {
    // 所有子任务完成后执行
    if (result == null) {
        result = 0L;
        // 遍历所有子任务结果
        for (SumCompleter child = rawResultOf(getFirstChild());
             child != null;
             child = rawResultOf(child.getNextSibling())) {
            result += child.result;
        }
    }
}

@SuppressWarnings("unchecked")
private static <T> T rawResultOf(CountedCompleter<?> c) {
    return c != null ? (T)c.getRawResult() : null;
}

public static void main(String[] args) {
    int[] data = new int[10000];
    for (int i = 0; i < data.length; i++) data[i] = 1;

    SumCompleter task = new SumCompleter(null, data, 0, data.length);
    ForkJoinPool.commonPool().invoke(task);

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

登录后复制

}

在这个例子中：