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

什么是 ForkJoinPool

news 来源：原创 2025/6/20 23:15:23

1. 什么是 ForkJoinPool？

定义

ForkJoinPool 是 Java 提供的一种特殊的线程池，专门用于处理可以分解成多个小任务的大任务。它的核心思想是“分而治之”（Divide and Conquer），即将一个复杂的问题拆分成多个小问题，并行解决后再合并结果。

适用场景

处理递归任务：例如排序算法（快速排序、归并排序）、文件搜索、大数据计算等。
需要高效管理大量细粒度任务的场景。

2. ForkJoinPool 的核心概念

2.1 分而治之（Divide and Conquer）

大任务：一个复杂的问题，比如对一个百万元素的数组进行排序。
小任务：将大任务拆分成多个更小的任务，比如分别对数组的前半部分和后半部分排序。
合并结果：将小任务的结果合并起来，得到最终结果。

2.2 工作窃取算法（Work-Stealing Algorithm）

每个线程都有自己的任务队列。
如果某个线程完成了自己的任务，它会从其他线程的任务队列中“偷”任务来执行。
这种机制可以减少线程空闲时间，提高资源利用率。

3. ForkJoinPool 的基本结构

3.1 核心类

ForkJoinPool：
- 负责管理线程池和任务调度。
- 默认情况下，JVM 提供了一个全局共享的线程池 ForkJoinPool.commonPool()。
ForkJoinTask：
- 表示一个可以被 ForkJoinPool 执行的任务。
- 常见的子类包括 RecursiveTask 和 RecursiveAction。

3.2 RecursiveTask 和 RecursiveAction

RecursiveTask<V>：
- 表示有返回值的任务。
- 需要实现 compute() 方法，在其中定义任务逻辑。
RecursiveAction：
- 表示没有返回值的任务。
- 同样需要实现 compute() 方法。

4. ForkJoinPool 的工作流程

步骤 1：任务分解

将一个大任务拆分成多个小任务。如果小任务仍然太大，则继续拆分，直到任务足够小。

步骤 2：并行执行

每个小任务会被分配到不同的线程中执行。

步骤 3：合并结果

当所有小任务完成后，将它们的结果合并，得到最终结果。

5. 示例代码

5.1 使用 RecursiveTask 计算数组的总和

import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;

public class ForkJoinExample {

    public static void main(String[] args) {
        // 创建 ForkJoinPool
        ForkJoinPool pool = new ForkJoinPool();

        // 创建任务
        int[] numbers = new int[100];
        for (int i = 0; i < numbers.length; i++) {
            numbers[i] = i + 1; // 初始化数组为 [1, 2, ..., 100]
        }

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

        // 提交任务并获取结果
        int result = pool.invoke(task);
        System.out.println("Sum: " + result); // 输出结果：5050
    }

    // 定义任务类
    static class SumTask extends RecursiveTask<Integer> {
        private final int[] numbers;
        private final int start;
        private final int end;

        // 如果任务量小于阈值（如 10），直接计算；否则继续拆分
        private static final int THRESHOLD = 10;

        public SumTask(int[] numbers, int start, int end) {
            this.numbers = numbers;
            this.start = start;
            this.end = end;
        }

        @Override
        protected Integer compute() {
            if (end - start <= THRESHOLD) {
                // 基础情况：直接计算
                int sum = 0;
                for (int i = start; i < end; i++) {
                    sum += numbers[i];
                }
                return sum;
            } else {
                // 分治：将任务拆分为两部分
                int mid = (start + end) / 2;
                SumTask leftTask = new SumTask(numbers, start, mid);
                SumTask rightTask = new SumTask(numbers, mid, end);

                // 异步执行左任务
                leftTask.fork();

                // 同步执行右任务
                int rightResult = rightTask.compute();

                // 等待左任务完成并合并结果
                int leftResult = leftTask.join();
                return leftResult + rightResult;
            }
        }
    }
}

解释

任务分解：
- 如果数组长度大于 THRESHOLD（10），将数组拆分为两部分。
- 递归调用 compute() 直到任务足够小。
并行执行：
- 左任务通过 fork() 异步执行。
- 右任务通过 compute() 同步执行。
合并结果：
- 使用 join() 获取左任务的结果。
- 将左右任务的结果相加，得到最终结果。

输出

Sum: 5050

6. ForkJoinPool 的默认线程池

默认线程池

JVM 提供了一个全局共享的线程池 ForkJoinPool.commonPool()。
默认线程数为 Runtime.getRuntime().availableProcessors() - 1（即 CPU 核心数减一）。

如何使用默认线程池

ForkJoinTask<Integer> task = new SumTask(numbers, 0, numbers.length);
int result = ForkJoinPool.commonPool().invoke(task);
System.out.println("Sum: " + result);