当前位置：首页 > article >正文

【Java】ForkJoin 框架

article 2025/8/23 18:03:24

在Java中，ForkJoin框架是并行编程的一个重要工具，它主要用于处理可以分解为多个子任务的复杂任务。ForkJoin框架的核心是ForkJoinPool，它是一个线程池，专门用于执行ForkJoinTask任务。通过将大任务分解为多个小任务，并在多个线程中并行执行这些小任务，ForkJoin框架可以显著提高程序的执行效率。

1.核心组件

• ForkJoinPool

• 这是ForkJoin框架的核心线程池，用于管理和调度ForkJoinTask任务。

• 它使用工作窃取算法（work-stealing algorithm），允许空闲的线程从其他线程的任务队列中窃取任务来执行，从而提高线程的利用率。

• 示例代码：

    ForkJoinPool pool = new ForkJoinPool();

• ForkJoinTask

• 这是ForkJoin框架中任务的抽象基类，所有自定义的任务都需要继承这个类。

• 通常使用RecursiveTask（有返回值的任务）或RecursiveAction（无返回值的任务）这两个子类来实现具体任务。

• 示例代码：

    public class MyTask extends RecursiveTask<Integer> {private int threshold;private int start;private int end;public MyTask(int threshold, int start, int end) {this.threshold = threshold;this.start = start;this.end = end;}@Overrideprotected Integer compute() {int sum = 0;if (end - start <= threshold) {for (int i = start; i < end; i++) {sum += i;}} else {int middle = (start + end) / 2;MyTask leftTask = new MyTask(threshold, start, middle);MyTask rightTask = new MyTask(threshold, middle, end);leftTask.fork(); // 异步执行子任务rightTask.fork(); // 异步执行子任务sum = leftTask.join() + rightTask.join(); // 等待子任务完成并获取结果}return sum;}}

2.工作窃取算法

• 工作窃取算法是一种高效的线程调度算法，用于解决线程空闲时的负载均衡问题。

• 每个线程都有自己的双端队列（deque），用于存储任务。

• 当一个线程的任务队列为空时，它可以从其他线程的任务队列中“窃取”任务来执行。

• 这种算法可以有效减少线程的空闲时间，提高线程的利用率。

3.使用示例

• 下面是一个完整的使用ForkJoin框架计算数组和的示例：

  import java.util.concurrent.ForkJoinPool;import java.util.concurrent.RecursiveTask;public class ForkJoinExample {public static void main(String[] args) {int[] array = new int[1000000];for (int i = 0; i < array.length; i++) {array[i] = i;}ForkJoinPool pool = new ForkJoinPool();SumTask task = new SumTask(array, 0, array.length);int sum = pool.invoke(task);System.out.println("Sum: " + sum);}}class SumTask extends RecursiveTask<Integer> {private static final int THRESHOLD = 1000;private int[] array;private int start;private int end;public SumTask(int[] array, int start, int end) {this.array = array;this.start = start;this.end = end;}@Overrideprotected Integer compute() {if (end - start <= THRESHOLD) {int sum = 0;for (int i = start; i < end; i++) {sum += array[i];}return sum;} else {int middle = (start + end) / 2;SumTask leftTask = new SumTask(array, start, middle);SumTask rightTask = new SumTask(array, middle, end);leftTask.fork();rightTask.fork();return leftTask.join() + rightTask.join();}}}

4.优势

• 并行处理：通过将任务分解为多个子任务，并在多个线程中并行执行，可以显著提高程序的执行效率。

• 负载均衡：工作窃取算法可以有效解决线程空闲时的负载均衡问题，提高线程的利用率。

• 易于使用：ForkJoin框架提供了简单易用的API，使得并行任务的实现变得非常方便。

5.注意事项

• 任务分解粒度：任务分解的粒度需要适中。如果任务分解得太细，会导致线程的调度开销过大；如果任务分解得太粗，又无法充分利用多核CPU的优势。

• 线程池大小：合理配置ForkJoinPool的线程池大小，通常建议设置为CPU核心数的两倍左右。

• 线程安全：虽然ForkJoin框架本身是线程安全的，但在任务执行过程中，如果需要访问共享资源，仍然需要注意线程安全问题。

通过合理使用ForkJoin框架，可以有效提高Java程序的并行处理能力，从而提升程序的性能。

public class ForkJoinPoolTest extends RecursiveTask<Long> {private static final int THRESHOLD = 500;long[] array;int start;int end;ForkJoinPoolTest(long[] array, int start, int end) {this.array = array;this.start = start;this.end = end;}public static void main(String[] args) {long[] arrays = new long[1000];long expecteSum = 0;for (int i = 0; i < arrays.length; i++) {arrays[i] = i;expecteSum = expecteSum + arrays[i];}System.out.println(expecteSum);ForkJoinPoolTest forkJoinPoolTest = new ForkJoinPoolTest(arrays, 0, arrays.length);long startTime = System.currentTimeMillis();Long result = ForkJoinPool.commonPool().invoke(forkJoinPoolTest);long endTime = System.currentTimeMillis();System.out.println("fork/join sum:" + result + ",耗时：" + (endTime - startTime));}@Overrideprotected Long compute() {//如果任务太小if (end - start <= THRESHOLD) {System.out.println("start=" + start + ",end=" + end);long sum = 0;for (int i = start; i < end; i++) {sum = sum + this.array[i];try {TimeUnit.MILLISECONDS.sleep(2);} catch (InterruptedException e) {e.printStackTrace();}}return sum;}int middle = (end + start) / 2;ForkJoinPoolTest joinPoolTest1 = new ForkJoinPoolTest(this.array, start, middle);ForkJoinPoolTest joinPoolTest2 = new ForkJoinPoolTest(this.array, middle, end);invokeAll(joinPoolTest1, joinPoolTest2);Long result1 = joinPoolTest1.join();Long result2 = joinPoolTest2.join();long result = result1 + result2;System.out.println("result = " + (result1 + result2));return result;}}