ForkJoinTask是Java中实现并行计算的核心，通过RecursiveTask（有返回值）或RecursiveAction（无返回值）进行任务拆分与合并，结合ForkJoinPool使用工作窃取算法提升性能，适用于求和、排序等分治场景。
在Java中，ForkJoinTask 是 ForkJoinPool 框架的核心组件，用于实现高效的任务并行计算。它特别适合将大任务拆分为多个小任务（分治法），然后合并结果。常见的使用方式是继承其子类 RecursiveTask（有返回值）或 RecursiveAction（无返回值）。
1. 引入 ForkJoinPool 和 ForkJoinTask
Java 7 引入了 java.util.concurrent.ForkJoinPool，它是线程池的一种，专为执行大量小型、异步任务而设计。它采用工作窃取算法（work-stealing），空闲线程可以从其他线程的任务队列中“窃取”任务，提高CPU利用率。
要使用 ForkJoinTask 执行并行计算，通常步骤如下：
- 创建一个继承 RecursiveTask 或 RecursiveAction 的任务类
- 重写 compute() 方法实现任务的拆分与合并逻辑
- 使用 ForkJoinPool 提交任务并获取结果
2. 使用 RecursiveTask 实现并行求和
下面以一个经典的例子——并行计算数组元素之和——来说明如何使用 RecursiveTask：
import java.util.concurrent.ForkJoinPool; import java.util.concurrent.RecursiveTask;public class SumTask extends RecursiveTask
{ private static final int THRESHOLD = 1000; // 任务拆分阈值 private long[] array; private int start, end; public SumTask(long[] array, int start, int end) { this.array = array; this.start = start; this.end = end; } @Override protected Long compute() { if (end - start zuojiankuohaophpcn= THRESHOLD) { // 小任务直接计算 long sum = 0; for (int i = start; i zuojiankuohaophpcn end; i++) { sum += array[i]; } return sum; } else { // 拆分为两个子任务 int mid = (start + end) / 2; SumTask leftTask = new SumTask(array, start, mid); SumTask rightTask = new SumTask(array, mid, end); leftTask.fork(); // 异步执行左任务 long rightResult = rightTask.compute(); // 当前线程执行右任务 long leftResult = leftTask.join(); // 等待左任务完成并获取结果 return leftResult + rightResult; } } public static void main(String[] args) { long[] data = new long[10000]; for (int i = 0; i zuojiankuohaophpcn data.length; i++) { data[i] = i + 1; } ForkJoinPool pool = new ForkJoinPool(); SumTask task = new SumTask(data, 0, data.length); long result = pool.invoke(task); // 提交任务并等待结果 System.out.println("并行计算结果: " + result); pool.shutdown(); }
}
3. RecursiveAction 的使用场景
如果任务不需要返回结果，比如对数组每个元素进行某种处理（如打印、修改），可以使用 RecursiveAction：
import java.util.concurrent.RecursiveAction;public class PrintTask extends RecursiveAction { private static final int THRESHOLD = 100; private int[] array; private int start, end;
public PrintTask(int[] array, int start, int end) { this.array = array; this.start = start; this.end = end; } @Override protected void compute() { if (end - start zuojiankuohaophpcn= THRESHOLD) { for (int i = start; i zuojiankuohaophpcn end; i++) { System.out.println("处理元素: " + array[i]); } } else { int mid = (start + end) / 2; PrintTask left = new PrintTask(array, start, mid); PrintTask right = new PrintTask(array, mid, end); left.fork(); right.compute(); left.join(); } }}
4. 注意事项与最佳实践
使用 ForkJoinTask 时要注意以下几点：
;
SumTask rightTask = new SumTask(array, mid, end);
leftTask.fork(); // 异步执行左任务
long rightResult = rightTask.compute(); // 当前线程执行右任务
long leftResult = leftTask.join(); // 等待左任务完成并获取结果
return leftResult + rightResult;
}
}
public static void main(String[] args) {
long[] data = new long[10000];
for (int i = 0; i zuojiankuohaophpcn data.length; i++) {
data[i] = i + 1;
}
ForkJoinPool pool = new ForkJoinPool();
SumTask task = new SumTask(data, 0, data.length);
long result = pool.invoke(task); // 提交任务并等待结果
System.out.println("并行计算结果: " + result);
pool.shutdown();
}- 避免阻塞操作：ForkJoinPool 的线程数量默认等于CPU核心数，执行阻塞任务会降低整体性能
- 合理设置阈值：任务拆分太细会导致调度开销过大；太大则无法充分利用并行性
- 优先使用 commonPool：可以通过 ForkJoinPool.commonPool() 获取公共线程池，避免频繁创建
- 异常处理：compute() 中抛出的异常会被封装，需通过 get() 或 isCompletedAbnormally() 检查
基本上就这些。ForkJoinTask 适合分治类问题，如归并排序、矩阵运算、树遍历等，能有效提升计算密集型任务的性能。
