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

Fork/Join 框架详解：分支合并的高性能并发编程

news 来源：原创 2025/5/17 8:19:27

引言

一、Fork/Join 框架概述

1.1 什么是 Fork/Join 框架？

1.2 Fork/Join 框架的核心组件

二、Fork/Join 框架的使用步骤

三、Fork/Join 框架的示例

3.1 示例 1：计算数组元素之和

代码实现

代码解析

3.2 示例 2：并行排序

代码实现

代码解析

四、Fork/Join 框架的优缺点

4.1 优点

4.2 缺点

五、总结

引言

在并发编程中，任务的分解与合并是一个常见的需求。Java 7 引入了 Fork/Join 框架，它是一种基于分治算法（Divide and Conquer）的并行计算框架，专门用于解决可以分解为多个子任务的问题。Fork/Join 框架的核心思想是将一个大任务拆分为多个小任务（Fork），然后将这些小任务的结果合并（Join），从而实现高效的并行计算。

一、Fork/Join 框架概述

1.1 什么是 Fork/Join 框架？

Fork/Join 框架是 Java 提供的一种用于并行计算的框架，它的核心思想是：

Fork（分解）：将一个大任务拆分为多个小任务。
Join（合并）：将小任务的结果合并为最终结果。

它的模型大致是这样的：线程池中的每个线程都有自己的工作队列（PS：这一点和ThreadPoolExecutor 不同，ThreadPoolExecutor是所有线程公用一个工作队列，所有线程都从这个工作队列中取任务），当自己队列中的任务都完成以后，会从其它线程的工作队列中偷一个任务执行，这样可以充分利用资源。

Fork/Join 框架通过 工作窃取算法（Work-Stealing Algorithm） 实现任务的动态分配和负载均衡，从而充分利用多核 CPU 的性能。

简单理解，就是一个工作线程下会维护一个包含多个子任务的双端队列。而对于每个工作线程来说，会从头部到尾部依次执行任务。这时，总会有一些线程执行的速度较快，很快就把所有任务消耗完了。那这个时候怎么办呢，总不能空等着吧，多浪费资源啊。

工作窃取算法是指某个线程从其他队列里窃取任务来执行。工作窃取的运行流程图如下：

那么为什么需要使用工作窃取算法呢？

假如我们需要做一个比较大的任务，我们可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，于是把这些子任务分别放到不同的队列里，并为每个队列创建一个单独的线程来执行队列里的任务，线程和队列一一对应，比如A线程负责处理A队列里的任务。但是有的线程会先把自己队列里的任务干完，而其他线程对应的队列里还有任务等待处理。干完活的线程与其等着，不如去帮其他线程干活，于是它就去其他线程的队列里窃取一个任务来执行。而在这时它们会访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。

工作窃取算法的优点是充分利用线程进行并行计算，并减少了线程间的竞争，其缺点是在某些情况下还是存在竞争，比如双端队列里只有一个任务时。并且消耗了更多的系统资源，比如创建多个线程和多个双端队列。

于是，先做完任务的工作线程会从其他未完成任务的线程尾部依次获取任务去执行。这样就可以充分利用CPU的资源。这个非常好理解，就比如有个妹子程序员做任务比较慢，那么其他猿就可以帮她分担一些任务，这简直是双赢的局面啊，妹子开心了，你也开心了。

1.2 Fork/Join 框架的核心组件

ForkJoinPool：任务执行的线程池，负责管理任务的调度和执行。
ForkJoinTask：表示一个可以被 Fork/Join 框架执行的任务。子类：
- RecursiveTask：用于有返回值的任务。
- RecursiveAction：用于无返回值的任务。

主要方法：

fork():在当前线程运行的线程池中安排一个异步执行。简单的理解就是再创建一个子任务。

join():当任务完成的时候返回计算结果。

invoke(): 开始执行任务，如果必要，等待计算完成

二、Fork/Join 框架的使用步骤

