深入理解ForkJoin

任务类型

线程池执行的任务可以分为两种：CPU密集型任务和IO密集型任务。在实际的业务场景中，我们需要根据任务的类型来选择对应的策略，最终达到充分并合理地使用CPU和内存等资源，最大限度地提高程序性能的目的。

CPU密集型任务

CPU密集型任务也称为计算密集型任务，包括加密、解密、压缩和计算等一系列需要大量耗费CPU资源的任务。对于CPU密集型的任务，并不是设置越多的线程性能越高，其最佳核心数是逻辑CPU核心数的1～2倍。因为对于计算任务较重的任务，CPU的每个核心基本都是满负荷的，设置更多的线程不仅不会提升性能，反而由于线程之间对CPU资源的争抢造成不必要的上下文切换导致性能下降。

IO密集型任务

IO密集型任务的特点是不会特别消耗CPU资源，但IO操作比较耗时，导致总体占用较多时间，数据库读写、文件读写、网络通信等任务都属于IO密集型任务。对于IO密集型任务，一般会将最大线程数设置为CPU核心数很多倍。IO读写速度相比于CPU计算的速度要慢很多，如果设置的线程数较少，线程可能都在处理IO操作，导致CPU资源的浪费。如果设置较多的线程数，当一部分线程在等待IO的时候，这部分线程不需要CPU资源，其他的线程就可以利用CPU资源去执行其他任务。

线程数计算方法

可以通过下面的公式计算出一个合理的线程数量。

线程数 = CPU核心数 × （1 + 任务平均等待时间/CPU平均工作时间）

可以看出线程数与任务平均等待时间成正比，任务的平均等待时间越长，线程数就越多；与CPU平均工作时间成反比，CPU平均工作时间越长，线程数就越少。例如CPU密集型任务的特点是CPU平均工作时间较长，而任务的平均等待时间较短，因此此类型所需的线程数较少；IO密集型任务的特点是任务平均等待时间较长，CPU平均工作时间较短，因此此类型所需的线程数较多。

分治算法

分治算法的基本思想是将一个规模较大的问题分解为多个规模较小的子问题，这些子问题之间相互独立且与原问题的性质相同。将所有子问题的解求出来，原问题的解也就求出来了。

分治算法的步骤如下：

分解：将要解决的问题划分成若干规模较小的同类子问题；
求解：当子问题划分得足够小时，可以用很简单的方式计算出子问题的解；
合并：将子问题的解逐层合并，最终得到原问题的解。

以上过程可以表示如下图：

ForkJoin框架

传统的线程池ThreadPoolExecutor有以下两个缺点：

无法对大任务进行拆分，即只能由单个线程去完成某项任务；
工作线程从队列中获取任务时存在竞争。

为了解决以上问题，JDK1.7引入了ForkJoin框架。ForkJoin框架允许其他线程向其提交任务，并将任务拆分成粒度更细的子任务，这些子任务由ForkJoin框架内部的工作线程来并行执行，并且这些工作线程之间可以互相窃取任务执行。

主要API

ForkJoin框架主要包含两部分：

分治任务的线程池ForkJoinPool类；
分治任务ForkJoinTask类。

ForkJoinPool

ForkJoinPool是用于执行ForkJoinTask任务的执行池，继承了AbstractExecutorService类。ForkJoinPool的构造函数有多个，此处我们介绍其中参数最全的一个，其实现如下：

public ForkJoinPool(int parallelism,
                    ForkJoinWorkerThreadFactory factory,
                    UncaughtExceptionHandler handler,
                    boolean asyncMode) {
    this(checkParallelism(parallelism),
         checkFactory(factory),
         handler,
         asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
         "ForkJoinPool-" + nextPoolId() + "-worker-");
    checkPermission();
}

parallelism：表示指令的并行级别，ForkJoinPool将根据这个值来决定工作线程的数量，默认使用Runtime.getRuntime().availableProcessors()来设置；
factory：ForkJoinPool内部创建线程使用的线程工厂，需要注意的是，此处的线程工厂的类型是ForkJoinWorkerThreadFactory而不是ThreadFactory，默认使用DefaultForkJoinWorkerThreadFactory；
handler：异常处理器，主要用于处理任务运行中出现的异常；
asyncMode：队列的工作模式，为true时使用先进先出模式，为false时使用先进后出模式。

