ForkJoinPool详解

是什么

ForkJoinPool 是 Java 并发包（java.util.concurrent）中的一种特殊线程池，用于执行分治任务。

ForkJoinPool 使用了一种工作窃取（work-stealing）算法，允许线程动态获取其他线程队列中的任务，以最大化 CPU 利用率和任务处理效率。

使用场景

ForkJoinPool 特别适合以下场景(CPU密集型)：

  1. 递归分治算法：如快速排序、归并排序等，需要将问题分解成更小子问题的算法。
  2. 大规模数据处理：需要并行处理大数据集，如数组、集合的并行操作。

核心组件

ForkJoinTask：这是 ForkJoinPool 中执行的任务的基本类型，有两个主要子类：

  • RecursiveTask：有返回结果的任务。
  • RecursiveAction：没有返回结果的任务。

工作窃取算法：每个工作线程都有自己的双端队列（deque），线程可以从自己的队列头部获取任务执行。当一个线程完成自己的任务时，它会从其他线程的队列尾部窃取任务，以减少空闲时间，提高并行性。

示例代码

下面是一个使用 ForkJoinPool 计算大数组总和的示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;

public class ForkJoinExample {

    // 定义任务类，继承 RecursiveTask
    static class SumTask extends RecursiveTask<Long> {
        private static final int THRESHOLD = 10_000; // 任务分割阈值
        private final int[] array;
        private final int start;
        private final int end;

        SumTask(int[] 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) / 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) {
        int[] array = new int[100_000]; // 初始化一个大数组
        for (int i = 0; i < array.length; i++) {
            array[i] = i;
        }

        ForkJoinPool pool = new ForkJoinPool(); // 创建 ForkJoinPool 实例
        SumTask task = new SumTask(array, 0, array.length);
        long result = pool.invoke(task); // 提交任务并等待结果

        System.out.println("Sum: " + result); // 输出结果
    }
}

ForkJoinPool 与线程池的对比

线程池（ThreadPoolExecutor）

线程池 是 Java 并发包中的另一种核心组件，用于管理一组线程，以执行多个并发任务。线程池通过复用线程，减少了创建和销毁线程的开销，提高了性能。

主要区别

  1. 任务类型：

    • ForkJoinPool：适用于分治任务，将大任务分解成多个小任务并行执行。
    • ThreadPoolExecutor：相比之下适用于独立的、不可分解的任务。

  2. 工作原理：

    • ForkJoinPool：使用工作窃取算法，线程在完成自己的任务后，可以窃取其他线程的任务，最大化 CPU 利用率。
    • ThreadPoolExecutor：使用固定或动态数量的线程，从一个共享的任务队列中获取任务执行。

  3. 编程模型：

    • ForkJoinPool：基于 Fork/Join 框架，需要将任务实现为 ForkJoinTask 的子类（如 RecursiveTask 或 RecursiveAction）。
    • ThreadPoolExecutor：使用更通用的 Runnable 或 Callable 接口，适用于各种并发任务。

  4. 性能优化：

    • ForkJoinPool：适合处理 CPU 密集型任务，通过工作窃取算法提高并行效率。
    • ThreadPoolExecutor：适合处理 I/O 密集型任务或 CPU 密集型任务，通过调节线程池大小和任务队列优化性能。

总结

ForkJoinPool 和 ThreadPoolExecutor 是 Java 并发编程中的两种重要工具，各有其适用场景。ForkJoinPool 通过工作窃取算法和递归任务处理，特别适合分治算法和大规模数据处理。而 ThreadPoolExecutor 则提供了一个通用的并发执行框架，适用于各种独立并发任务。选择合适的工具，可以显著提高程序的并发性能和资源利用率。