【揭秘】ForkJoinPool全面解析

【揭秘】ForkJoinPool全面解析 - 程序员古德文章摘要

ForkJoinPool是Java中的并行计算框架，其优点在于能够高效利用多核处理器资源，它采用分治策略将大任务拆分成小任务，通过工作窃取算法平衡负载，从而实现任务的并行执行和快速完成，此外，ForkJoinPool还提供了简洁的API和丰富的任务控制机制，支撑开发人员开发高效的并行代码。

核心概念

ForkJoinPool 的主要特点包括：

工作窃取算法（Work-Stealing Algorithm）：当一个线程完成了自己的任务后，它可以从其他线程的任务队列中“窃取”任务来执行，这有助于平衡负载和提高处理器的利用率。

递归分解与合并：非常适合处理可以递归分解的问题，如排序、搜索、数值计算等，开发者需要实现 ForkJoinTask 接口（通常使用它的子类 RecursiveAction 用于无返回值的任务，或使用 RecursiveTask 用于有返回值的任务）来定义问题的分解和结果的合并。

非阻塞设计：使用内部队列来管理任务，避免了使用锁或其他同步机制，从而减少了线程间的竞争和阻塞。

并行度控制：允许开发者控制并行执行的线程数量，可以根据处理器的核心数来优化性能。

ForkJoinPool 适用于那些可以自然分解为多个独立子任务，并且这些子任务之间不需要太多通信或同步的问题，常见的使用场景包括并行数组处理（如排序、过滤、映射）、并行集合处理（如归约操作）、科学计算中的并行算法（如矩阵乘法、快速傅里叶变换）等。

代码案例

import java.util.concurrent.ForkJoinPool;

import java.util.concurrent.RecursiveAction;

public class ForkJoinSumCalculator {

public static void main(String[] args) {

// 定义一个需要求和的数组

int[] numbers = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};

// 创建一个ForkJoinPool实例，它将使用可用的所有处理器

ForkJoinPool pool = new ForkJoinPool();

// 创建一个ForkJoinTask来执行求和操作

SumTask task = new SumTask(numbers, 0, numbers.length);

// 提交任务到ForkJoinPool并等待它的完成

pool.invoke(task);

// 输出最终求和结果

System.out.println("Sum of all numbers: " + task.getSum());

// 关闭ForkJoinPool（虽然在这个例子中它并不是严格必要的，因为程序即将退出）

pool.shutdown();

}

// 定义一个继承自RecursiveAction的任务类

static class SumTask extends RecursiveAction {

private static final long serialVersionUID = 1L;

// 阈值，当数组长度小于这个值时，直接计算结果而不再拆分

private static final int THRESHOLD = 5;

private int[] numbers;

private int startIndex;

private int endIndex;

private int sum; // 存储子数组的和

public SumTask(int[] numbers, int startIndex, int endIndex) {

this.numbers = numbers;

this.startIndex = startIndex;

this.endIndex = endIndex;

}

// 获取当前任务计算的和

public int getSum() {

return sum;

}

@Override

protected void compute() {

// 如果任务足够小，直接计算

if (endIndex - startIndex <= THRESHOLD) {

sum = calculateDirectly();

} else {

// 否则，拆分任务

int middleIndex = startIndex + (endIndex - startIndex) / 2;

SumTask leftTask = new SumTask(numbers, startIndex, middleIndex);

SumTask rightTask = new SumTask(numbers, middleIndex, endIndex);

// 递归执行任务

invokeAll(leftTask, rightTask);

// 合并结果

sum = leftTask.getSum() + rightTask.getSum();

}

}

// 直接计算子数组的和

private int calculateDirectly() {

int localSum = 0;

for (int i = startIndex; i < endIndex; i++) {

localSum += numbers[i];

}

return localSum;

}

}

}

在上面代码中，SumTask类有一个sum字段来存储计算的和，以及一个getSum方法来检索它，在compute方法中，如果任务的大小超过阈值，任务将被拆分为两个子任务，并且递归地执行，然后，将子任务的结果合并以计算总和，如果任务的大小小于或等于阈值，将直接计算子数组的和。

核心API

ForkJoinPool 提供了一个框架，用于将大任务分解成小任务，然后并行地执行这些小任务，最后再将结果合并起来，它提供的方法主要涉及到任务的提交、执行、管理和配置等方面，下面是一些常用方法的简要说明。

构造方法

ForkJoinPool(): 创建一个默认并行级别的 ForkJoinPool，通常使用可用的处理器数量作为并行级别。

ForkJoinPool(int parallelism): 创建一个具有指定并行级别的 ForkJoinPool。

任务提交

invoke(ForkJoinTask<?> task): 同步执行指定的任务，并等待其完成。

submit(ForkJoinTask<?> task): 异步提交一个任务以供执行，并返回一个表示该任务的 Future。

execute(ForkJoinTask<?> task): 安排一个任务的执行，但不等待其完成。

任务管理

awaitQuiescence(long timeout, TimeUnit unit): 等待所有任务完成执行，或者直到超时。

shutdown(): 可能启动有序关闭，在该过程中执行现有任务但不接受新任务。

shutdownNow(): 试图停止所有正在执行的活动任务，暂停处理正在等待的任务，并返回等待执行的任务列表。

isShutdown(): 如果此池已关闭或正在关闭，则返回 true。

isTerminated(): 如果关闭后所有任务都已完成，则返回 true。

awaitTermination(long timeout, TimeUnit unit): 请求关闭并等待所有任务完成执行，或者直到超时。

获取任务结果

对于 RecursiveTask（有返回值的任务），通常会在调用任务的 join 方法时获取任务结果。

配置和状态

getParallelism(): 返回此 ForkJoinPool 的并行级别。

getPoolSize(): 返回此 ForkJoinPool 中的活动线程估计数。

getActiveThreadCount(): 返回此 ForkJoinPool 中当前活动的线程数。

getRunningThreadCount(): 返回此 ForkJoinPool 中正在运行任务的线程数。

getQueuedTaskCount(): 返回此 ForkJoinPool 工作队列中待处理的任务数估计值。

getStealCount(): 返回从此 ForkJoinPool 中成功窃取的任务数估计值。

核心总结

【揭秘】ForkJoinPool全面解析 - 程序员古德

ForkJoinPool是Java并行计算的利器，其优点在于能高效地将大任务拆成小任务，通过工作窃取机制充分利用多核处理器，加速任务执行，但它也有缺点，比如任务划分不均可能导致部分处理器闲置，且更适合计算密集型而非IO密集型任务，使用时，建议合理划分任务，保持任务均衡，同时注意异常处理和线程资源管理。