ForkJoinPool

ForkJoinPool 是 Java 7 引入的一种用于并行任务处理的高级线程池，专门为解决可以递归分解的问题而设计。它背后的主要思想是工作窃取算法，能够有效处理大规模并行计算问题。以下是关于 ForkJoinPool 的详细讲解，从架构、工作窃取原理、底层机制、典型应用场景等方面进行分析。

1. ForkJoinPool 概览

ForkJoinPool 是 java.util.concurrent 包中的核心并行处理框架。它专门用于解决可以通过“分治法”处理的问题，即把大任务分解成更小的子任务，然后递归处理这些子任务，最后合并结果。

• 核心方法：
  • fork()：将任务提交给线程池，异步执行。
  • join()：等待并获取子任务的结果，阻塞当前线程直到任务完成。
• 核心类：
  • ForkJoinPool：管理线程的池，负责任务的调度。
  • ForkJoinTask<V>：所有可以分解和合并的任务的基类。RecursiveTask<V> 处理有返回值的任务，RecursiveAction 处理没有返回值的任务。

2. 分治模型（Divide and Conquer）

ForkJoinPool 的工作方式基于“分治法”：

1. 分解任务：一个大的任务被分解为多个小任务，每个任务相对简单，容易执行。
2. 并行执行：分解后的任务递归调用 fork()，每个任务被提交到线程池并行执行。
3. 合并结果：通过 join() 等待所有子任务完成，并将结果合并，形成最终的解决方案。

例如：并行排序、矩阵乘法、递归问题等可以通过分治法高效解决。

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

ForkJoinPool 的核心思想是工作窃取算法。在此模型中，每个工作线程都有自己的双端队列（Deque），每个线程优先执行自己队列中的任务。当一个线程的任务完成后，如果没有新的任务，它会从其他线程的双端队列“窃取”任务，以保持 CPU 资源的高效利用。

工作窃取的关键点：

1. 双端队列：每个工作线程都有一个双端队列，任务递归分解后放入队列的末尾。线程首先从末尾取任务进行执行（LIFO 模型），即优先执行最近生成的任务。
2. 窃取任务：当一个线程的双端队列为空时，它会尝试从其他线程的队列的头部窃取任务（FIFO 模型），以避免资源闲置。通过窃取最早生成的任务，平衡了负载。
3. 锁竞争最小化：由于线程只会操作自己的任务队列，只有在窃取任务时才会与其他线程发生竞争，这减少了锁争用。
4. 任务粒度的选择：如果任务被分解得过细，会导致过多的小任务在队列中增加管理开销；分解过粗则可能导致负载不均衡。因此，任务分解的粒度需要适中。

4. ForkJoinPool 内部机制

ForkJoinPool 通过一组 ForkJoinWorkerThread 来执行 ForkJoinTask。这些线程主要负责以下工作：

• 提交任务：通过 fork() 将任务放入自己的队列尾部。
• 执行任务：通过 poll() 从队列尾部取任务并执行。
• 窃取任务：当队列为空时，通过 steal() 从其他线程的队列头部窃取任务。

ForkJoinPool 的工作流：

1. 任务被递归分解，通过 fork() 提交到线程池中。
2. 每个工作线程优先从自己队列末尾取任务执行。
3. 当任务执行完或没有任务时，线程会从其他线程的队列头部窃取任务。
4. join() 用于等待子任务完成，合并结果。

这种设计保证了高效的并行执行，同时最大限度利用了系统的多核处理能力。

5. 底层实现机制

1. 双端队列：每个线程持有一个 ForkJoinTask 的双端队列。任务在递归分解时插入队列的末端，而任务的执行则从队列尾部取出任务（LIFO）。其他线程窃取任务时，从队列的头部开始窃取（FIFO）。

2. ForkJoinTask 管理：ForkJoinTask 本质上是一个轻量级的任务单元，继承自 Future 接口。它通过状态位管理任务的状态（如是否完成），并提供了 fork() 和 join() 方法来控制任务的调度。

3. 多线程协作：每个工作线程独立处理自己的任务队列，通过减少锁争用来提升性能。只有在任务窃取的过程中，线程才会进入短暂的同步块，确保线程安全。

