并发(十五)、Fork/Join框架基本使用(转)

概述
ava.util.concurrent.ForkJoinPool由Java大师Doug Lea主持编写，它可以将一个大的任务拆分成多个子任务进行并行处理，最后将子任务结果合并成最后的计算结果，并进行输出。本文中对Fork/Join框架的讲解，基于JDK1.8+中的Fork/Join框架实现，参考的Fork/Join框架主要源代码也基于JDK1.8+。

这几篇文章将试图解释Fork/Join框架的知识点，以便对自己、对各位读者在并发程序的设计思路上进行一些启发。文章将首先讲解Fork/Join框架的基本使用，以及其中需要注意的使用要点；接着使用Fork/Join框架解决一些实际问题；最后再讲解Fork/Join框架的工作原理

Fork/Join框架基本使用
这里是一个简单的Fork/Join框架使用示例，在这个示例中我们计算了1-1001累加后的值：

/**
* 这是一个简单的Join/Fork计算过程，将1—1001数字相加
*/
public class TestForkJoinPool {
 private static final Integer MAX = 200;
 static class MyForkJoinTask extends RecursiveTask<Integer> {
     // 子任务开始计算的值
     private Integer startValue;
     // 子任务结束计算的值
     private Integer endValue;
     public MyForkJoinTask(Integer startValue , Integer endValue) {
         this.startValue = startValue;
         this.endValue = endValue;
     }
     @Override
     protected Integer compute() {
         // 如果条件成立，说明这个任务所需要计算的数值分为足够小了
         // 可以正式进行累加计算了
         if(endValue - startValue < MAX) {
             System.out.println("开始计算的部分：startValue = " + startValue + ";endValue = " + endValue);
             Integer totalValue = 0;
             for(int index = this.startValue ; index <= this.endValue  ; index++) {
                 totalValue += index;
             }
             return totalValue;
         }
         // 否则再进行任务拆分，拆分成两个任务
         else {
             MyForkJoinTask subTask1 = new MyForkJoinTask(startValue, (startValue + endValue) / 2);
             subTask1.fork();
             MyForkJoinTask subTask2 = new MyForkJoinTask((startValue + endValue) / 2 + 1 , endValue);
             subTask2.fork();
             return subTask1.join() + subTask2.join();
         }
     }
 }
 public static void main(String[] args) {
     // 这是Fork/Join框架的线程池
     ForkJoinPool pool = new ForkJoinPool();
     ForkJoinTask<Integer> taskFuture =  pool.submit(new MyForkJoinTask(1,1001));
     try {
         Integer result = taskFuture.get();
         System.out.println("result = " + result);
     } catch (InterruptedException | ExecutionException e) {
         e.printStackTrace(System.out);
     }
 }
}

以上代码很简单，在关键的位置有相关的注释说明。这里本文再对以上示例中的要点进行说明。首先看看以上示例代码的可能执行结果：
```
开始计算的部分：startValue = 1;endValue = 126
开始计算的部分：startValue = 127;endValue = 251
开始计算的部分：startValue = 252;endValue = 376
开始计算的部分：startValue = 377;endValue = 501
开始计算的部分：startValue = 502;endValue = 626
开始计算的部分：startValue = 627;endValue = 751
开始计算的部分：startValue = 752;endValue = 876
开始计算的部分：startValue = 877;endValue = 1001

result = 501501

2-1. 工作顺序图
下图展示了以上代码的工作过程概要，但实际上Fork/Join框架的内部工作过程要比这张图复杂得多，例如如何决定某一个recursive task是使用哪条线程进行运行；再例如如何决定当一个任务/子任务提交到Fork/Join框架内部后，是创建一个新的线程去运行还是让它进行队列等待。
所以如果不深入理解Fork/Join框架的运行原理，只是根据之上最简单的使用例子观察运行效果，那么我们只能知道子任务在Fork/Join框架中被拆分得足够小后，并且其内部使用多线程并行完成这些小任务的计算后再进行结果向上的合并动作，最终形成顶层结果。不急，一步一步来，我们先从这张概要的过程图开始讨论。
![](/fileupload/images/202105/1621222229807.png) 
图中最顶层的任务使用submit方式被提交到Fork/Join框架中，后者将前者放入到某个线程中运行，工作任务中的compute方法的代码开始对这个任务T1进行分析。如果当前任务需要累加的数字范围过大（代码中设定的是大于200），则将这个计算任务拆分成两个子任务（T1.1和T1.2），每个子任务各自负责计算一半的数据累加，请参见代码中的fork方法。如果当前子任务中需要累加的数字范围足够小（小于等于200），就进行累加然后返回到上层任务中。
2-2. ForkJoinPool构造函数
ForkJoinPool有四个构造函数，其中参数最全的那个构造函数如下所示：

public ForkJoinPool(int parallelism,
ForkJoinWorkerThreadFactory factory,
UncaughtExceptionHandler handler,
boolean asyncMode)
```
parallelism：可并行级别，Fork/Join框架将依据这个并行级别的设定，决定框架内并行执行的线程数量。并行的每一个任务都会有一个线程进行处理，但是千万不要将这个属性理解成Fork/Join框架中最多存在的线程数量，也不要将这个属性和ThreadPoolExecutor线程池中的corePoolSize、maximumPoolSize属性进行比较，因为ForkJoinPool的组织结构和工作方式与后者完全不一样。而后续的讨论中，读者还可以发现Fork/Join框架中可存在的线程数量和这个参数值的关系并不是绝对的关联（有依据但并不全由它决定）。

factory：当Fork/Join框架创建一个新的线程时，同样会用到线程创建工厂。只不过这个线程工厂不再需要实现ThreadFactory接口，而是需要实现ForkJoinWorkerThreadFactory接口。后者是一个函数式接口，只需要实现一个名叫newThread的方法。在Fork/Join框架中有一个默认的ForkJoinWorkerThreadFactory接口实现：DefaultForkJoinWorkerThreadFactory。

handler：异常捕获处理器。当执行的任务中出现异常，并从任务中被抛出时，就会被handler捕获。

asyncMode：这个参数也非常重要，从字面意思来看是指的异步模式，它并不是说Fork/Join框架是采用同步模式还是采用异步模式工作。Fork/Join框架中为每一个独立工作的线程准备了对应的待执行任务队列，这个任务队列是使用数组进行组合的双向队列。即是说存在于队列中的待执行任务，即可以使用先进先出的工作模式，也可以使用后进先出的工作模式。