使用 Fork/Join 框架通常包括以下步骤：

定义任务：继承 RecursiveTask 或 RecursiveAction，实现任务的分解和合并逻辑。
创建线程池：使用 ForkJoinPool 创建线程池。
提交任务：将任务提交到线程池中执行。
获取结果：如果是 RecursiveTask，可以通过 join() 方法获取任务的结果。

三、Fork/Join 框架的示例

3.1 示例 1：计算数组元素之和

以下是一个使用 Fork/Join 框架计算数组元素之和的示例。

代码实现

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

public class SumTask extends RecursiveTask<Integer> {
    private static final int THRESHOLD = 10; // 任务拆分的阈值
    private int[] array;
    private int start;
    private 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 <= THRESHOLD) {
            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();
            rightTask.fork();

            // 合并子任务的结果
            return leftTask.join() + rightTask.join();
        }
    }

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

        // 创建 ForkJoinPool
        ForkJoinPool pool = new ForkJoinPool();
        // 提交任务
        SumTask task = new SumTask(array, 0, array.length);
        int result = pool.invoke(task);

        System.out.println("Sum: " + result); // 输出 Sum: 5050
    }
}

代码解析

任务拆分：
- 如果任务的大小（end - start）小于等于阈值（THRESHOLD），则直接计算。
- 否则，将任务拆分为两个子任务（leftTask 和 rightTask），并通过 fork() 方法并行执行。
结果合并：
- 使用 join() 方法获取子任务的结果，并将其合并。
线程池执行：
- 使用 ForkJoinPool 提交任务，并通过 invoke() 方法触发任务执行。

3.2 示例 2：并行排序

以下是一个使用 Fork/Join 框架实现并行排序的示例。

代码实现

import java.util.Arrays;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.ForkJoinPool;

public class ParallelMergeSort extends RecursiveAction {
    private static final int THRESHOLD = 10; // 任务拆分的阈值
    private int[] array;
    private int start;
    private int end;

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

    @Override
    protected void compute() {
        if (end - start <= THRESHOLD) {
            // 如果任务足够小，直接排序
            Arrays.sort(array, start, end);
        } else {
            // 拆分任务
            int mid = (start + end) / 2;
            ParallelMergeSort leftTask = new ParallelMergeSort(array, start, mid);
            ParallelMergeSort rightTask = new ParallelMergeSort(array, mid, end);

            // 并行执行子任务
            invokeAll(leftTask, rightTask);

            // 合并子任务的结果
            merge(array, start, mid, end);
        }
    }

    private void merge(int[] array, int start, int mid, int end) {
        int[] temp = new int[end - start];
        int i = start, j = mid, k = 0;

        while (i < mid && j < end) {
            if (array[i] < array[j]) {
                temp[k++] = array[i++];
            } else {
                temp[k++] = array[j++];
            }
        }

        while (i < mid) {
            temp[k++] = array[i++];
        }

        while (j < end) {
            temp[k++] = array[j++];
        }

        System.arraycopy(temp, 0, array, start, temp.length);
    }

    public static void main(String[] args) {
        int[] array = {5, 3, 9, 1, 7, 2, 8, 4, 6};

        // 创建 ForkJoinPool
        ForkJoinPool pool = new ForkJoinPool();
        // 提交任务
        ParallelMergeSort task = new ParallelMergeSort(array, 0, array.length);
        pool.invoke(task);

        System.out.println("Sorted Array: " + Arrays.toString(array));
    }
}

代码解析

任务拆分：
- 如果任务的大小（end - start）小于等于阈值（THRESHOLD），则直接排序。
- 否则，将任务拆分为两个子任务（leftTask 和 rightTask），并通过 invokeAll() 方法并行执行。
结果合并：
- 使用 merge() 方法将两个有序子数组合并为一个有序数组。
线程池执行：
- 使用 ForkJoinPool 提交任务，并通过 invoke() 方法触发任务执行。