4. 线程管理：ForkJoinPool 会根据任务的工作量动态调整线程池中的线程数量。在空闲时，线程池中的线程可能会进入休眠状态，等待新的任务到达。

6. 应用场景

ForkJoinPool 适用于以下并行任务：

• 并行递归计算：如斐波那契数列、递归矩阵计算。
• 大数据处理：可将数据分块并行处理，如 MapReduce。
• 并行排序算法：如并行归并排序、并行快速排序。
• 大规模图算法：通过并行计算图的各个部分来提高计算速度。

7. ForkJoinPool 的局限性

• 过度任务分解：如果任务分解得太细，会导致任务调度和管理的开销过大，抵消并行执行的优势。
• 负载不均衡：如果任务粒度差距过大，可能会导致某些线程负载过重，其他线程闲置。
• 递归栈深度问题：深度递归的任务可能导致栈溢出问题，需要进行递归优化。

总结

ForkJoinPool 是 Java 7 及之后版本中并行计算的核心工具，基于工作窃取算法高效利用多核 CPU 资源。它通过任务的递归分解、双端队列和任务窃取等机制，实现了线程间的高效协作和负载均衡，在大规模并行任务处理场景下表现优异。

下面是 ForkJoinPool 的详细示例，展示了如何使用它进行并行计算。这段代码计算一个数组的元素之和，通过递归将大任务分解成更小的子任务来实现。

示例：ForkJoinPool 计算数组元素和

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

class SumTask extends RecursiveTask<Long> {
    private static final int THRESHOLD = 1000; // 拆分任务的阈值
    private long[] array;
    private int start;
    private int end;

