Java并发特性之 ForkJoinPool详解！-工具盒子

嗨，你好呀，我是猿java

ForkJoinPool是Java 7 引入的一种线程池实现，专门用于支持"大规模并行"任务的执行。那么，它和普通的线程池（ThreadPoolExecutor）有什么本质的区别呢？这篇文章我们将深入探讨 Fork/Join 框架的工作原理。

Fork/Join 框架简介 {#1-Fork-Join-框架简介} =====================================

Fork/Join 框架是一种并行计算框架，设计目的是提高具有递归性质任务的执行速度。典型的任务是将问题逐步分解成较小的任务，直到每一个子任务足够简单可以直接解决，然后再将结果聚合起来。

1.1 工作原理 {#1-1-工作原理}

Fork/Join 框架基于"工作窃取"算法 (Work Stealing Algorithm)，该算法的核心思想是每个工作线程有自己的任务队列(双端队列, Deque)。当一个线程完成了自己队列中的任务时，便会窃取其他线程队列中的任务执行，这样就不会因为某个线程在等待而浪费 CPU 资源。

具体的工作原理如下：

任务拆分：框架会将任务递归地拆分成更小的任务，分别放入不同的队列。
工作窃取：每个线程都尝试从队列中取任务执行。当一个线程完成了自己的任务队列后会尝试随机从其他队列拿任务继续执行，保证 CPU 资源尽可能地不闲置。
任务合并：线程在执行完任务后，会尝试合并(Join)这些任务的结果，直到获得最终结果。

ForkJoin Pool 核心组件 {#2-ForkJoin-Pool-核心组件} =============================================

ForkJoin 框架是由以下 3个重要组件组成的：

ForkJoinPool
ForkJoinTask
RecursiveTask & RecursiveAction

2.1 ForkJoinPool {#2-1-ForkJoinPool}

ForkJoinPool 是整个框架的核心，它是一个线程池，负责调度和分发任务。内部虽然类似于 ThreadPoolExecutor，但是与普通线程池有显著的不同：

工作窃取机制：每个工作者线程会有自己的任务队列，并且工作者线程可以相互"偷窃"任务。
任务分解与合并：该池在运行时会递归地分割大任务，并使其尽量并行化。
最优并发级别：默认情况下，它与 CPU 核心线程数量相同，确保最大限度地利用多核 CPU。

ForkJoinPool 具有两种模式：

普通模式：适用于简单任务的并行拆分和合并。
自定义模式：通过提供特定的策略，可以更灵活地控制任务执行的过程与行为。

2.2 ForkJoinTask {#2-2-ForkJoinTask}

ForkJoinTask是 Fork/Join 框架中的基础任务对象。ForkJoinTask 是一个抽象类，它提供了 fork 和 join 这两个关键的操作。在具体使用过程中，一般情况下我们不会直接使用它，而是使用它的两个子类：

RecursiveTask: 适用于有返回值的任务。
RecursiveAction: 适用于无返回值的任务。

`fork()` 和 `join()` {#fork-和-join}

fork(): 将任务提交给线程池，让线程池执行任务。
join(): 等待任务执行完成，并获取任务结果。

下面，我们通过一个简单的示例进行说明：

假设有一个递归算法任务，可以通过 RecursiveTask 实现这样一个任务：

|------------------------------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | 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 | class FibonacciTask extends RecursiveTask<Integer> { private final int n; FibonacciTask(int n) { this.n = n; } @Override protected Integer compute() { if (n <= 1) { return n; } FibonacciTask f1 = new FibonacciTask(n - 1); FibonacciTask f2 = new FibonacciTask(n - 2); f1.fork(); // 异步执行 return f2.compute() + f1.join(); // 等待结果并合并 } } public class ForkJoinExample { public static void main(String[] args) { ForkJoinPool pool = new ForkJoinPool(); FibonacciTask task = new FibonacciTask(10); System.out.println(pool.invoke(task)); // 输出 Fibonacci(10) 的结果：55 } } |

在上面的例子中，FibonacciTask是一个递归计算斐波那契数列的任务，使用了fork()将递归任务分解并提交给ForkJoinPool，然后通过join()合并结果。

2.3 RecursiveTask & RecursiveAction {#2-3-RecursiveTask-RecursiveAction}

RecursiveTask: 适合有返回值的递归任务。
RecursiveAction: 适合无返回值的递归任务，比如可以用于文件或者目录的遍历操作，在这种场景中任务只是执行不需要有返回结果。

RecursiveTask 和 RecursiveAction 都是 ForkJoinTask 的子类，设计上它们旨在有效地利用多核处理器分而治之，提升计算速度。

RecursiveAction 示例：

|------------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | 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 | class ArrayTransformAction extends RecursiveAction { private final int[] arr; private final int start, end; ArrayTransformAction(int[] arr, int start, int end) { this.arr = arr; this.start = start; this.end = end; } @Override protected void compute() { if (end - start <= 10) { // 当任务足够小直接计算 for (int i = start; i < end; i++) { arr[i] *= 2; // 假定简单的任务：每一个数字乘以 2 } } else { // 任务切分 int middle = (start + end) / 2; ArrayTransformAction task1 = new ArrayTransformAction(arr, start, middle); ArrayTransformAction task2 = new ArrayTransformAction(arr, middle, end); invokeAll(task1, task2); // 并行处理两个子任务 } } } |

在这个例子中，ArrayTransformAction 是一个无返回值的递归任务，利用 ForkJoinPool 执行可以使代码有效利用多核 CPU 并行处理任务。

与普通线程池对比 {#3-与普通线程池对比} =========================

任务分解

普通线程池（如ThreadPoolExecutor）通常用于处理相对独立的任务，每个任务通常不会再被拆分。
ForkJoinPool则专注于可以递归拆分的任务。

工作窃取

普通线程池没有实现工作窃取机制，这意味着如果一个线程完成了任务，它可能会闲置。
ForkJoinPool通过工作窃取算法，确保线程在完成自己的任务后可以继续从其他线程中获取任务，提高了资源利用率。
线程管理
普通线程池可以根据配置动态调整线程的数量。
ForkJoinPool通常在初始化时确定线程数量，通常设置为等于或略大于可用处理器的数量。
适用场景
普通线程池适用于需要处理大量独立任务的场景，如Web服务器处理请求。
ForkJoinPool适用于需要处理大规模数据并可以分解为子任务的场景。

使用场景 {#4-使用场景} =================

Fork/Join 框架非常适合以下这些工作负载：

递归任务：如斐波那契数列、归并排序等分治算法。
大规模数据处理：快速对集合、数组等进行并行操作。
图像处理：图像处理等数据量大的任务可以被分成多个小任务并行处理。

此外，Fork/Join 在某些场景下的效率甚至优于类似的 MapReduce 计算框架。对 Java 并行流 (Stream API parallelism) 的支持也使用了 ForkJoin 框架，因此在 Java Stream 中进行并行处理的场景中，底层就是通过 Fork/JoinPool 来处理的。

注意事项 {#5-注意事项} =================

对于每种线程池都有其擅长的领域，同时存在局限性，对于ForkJoinPool也一样，因此，在实际使用中，我们应该注意以下事项：

1. 控制任务粒度

如果 Fork/Join 任务拆分得过于细小，会导致过多的上下文切换及不必要的线程创建消耗性能，通常建议其中的任务不到一个门槛便停止分裂。你可以根据任务执行时间、负载平衡等条件，动态地设置任务分解的阈值。

2. 避免 IO 密集型任务

ForkJoin 优化了 CPU 密集型任务。而包含大量 IO 操作的任务，容易导致线程阻塞， Fork/Join 效率并不高。因此，对于 IO 密集型任务，推荐使用传统的线程池来控制线程数量和资源分配，而避免使用 Fork/Join。

3. CPU 核心数的考量

ForkJoinPool 的默认并行度是 Runtime.getRuntime().availableProcessors()，即根据 CPU 核心数来确定并行度。这符合 CPU 密集型任务的特点。但你也可以自定义 ForkJoinPool 的并行度。

4. 异常处理

在 Fork/Join 框架中，所有提交到池中的任务都是 ForkJoinTask 的子类，我们应当注意捕获异常防止任务执行中止。测试和异常处理可以通过提供自定义的方法钩子来协助调试。

总结 {#6-总结} =============

这篇文章，我们详细地分析了 ForkJoinPool线程池，Fork/Join 是专为递归分治设计的，它充分了利用了现代多核计算能力和工作窃取算法，为某个任务的并行化提供了高效的解决方案。但是，需要特别注意，Fork/Join 更适用于 CPU 密集型任务，对于 IO 密集型任务，其表现不一定理想。因此，实际工作中，对于Java提供的 ThreadPoolExecutor和ForkJoinPool线程池，一定要选择合适的适用场景。