ForkJoinPool提交任务的方法主要有以下几种：

execute()：在提交任务后不会返回结果，支持ForkJoinTask类型和Runnable两种类型的任务；

public void execute(ForkJoinTask<?> task);
public void execute(Runnable task);

invoke()：在任务执行结束后返回泛型执行结果，支持ForkJoinTask类型的任务；

public <T> T invoke(ForkJoinTask<T> task);

submit()：在提交任务后返回ForkJoinTask类型的结果，如果任务不能按计划执行则抛出任务拒绝异常，支持ForkJoinTask类型、Callable类型和Runnable类型的任务。

public <T> ForkJoinTask<T> submit(Callable<T> task);
public <T> ForkJoinTask<T> submit(Runnable task, T result);
public ForkJoinTask<?> submit(Runnable task);

ForkJoinTask

ForkJoinTask是ForkJoinPoll的核心之一，是任务的实际载体，定义了任务执行时的具体逻辑和任务拆分逻辑。ForkJoinTask是一个抽象类，实现了Future接口，因此也可以将其看成轻量级的Future。

ForkJoinTask的核心方法主要有以下两个：

fork()：用于向当前任务所运行的线程池中提交任务，如果当前线程是ForkJoinWorkThread类型（即ForkJoinPool内部工作线程），则将任务放入该线程的工作队列，否则放入common线程池的工作队列；
join()：用于获取任务的执行结果，调用此方法时将阻塞当前线程直到对应的子任务完成运行并返回结果。

ForkJoinTask提供了以下三个子类，在实际使用时可以根据需要继承这三个子类即可。

RecursiveAction：用于递归执行但不需要返回结果的任务；
RecursiveTask：用于递归执行且需要返回结果的任务；
CountedCompleter<T>：在任务完成后会触发执行一个自定义的钩子方法。

WorkQueue

WorkQueue是ForkJoinPool的静态内部类，主要用来表示一个工作队列，是基于一个ForkJoinTask类型的数组实现的双端队列。WorkQueue的几个较重要的成员变量如下：

//在对该WorkQueue操作时搭配CAS加锁
volatile int qlock;
//数组的底端，工作窃取时从此处取出任务执行
volatile int base;
//任务数组，用于保存当前工作队列的所有任务
ForkJoinTask<?>[] array;
//当前工作队列的ForkJoinWorkerThread类型的工作线程，如果是外部提交生成的工作队列这个属性为null
final ForkJoinWorkerThread owner;

ForkJoinPool内部维护了一个WorkQueue工作队列数组，当我们调用invoke()或submit()方法提交一个任务时，ForkJoinPool根据某个路由规则将该任务放到一个工作队列中，如果任务在执行过程中会创建出子任务，则子任务会提交到工作线程对应的工作队列中。

工作队列WorkQueue是ForkJoinPool类的一个内部类，是一个双端队列。如果是ForkJoinPool内部划分的任务，在WorkQueue数组中的下标是奇数位；如果是外部提交的任务，在WorkQueue数组中的下标是偶数位。

每个工作线程在处理自己的工作队列时采用的是FILO先进后出的方式，即在运行中产生新的任务时，会将其放到工作队列的尾端；在需要执行新的任务时，同样在工作队列的尾端取任务。工作线程处理自己的工作队列的同时，还可以窃取其他工作线程的任务，为了降低冲突的发生，窃取的位置在工作队列的首端。

ForkJoinWorkerThread

ForkJoinWorkerThread是ForkJoinPool中用于执行任务的线程，并且是ForkJoinPool内部专门为执行划分的子任务创建的工作线程。每个ForkJoinWorkerThread都有其对应的一个WorkQueue，这个关系由WorkQueue的一个owner属性来维护，并且这种WorkQueue在ForkJoinPool的WorkQueue数组中的下标一定是奇数位。