    // 构造方法
    public SumTask(long[] 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 class ForkJoinExample {
    public static void main(String[] args) {
        // 创建一个包含一百万个随机数的数组
        long[] array = new long[1000000];
        for (int i = 0; i < array.length; i++) {
            array[i] = (long) (Math.random() * 100);
        }

        // 创建ForkJoinPool
        ForkJoinPool pool = new ForkJoinPool();

        // 创建任务
        SumTask task = new SumTask(array, 0, array.length);

        // 提交任务并获得结果
        long result = pool.invoke(task);

        // 打印结果
        System.out.println("数组元素和: " + result);

        // 关闭线程池
        pool.shutdown();
    }
}

代码说明：

1. RecursiveTask：这是 ForkJoinTask 的子类，适用于有返回值的任务。SumTask 继承了 RecursiveTask<Long>，表示任务最终返回一个 Long 类型的结果。
2. 分治策略：
   • 当任务长度小于等于 THRESHOLD（阈值 1000）时，直接通过简单的 for 循环计算数组的部分和。
   • 当任务大于阈值时，将任务分成两部分，分别创建新的子任务 leftTask 和 rightTask，递归处理。
3. 并行执行：
   • leftTask.fork() 异步执行左半部分任务。
   • 当前线程继续执行右半部分任务（通过 rightTask.compute()）。
   • leftTask.join() 等待左半部分任务执行完毕，并获取结果。
4. 线程池管理：使用 ForkJoinPool 来管理并发任务执行。pool.invoke(task) 提交主任务并等待最终结果。

ForkJoinPool 执行过程：

• 主任务 SumTask 将大任务递归分解为小任务，每个任务计算一部分数组的和。
• ForkJoinPool 管理多个工作线程，每个线程并行处理子任务。
• 当一个线程没有任务时，它会通过工作窃取算法从其他线程的队列中窃取任务。

适用场景：

这个例子展示了如何通过 ForkJoinPool 并行处理大规模计算任务，如大数据处理、矩阵运算或并行排序等场景。

这个实现体现了 ForkJoin 模式的核心思想：分治法 和 并行递归处理。

ForkJoinPool 的公共线程池（即 ForkJoinPool.commonPool()）是 Java 8 引入的一种共享线程池，用于支持并行流、CompletableFuture、以及其他基于分治（fork-join）框架的任务处理。它是一个全局线程池，Java 应用中各类并行任务都可以共享这个池。

1. 公共线程池的概述

• 全局共享：ForkJoinPool.commonPool() 是一个全局共享的线程池，适用于并行任务的执行。所有基于 ForkJoinPool 的异步任务（如 CompletableFuture）在默认情况下都会使用这个公共线程池。
• 适用场景：公共线程池被设计用于计算密集型任务，尤其是需要分治法（Divide-and-Conquer）处理的任务。

2. 公共线程池的创建方式

在 JVM 启动时，ForkJoinPool.commonPool() 会被初始化，并在 JVM 进程中作为全局唯一的 ForkJoinPool 实例存在。

2.1 构建过程

公共线程池在 ForkJoinPool 类中通过静态块进行初始化。

javaCopy codestatic {
    int parallelism = Runtime.getRuntime().availableProcessors() - 1;
    common = new ForkJoinPool(parallelism, defaultForkJoinWorkerThreadFactory,
                              null, true);
}

• 并行度计算：默认的并行度设置为 CPU核心数 - 1，这是因为在大多数情况下，ForkJoinPool 会将主线程也算作一个工作线程，这样可以充分利用所有的 CPU 核心。
• 默认线程工厂：公共线程池使用 defaultForkJoinWorkerThreadFactory 创建线程，这是一种默认的线程工厂，用于创建 ForkJoinWorkerThread，这些线程负责执行任务。
• 无阻塞任务处理：公共线程池适用于计算密集型任务，而不适合 I/O 密集型任务，因为它的线程数有限。默认线程数等于 CPU 核心数（或少 1），并且这些线程是短生命周期的工作线程。

2.2 懒加载机制

• 公共线程池采用懒加载机制，这意味着它在首次被使用时才会被初始化。通常，当应用中有并行流、CompletableFuture、或显式调用 ForkJoinPool.commonPool() 时，才会触发公共线程池的初始化。

在 Java 8 中，ForkJoinPool.commonPool() 的初始化并不是通过传统的静态块完成的，而是通过一种懒加载机制。在 ForkJoinPool 类中，commonPool() 是一个静态方法，当该方法首次被调用时，commonPool 会被创建和初始化。

公共线程池的初始化过程

1. 懒加载机制：公共线程池的实例会在第一次调用时初始化。该机制通过 ForkJoinPool 类的静态变量来延迟创建，这样可以避免在 JVM 启动时立即消耗大量资源。
2. 默认并行度设置：ForkJoinPool 默认的并行度为 CPU 核心数 - 1。这是为了留出一个线程供主线程使用，从而确保所有 CPU 核心的充分利用。
3. 守护线程：公共线程池中的线程都是守护线程，这意味着它们不会阻止 JVM 退出。守护线程在没有任务可执行时会被自动回收，并在需要时再次创建。

创建流程

ForkJoinPool.commonPool() 的初始化过程通过以下步骤实现：

• 在首次调用 commonPool() 方法时，JVM 会检查公共线程池是否已创建。如果未创建，则会初始化一个 ForkJoinPool 实例。
• 初始化时，ForkJoinPool 会根据当前的 CPU 核心数设定默认并行度，并使用默认的线程工厂创建 ForkJoinWorkerThread。
• 公共线程池的线程会在没有任务时被自动回收，但一旦有新任务提交，它们会被重新启动​GitHub​GitHub​GitHub。

公共线程池的源码实现

在源码中，commonPool() 方法定义在 ForkJoinPool 类中，它是一个静态方法，用于返回单例的公共线程池实例。具体代码实现中，ForkJoinPool 会根据并行度配置创建工作线程，并确保任务分发的高效性​

GitHub​

GitHub。

总结

ForkJoinPool.commonPool() 是通过懒加载方式创建的全局共享线程池。通过这种实现方式，JVM 能够有效地管理资源，确保并行任务的高效处理，同时避免在不需要公共线程池时占用过多资源。

ForkJoinPool.commonPool() 是 Java 中一个全局共享的线程池。它在 Java 8 中引入，用于并行执行任务，尤其是支持并行流（Parallel Streams）、CompletableFuture 等异步任务。

1. 公共线程池的具体特性

1.1 线程池的并行度

• 默认并行度：commonPool 的默认并行度设置为 CPU 核心数 - 1。
  • 这是因为 ForkJoinPool 设计的初衷是为计算密集型任务提供最大并行度，同时保留一个线程供主线程使用。
  • 例如，如果系统是 8 核 CPU，则默认的并行度是 7。
  • 如果需要修改默认并行度，可以通过 JVM 参数 -Djava.util.concurrent.ForkJoinPool.common.parallelism=N 来设定，其中 N 是新的并行度。

1.2 线程池的最大/最小线程数量

• 最小线程数量：最小线程数与并行度一致，即 CPU 核心数 - 1。
• 最大线程数量：commonPool 没有严格限制的最大线程数，因为它支持 动态扩展，使用一种“托管阻塞”（ManagedBlocker）机制来应对线程阻塞场景。
  • 如果 ForkJoinPool 中的工作线程被阻塞，并且需要更多线程来确保任务的继续执行，它会临时增加新的工作线程。
  • 这些新增线程是临时的，会在任务完成后被回收。

1.3 线程的类型

• 公共线程池中的线程都是 守护线程，这意味着它们不会阻止 JVM 退出。
• 线程的生命周期由 ForkJoinPool 自行管理，它会根据任务量和并行度自动调整线程数。

1.4 工作窃取算法

• commonPool 采用 工作窃取算法 来调度任务。
  • 每个线程都有一个任务队列，当一个线程的任务完成时，它会尝试从其他线程的任务队列中“窃取”任务。
  • 这种算法可以最大程度地利用多核 CPU，并行执行计算密集型任务，提高任务处理效率。

2. 拒绝策略

• ForkJoinPool 没有传统的拒绝策略（如 ThreadPoolExecutor 中的 RejectedExecutionHandler），因为 ForkJoinPool 是基于工作窃取和动态扩展机制设计的。
• 如果任务无法在当前线程数内完成，ForkJoinPool 会动态创建新线程来处理任务，而不是立即拒绝任务。
• 只有在资源耗尽的极端情况下，任务可能会被丢弃，这种情况通常通过异常传播，而不是传统的拒绝策略。

3. ForkJoinPool.commonPool() 的限制和风险

• 线程饱和问题：尽管 ForkJoinPool 可以动态扩展，但在 I/O 密集型任务中可能会出现线程饱和的风险，因为线程会被长时间阻塞。
  • 为了解决这个问题，可以使用自定义线程池来处理 I/O 密集型任务，而将 ForkJoinPool 留给计算密集型任务。
• 共享线程池的冲突：因为 commonPool 是全局共享的，当并行流和 CompletableFuture 都在使用公共线程池时，它们可能会相互干扰，导致资源竞争和性能下降。

4. 优化建议

• 调优并行度：通过 -Djava.util.concurrent.ForkJoinPool.common.parallelism 参数来调整并行度，尤其是在特定的应用场景中，如大规模并行计算。
• 使用自定义线程池：如果公共线程池无法满足任务需求，可以为不同类型的任务（如 I/O 密集型任务和计算密集型任务）配置自定义线程池。

5. 代码示例

以下是一个简单示例，展示如何使用 ForkJoinPool 公共线程池和设置并行度：

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

public class CommonPoolExample {
    static class SumTask extends RecursiveTask<Integer> {
        private final int[] array;
        private final int start;
        private final int end;

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

        @Override
        protected Integer compute() {
            if (end - start <= 10) { // 简单的任务阈值
                int 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();
                int rightResult = rightTask.compute();
                int leftResult = leftTask.join();
                return leftResult + rightResult;
            }
        }
    }

    public static void main(String[] args) {
        int[] array = new int[100];
        for (int i = 0; i < array.length; i++) {
            array[i] = i + 1;
        }

        ForkJoinPool commonPool = ForkJoinPool.commonPool();
        System.out.println("并行度: " + commonPool.getParallelism());

        SumTask task = new SumTask(array, 0, array.length);
        int result = commonPool.invoke(task);
        System.out.println("结果: " + result);
    }
}

在这个示例中，commonPool.getParallelism() 会显示公共线程池的并行度，它等于 CPU 核心数 - 1。

总结

• 并行度：默认并行度是 CPU 核心数 - 1，通过 JVM 参数可以调整。
• 最大线程数：没有严格的最大线程数限制，会根据需求动态扩展。
• 拒绝策略：无传统拒绝策略，依赖工作窃取和动态扩展机制。
• 适用场景：ForkJoinPool 适合计算密集型任务，不适合 I/O 密集型任务。