如何在Java中使用Fork Join RecursiveTask-java教程-PHP中文网

如何在Java中使用Fork Join RecursiveTask

P粉602998670

发布： 2025-09-22 23:06:01

原创

344人浏览过

ForkJoin框架通过分而治之和工作窃取实现高效并行计算，适用于可递归分解的计算密集型任务。

如何在java中使用fork join recursivetask

Java的ForkJoin框架提供了一种高效并行处理任务的机制，特别是针对那些可以被递归分解成更小独立子任务的计算。其中

RecursiveTask

登录后复制

用于处理需要返回结果的任务，它通过“分而治之”的思想，将大问题拆解，并行计算，最终将子结果合并，从而充分利用多核处理器的性能。

要在Java中有效地使用

ForkJoinRecursiveTask

登录后复制

，核心在于理解其“分而治之”的策略。这通常涉及几个关键步骤，而且说实话，第一次接触时可能会觉得有点绕，但一旦掌握了模式，就会发现它非常强大。

首先，你需要一个

ForkJoinPool

登录后复制

，这是所有ForkJoin任务的执行场所。它管理着一组工作线程，并实现了“工作窃取”算法，确保CPU核心不会闲置。

接着，你需要定义一个继承自

RecursiveTask<V>

登录后复制

（

登录后复制

是你任务返回结果的类型）的类。这个类里最关键的就是重写

compute()

登录后复制

方法。

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

在

compute()

登录后复制

方法内部，你需要实现你的“分而治之”逻辑：

如知AI笔记

如知笔记——支持markdown的在线笔记，支持ai智能写作、AI搜索，支持DeepseekR1满血大模型

查看详情

确定基础任务（Base Case）：这是任务分解的最小单元。当一个任务足够小，不再需要进一步分解时，就直接执行它并返回结果。这个“足够小”的阈值（通常称为
```
THRESHOLD
```
登录后复制
）是性能调优的关键点之一，过大或过小都会影响效率。
递归分解（Recursive Case）：如果任务仍然太大，就把它拆分成两个（或更多）更小的子任务。
- 创建新的
```
RecursiveTask
```
  登录后复制
  实例来代表这些子任务。
- 使用
```
fork()
```
  登录后复制
  方法异步地提交一个子任务到
```
ForkJoinPool
```
  登录后复制
  。
```
fork()
```
  登录后复制
  会安排这个任务在一个可用的工作线程上执行。
- 另一个子任务可以选择直接调用其
```
compute()
```
  登录后复制
  方法（这通常被称为“帮助执行”或“就地执行”），或者也
```
fork()
```
  登录后复制
  出去。实践中，通常会
```
fork()
```
  登录后复制
  一个，然后当前线程直接处理另一个，这样可以减少线程切换的开销。
- 使用
```
join()
```
  登录后复制
  方法等待已
```
fork()
```
  登录后复制
  的子任务完成并获取其结果。
```
join()
```
  登录后复制
  会阻塞当前线程，直到对应的任务完成。
- 最后，将所有子任务的结果合并，形成当前任务的最终结果。

下面是一个简单的例子，演示如何使用

RecursiveTask

登录后复制

来并行计算一个大数组中所有元素的和：

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.stream.LongStream; // 用于生成测试数据

// 继承RecursiveTask，并指定返回类型为Long
class SumArrayTask extends RecursiveTask<Long> {
    private final long[] array;
    private final int start;
    private final int end;
    // 定义一个阈值，当子任务的长度小于等于这个值时，就直接计算
    private static final int THRESHOLD = 10_000; 

    public SumArrayTask(long[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        // 如果任务规模小于等于阈值，直接计算
        if (end - start <= THRESHOLD) {
            long sum = 0;
            for (int i = start; i < end; i++) {
                sum += array[i];
            }
            return sum;
        } else {
            // 否则，将任务分解成两个子任务
            int mid = start + (end - start) / 2;
            SumArrayTask leftTask = new SumArrayTask(array, start, mid);
            SumArrayTask rightTask = new SumArrayTask(array, mid, end);

            // 异步执行左侧子任务
            leftTask.fork();

            // 同步执行右侧子任务（当前线程可能直接执行）
            Long rightResult = rightTask.compute();

            // 等待左侧子任务完成并获取结果
            Long leftResult = leftTask.join();

            // 合并结果
            return leftResult + rightResult;
        }
    }

    public static void main(String[] args) {
        long[] data = LongStream.rangeClosed(1, 10_000_000).toArray(); // 创建一个大数组

        // 创建ForkJoinPool，通常使用默认的公共池
        ForkJoinPool pool = new ForkJoinPool(); 
        // 或者使用 ForkJoinPool.commonPool();

        // 创建主任务
        SumArrayTask mainTask = new SumArrayTask(data, 0, data.length);

        // 提交任务并获取结果
        long startTime = System.currentTimeMillis();
        Long result = pool.invoke(mainTask); // invoke()会阻塞直到任务完成
        long endTime = System.currentTimeMillis();

        System.out.println("计算结果: " + result);
        System.out.println("耗时: " + (endTime - startTime) + " ms");

        // 验证结果（可选）
        long expectedSum = LongStream.rangeClosed(1, 10_000_000).sum();
        System.out.println("预期结果: " + expectedSum);
        System.out.println("结果是否正确: " + (result == expectedSum));

        // 关闭线程池，如果使用的是commonPool则不需要手动关闭
        // pool.shutdown(); 
    }
}

登录后复制

通过