文章摘要
ForkJoinPool
是Java中的并行计算框架,其优点在于能够高效利用多核处理器资源,它采用分治策略将大任务拆分成小任务,通过工作窃取算法平衡负载,从而实现任务的并行执行和快速完成,此外,ForkJoinPool
还提供了简洁的API和丰富的任务控制机制,支撑开发人员开发高效的并行代码。
核心概念
ForkJoinPool 是 Java 并发包 java.util.concurrent
中的一个类,它主要用于解决可以通过分治策略(Divide-and-Conquer)来并行处理的问题,这类问题通常可以被分解为更小的子问题,子问题和原问题在结构上相同或类似,只不过规模不同,通过递归地将问题分解为更小的部分,ForkJoinPool
可以利用多核处理器并行地处理这些子问题,然后再将结果合并起来,从而高效地解决问题。
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是Java并行计算的利器,其优点在于能高效地将大任务拆成小任务,通过工作窃取机制充分利用多核处理器,加速任务执行,但它也有缺点,比如任务划分不均可能导致部分处理器闲置,且更适合计算密集型而非IO密集型任务,使用时,建议合理划分任务,保持任务均衡,同时注意异常处理和线程资源管理。