概述

Oracle官方对Fork/Join框架的定义原文如下

The fork/join framework is an implementation of the ExecutorService interface that helps you take advantage of multiple processors. It is designed for work that can be broken into smaller pieces recursively. The goal is to use all the available processing power to enhance the performance of your application. As with any ExecutorService implementation, the fork/join framework distributes tasks to worker threads in a thread pool. The fork/join framework is distinct because it uses a work-stealing algorithm. Worker threads that run out of things to do can steal tasks from other threads that are still busy. The center of the fork/join framework is the ForkJoinPool class, an extension of the AbstractExecutorService class. ForkJoinPool implements the core work-stealing algorithm and can execute ForkJoinTask processes.

大致意思是，Fork/Join框架是一个实现了ExecutorService接口的多线程处理器。它可以将一个大任务通过递归分解成多个小任务并发执行，以此提高任务执行的效率。Fork/Join框架和ExecutorService一样需要将任务交给线程池来执行，但不同的是Fork/Join框架使用了工作窃取算法，即工作用尽的工作线程可以从其他仍很忙的线程中窃取任务。

fork / join框架的中心是ForkJoinPool类，它是AbstractExecutorService类的扩展。 ForkJoinPool实现了核心的工作窃取算法，并且可以执行ForkJoinTask进程。

思想

分治编程

分治算法 ：把一个复杂的问题分成两个或更多的相同或相似的子问题，再把子问题分成更小的子问题……直到最后子问题可以简单的直接求解，原问题的解即子问题的解的合并。

利用分治法思想的编程方式称为分治编程，而Fork/Join框架就是使用了分治编程，通过设定阈值，对大任务分解（Fork）成小任务，再对每个小任务并发执行，将结果合并（Join）获得最终结果，提高任务执行的效率和CPU的使用率。

Fork/Join - 图1

工作窃取算法

工作窃取(work-stealing)算法是指某个线程从其他队列里窃取任务来执行。

那么为什么需要使用工作窃取算法呢？
当我们需要执行一个大任务时，我们使用分治将其分解成若干个小任务并发执行，为了减少线程间的竞争，使用不同的队列对应不同的线程，每个线程执行自己队列里的任务。
当有的线程已经完成自己队列里的任务时，该线程便空闲了，此时可以让它去帮其他线程，让它窃取其他线程队列里的任务来执行。
而在这时它们会访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。

Fork/Join - 图2

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

基本类和接口

Fork/Join框架主要有以下重要的类

  • ForkJoinPool
  • ForkJoinTask
  • RecursiveAction
  • RecursiveTask
  • ForkJoinWorkerThread

ForkJoinPool

ForkJoinPool类继承了AbstractExecutorService，该类是一个线程池，是Fork/Join框架的核心，保证了工作窃取算法和ForkJoinTask的正常工作。

image.png

  1. public class ForkJoinPool extends AbstractExecutorService {
  2.     //执行ForkJoinTask任务，无返回值，不阻塞
  3.     public void execute(ForkJoinTask<?> task);
  4.     //执行ForkJoinTask任务，有任务返回值，阻塞
  5.     public <T> T invoke(ForkJoinTask<T> task);
  6.     //执行ForkJoinTask任务，有任务返回值，不阻塞
  7.     public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task);
  9. }

ForkJoinTask

ForkJoinTask是一个抽象任务类，其提供了两个主要方法fork()和join()。Fork（分解），Join（合并），依赖这两操作实现分治编程。
其有两个主要子类，一个是RecursiveTask，一个是RecursiveAction。
在使用时我们通常继承其子类，并实现其compute()方法来实现自己的分治逻辑。

  1. public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
  2.      //在当前任务正在运行的池中异步执行此任务
  3.     public final ForkJoinTask<V> fork();
  4.     //返回计算结果
  5.     public final V join();
  7. }
  1. public abstract class RecursiveTask<V> extends ForkJoinTask<V> {
  2.     //完成任务，有返回值
  3.     protected abstract V compute();
  5. }
  1. public abstract class RecursiveAction extends ForkJoinTask<Void> {
  2.     //完成任务，无返回值
  3.     protected abstract void compute();
  5. }

对于两个子类，它们有以下区别

  • RecursiveTask是有返回值的任务，RecursiveAction是无返回值的任务
  • RecursiveTask通常用于只fork不join的情形，RecursiveAction通常用于fork+join的情形
  1. /**
  2.  * @author Chavy
  3.  * @date 2020/4/19 19:26
  4.  */
  5. public class ForkJoinDemo {
  7.     /**
  8.      * 10w大小的数组
  9.      */
  10.     static int[] nums = new int[100000];
  12.     static final int MAX_NUM = 5000;
  14.     static Random random = new Random();
  16.     static {
  17.         for(int i = 0; i < nums.length; i++){
  18.             nums[i] = random.nextInt(100);
  19.         }
  20.         System.out.println("sum == " + Arrays.stream(nums).sum());
  21.     }
  23.     static class AddTask extends RecursiveAction{
  24.         int start, end;
  26.         AddTask(int start, int end){
  27.             this.start = start;
  28.             this.end = end;
  29.         }
  31.         @Override
  32.         protected void compute() {
  33.             if(end - start <= MAX_NUM){
  34.                 long sum = 0;
  35.                 for(int i = start; i < end; i++){
  36.                     sum += nums[i];
  37.                 }
  38.                 System.out.println("From " + start + " to " + end + " = " + sum);
  39.             }else {
  40.                 int middle = start + (end - start) / 2;
  42.                 AddTask addTask1 = new AddTask(start, middle);
  43.                 AddTask addTask2 = new AddTask(middle, end);
  44.                 addTask1.fork();
  45.                 addTask2.fork();
  46.             }
  47.         }
  48.     }
  50.     static class AddTaskRec extends RecursiveTask<Long> {
  52.         int start, end;
  54.         AddTaskRec(int start, int end){
  55.             this.start = start;
  56.             this.end = end;
  57.         }
  59.         @Override
  60.         protected Long compute() {
  61.             if(end - start <= MAX_NUM){
  62.                 long sum = 0;
  63.                 for(int i = start; i < end; i++){
  64.                     sum += nums[i];
  65.                 }
  66.                 return sum;
  67.             }
  68.             int middle = start + (end - start) / 2;
  70.             AddTaskRec addTaskRec1 = new AddTaskRec(start, middle);
  71.             AddTaskRec addTaskRec2 = new AddTaskRec(middle, end);
  72.             addTaskRec1.fork();
  73.             addTaskRec2.fork();
  74.             try{
  75.                 return addTaskRec1.fork().get() + addTaskRec2.fork().get();
  76.             }catch (ExecutionException  | InterruptedException e){
  77.                 System.out.println("Error");
  78.             }
  79.             return null;
  80.         }
  81.     }
  83.     public static void main(String[] args) {
  84.         ForkJoinDemo forkJoinDemo = new ForkJoinDemo();
  86.         ForkJoinPool forkJoinPool = new ForkJoinPool();
  87.         AddTask addTask = new AddTask(0, nums.length);
  88. //        forkJoinPool.execute(addTask);
  89. //        addTask.join();
  90.         AddTaskRec addTaskRec = new AddTaskRec(0, nums.length);
  91.         forkJoinPool.execute(addTaskRec);
  92.         System.out.println(addTaskRec.join());
  93.     }
  94. }

上面就是一个例子，要计算一个数组内的数的总和，使用两个不同的任务子类来实现该Fork/Join的逻辑。

ForkJoinWorkerThread

ForkJoinWorkerThread线程是一种在Fork/Join框架中运行的特性线程，它除了具有普通线程的特性外，最主要的特点是每一个ForkJoinWorkerThread线程都具有一个独立的任务等待队列（work queue），这个任务队列用于存储在本线程中被拆分的若干子任务。

  1. public class ForkJoinWorkerThread extends Thread {
  3.     final ForkJoinPool pool;                // the pool this thread works in
  4.     final ForkJoinPool.WorkQueue workQueue; // work-stealing mechanics
  6. }

使用

这是一个使用Fork/Join框架求数组最大值的例子：

  1. import java.util.concurrent.*;
  3. public class ForkJoinTest extends RecursiveTask<Integer> {
  5.     //阈值设定为2
  6.     private int threshold = 2;
  8.     private int[] array;
  10.     private int index0 = 0;
  11.     private int index1 = 1;
  13.     public ForkJoinTest(int[] array, int index0, int index1){
  14.         this.array = array;
  15.         this.index0 = index0;
  16.         this.index1 = index1;
  17.     }
  19.     @Override
  20.     protected Integer compute() {
  21.         //最大值的结果
  22.         int max = Integer.MIN_VALUE;
  23.         //分治法划分子任务逻辑
  24.         //如果子任务已达阈值，则直接操作
  25.         if ((index1 - index0) <= threshold) {
  27.             for (int i = index0;i <= index1; i ++) {
  28.                 max = Math.max(max, array[i]);
  29.             }
  31.         } else {
  32.             //fork/join
  33.             //划分子任务
  34.             int mid = index0 + (index1 - index0) / 2;
  35.             ForkJoinTest lMax = new ForkJoinTest(array, index0, mid);
  36.             ForkJoinTest rMax = new ForkJoinTest(array, mid + 1, index1);
  37.             //将子任务在当前运行池中异步执行
  38.             lMax.fork();
  39.             rMax.fork();
  40.             //获取返回结果
  41.             int lm = lMax.join();
  42.             int rm = rMax.join();
  44.             max = Math.max(lm, rm);
  46.         }
  48.         return max;
  49.     }
  51.     public static void main(String ... args) throws ExecutionException, InterruptedException, TimeoutException {
  52.         //创建ForkJoinPool线程池
  53.         ForkJoinPool pool = new ForkJoinPool();
  55.         int[] array = {100,400,200,90,80,300,600,10,20,-10,30,2000,1000};
  57.         ForkJoinTest task = new ForkJoinTest(array, 0, array.length - 1);
  58.         //将任务提交到线程池里
  59.         Future<Integer> future = pool.submit(task);
  61.         System.out.println("Result:" + future.get(1, TimeUnit.SECONDS));
  63.     }
  64. }

从上面的例子里可以看出几点使用步骤：

  • 先创建一个ForkJoinTask类，按情况继承RecursiveAction还是RecursiveTask，并实现其compute()方法
  • compute方法里实现分治逻辑，主要使用if-else结构，规定阈值，达到阈值则执行，否则划分子任务，并将子任务fork()异步执行（若有结果返回，使用join获取结果）
  • 创建ForkJoinPool线程池，提交任务给线程池并执行
  • 获取处理Future对象结果

原理

属性

  1. @sun.misc.Contended
  2. public class ForkJoinPool extends AbstractExecutorService {
  4.     private static final int  RSLOCK     = 1;
  5.     private static final int  RSIGNAL    = 1 << 1;
  6.     private static final int  STARTED    = 1 << 2;
  7.     private static final int  STOP       = 1 << 29;
  8.     private static final int  TERMINATED = 1 << 30;
  9.     private static final int  SHUTDOWN   = 1 << 31;
  11.     volatile long ctl;                   // 控制中心，重要
  12.     volatile int runState;               // 运行状态，如上，SHUWDOWN是负数，其他是2的次方
  13.     final int config;                    // 配置：并行度（parallelism），二进制的低16位代表
  14.     int indexSeed;                       // 生成worker的queue索引
  15.     volatile WorkQueue[] workQueues;     // 主要保存工作队列的数组
  16.     final ForkJoinWorkerThreadFactory factory;
  17.     final UncaughtExceptionHandler ueh;  // per-worker UEH
  18.     final String workerNamePrefix;       // to create worker name string
  19.     volatile AtomicLong stealCounter;    // also used as sync monitor
  21.     /**
  22.      * 主要构造器
  23.      * parallelism
  24.      */
  25.     private ForkJoinPool(int parallelism,
  26.                          ForkJoinWorkerThreadFactory factory,
  27.                          UncaughtExceptionHandler handler,
  28.                          int mode,
  29.                          String workerNamePrefix) {
  30.         this.workerNamePrefix = workerNamePrefix;
  31.         this.factory = factory;
  32.         this.ueh = handler;
  33.         this.config = (parallelism & SMASK) | mode;
  34.         long np = (long)(-parallelism); // offset ctl counts
  35.         this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
  36.     }
  37. }

WorkQueue

  1.     @sun.misc.Contended
  2.     static final class WorkQueue {
  4.         //初始容量 2的13次方  32M
  5.         static final int INITIAL_QUEUE_CAPACITY = 1 << 13;
  6.         //最大容量 2的26次方  64M
  7.         static final int MAXIMUM_QUEUE_CAPACITY = 1 << 26; // 64M
  9.         // Instance fields
  10.         volatile int scanState;    // versioned, <0: inactive; odd:scanning
  11.         int stackPred;             // pool stack (ctl) predecessor
  12.         int nsteals;               // 窃取的任务数
  13.         int hint;                  // 一个随机数，用来帮助任务窃取，在 helpXXXX()的方法中会用到
  14.         int config;                // 配置：二进制的低16位代表 在 queue[] 中的索引，
  15.          // 高16位：mode可选FIFO_QUEUE（1 << 16）和LIFO_QUEUE（1 << 31），默认是LIFO_QUEUE
  16.         volatile int qlock;        // 锁定标示位：1: locked, < 0: terminate; else 0
  17.         volatile int base;         // index of next slot for poll
  18.         int top;                   // index of next slot for push
  19.         ForkJoinTask<?>[] array;   // ForkJoinTask的任务数组
  20.         final ForkJoinPool pool;   // 保存的线程池
  21.         final ForkJoinWorkerThread owner; // owning thread or null if shared
  22.         volatile Thread parker;    // == owner during call to park; else null
  23.         volatile ForkJoinTask<?> currentJoin;  // task being joined in awaitJoin
  24.         volatile ForkJoinTask<?> currentSteal; // mainly used by helpStealer
  26.         WorkQueue(ForkJoinPool pool, ForkJoinWorkerThread owner) {
  27.             this.pool = pool;
  28.             this.owner = owner;
  29.             // Place indices in the center of array (that is not yet allocated)
  30.             base = top = INITIAL_QUEUE_CAPACITY >>> 1;
  31.         }
  33.     }

比较重要和核心的就是上面这个内部类 WorkQueue，它是一个数组实现的队列。它对应的外部传入的一个任务，内部使用了一个数组来保存自己Fork出来的子任务。另外 WorkQueue 里面也保存了从其他线程里窃取来的任务数量和任务本身，这就是上面提到的工作窃取算法第一次出现的地方了。
该队列支持三种形式的列操作：push、pop、poll（也叫steal），push和pop只能被队列内部持有的线程调用，poll可被其他线程偷取任务时调用。

  1. final void push(ForkJoinTask<?> task) {
  2.     ForkJoinTask<?>[] a; ForkJoinPool p;
  3.     int b = base, s = top, n;
  4.     //非空才操作，否则不管你
  5.     if ((a = array) != null) {    // ignore if queue removed
  6.         int m = a.length - 1;     // fenced write for task visibility
  7.         //把任务添加进去，然后队顶加1
  8.         U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task);
  9.         U.putOrderedInt(this, QTOP, s + 1);
  10.         //如果队列内的元素只剩一件或更少，那就叫线程池干活
  11.         if ((n = s - b) <= 1) {
  12.             if ((p = pool) != null)
  13.                 p.signalWork(p.workQueues, this);
  14.         }
  15.         //如果元素大于数组长度，扩容
  16.         else if (n >= m)
  17.             growArray();
  18.     }
  19. }
  21. //出队
  22. final ForkJoinTask<?> pop() {
  23.     ForkJoinTask<?>[] a; ForkJoinTask<?> t; int m;
  24.     //如果队列不空，循环将Top位值CAS操作置为空，成功便返回null值。
  25.     if ((a = array) != null && (m = a.length - 1) >= 0) {
  26.         for (int s; (s = top - 1) - base >= 0;) {
  27.             long j = ((m & s) << ASHIFT) + ABASE;
  28.             if ((t = (ForkJoinTask<?>)U.getObject(a, j)) == null)
  29.                 break;
  30.             if (U.compareAndSwapObject(a, j, t, null)) {
  31.                 U.putOrderedInt(this, QTOP, s);
  32.                 return t;
  33.             }
  34.         }
  35.     }
  36.     return null;
  37. }
  39. //窃取出队
  40. final ForkJoinTask<?> poll() {
  41.     ForkJoinTask<?>[] a; int b; ForkJoinTask<?> t;
  42.     //判断队列还有东西，数组不为空
  43.     while ((b = base) - top < 0 && (a = array) != null) {
  44.         int j = (((a.length - 1) & b) << ASHIFT) + ABASE;
  45.         t = (ForkJoinTask<?>)U.getObjectVolatile(a, j);
  46.         //base还没被改变，且任务不为空，就CAS置null，拿走任务，基底加一
  47.         if (base == b) {
  48.             if (t != null) {
  49.                 if (U.compareAndSwapObject(a, j, t, null)) {
  50.                     base = b + 1;
  51.                     return t;
  52.                 }
  53.             }
  54.             //否则现在可能空咯，不做了
  55.             else if (b + 1 == top) // now empty
  56.                 break;
  57.         }
  58.     }
  59.     return null;
  60. }

由于 WorkQueue 使用了 CAS 引用操作 workerQueue 数组，所以在队列 base 和 top 不必使用标志位了。

对索引的更新保证了当 top==base 时代表这个队列为空，但是如果在push、pop或poll没有完全完成的情况下，可能出现即使 base==top 但队列为非空的错误情况（isEmpty方法可以检查“移除最后一个元素操作”部分完成的情况）。所以，单独考虑 poll 操作，它并不是 wait-free 的(无等待算法)。在一个线程正在偷取任务时，另外一个线程是无法完成偷取操作的。大体上讲，我们起码有一定概率保证了阻塞性。如果一个偷取操作失败，偷取线程会选择另外一个随机目标继续尝试。所以，为了保证偷取线程能够正常执行，它必须能够满足任何正在执行的对 queue 的 poll 或 push 操作都能完成（这就是为什么我们通常使用 pollAt 方法和它的变体，在已知的 base 索引中先尝试一次，然后再考虑可替代的操作，而不直接使用可以重试的 poll 方法）。

XorShifts算法

不懂，这个网上看到CV下来的 https://www.jianshu.com/p/32a15ef2f1bf

在上面提到了有对poll操作的变体的操作方法。然后我们回来这个队列，我们可以看到 workQueue 是一个 LIFO 队列，更像是栈这种数据结构。而这种方法也适用于FIFO。

FIFO 和 LIFO 这两种模式都不会考虑共用性、加载、缓存地址等，所以很少能在给定的机器上提供最好的性能，但通过对这些因素进行平均，可以提供良好的吞吐量。更进一步来讲，即使我们尝试使用这些信息，也没有能利用它的基础。例如，一些任务集从缓存共用中获取到良好的性能收益，但其他任务集会因此受到它的影响。另外，虽然队列中提供了扫描功能，但是从长远看来为了吞吐量通常最好使用随机选择，而非直接选择。所以，在ForkJoinPool 中我们也就使用了一种 XorShifts（一种随机算法，有些带有不同的偏移量） 随机算法。

WorkQueue 对于提交到池中的任务也使用类似的随机插入方式。我们不能把这些被工作线程使用的任务混合同一个队列中，所以，我们会使用一种随机哈希算法（有点类似ConcurrentHashMap的随机算法）将工作队列与工作线程关联起来。ThreadLocalRandom的probe(探针值)会为选中的已存在的队列提供一个哈希值，在与其他提交任务的线程（submitters）竞争时，也可以利用probe来随机移位。实际上，submitters 就像工作线程（worker）一样，只不过他们被限制只能执行它们提交的本地任务（CountedCompleter类型任务也不能执行）。在共享模式里提交的任务需要锁来控制（在扩容情况下提供保护），我们使用了一个简单的自旋锁(qlock字段)，因为 submitters 遇到一个繁忙队列时会继续尝试提交其他队列或创建新的队列-只有在 submitters 创建和注册新队列时阻塞。另外，”qlock”会在 shutdown 时饱和到不可锁定值(-1)，但是解锁操作依然可以执行。

externalPush()

现在就从头开始，像上面的demo一样，当我们调用了 ForkJoinPool 的一个 execute() 方法来执行我们的任务的时候，会发生什么呢？

  1. /**
  2.  * 只是一个判空操作，实际操作在externalPush() 方法里
  3.  */
  4. public void execute(ForkJoinTask<?> task) {
  5.     if (task == null)
  6.         throw new NullPointerException();
  7.     externalPush(task);
  8. }

execute() 实际上只是做了个判空操作，真正的执行逻辑在 externalPush() 方法里。

  1. final void externalPush(ForkJoinTask<?> task) {
  2.     WorkQueue[] ws; WorkQueue q; int m;
  3.     //计算一个哈希值，获取线程池状态
  4.     int r = ThreadLocalRandom.getProbe();
  5.     int rs = runState;
  6.     //获取当前WorkQueue数组，判断数组已经被初始化且已有任务，并且按m & r & SQMASK上的元素不为空，
  7.     //线程池正常运行，获取的随机数也不是0，且CAS上锁成功
  8.     if ((ws = workQueues) != null && (m = (ws.length - 1)) >= 0 &&
  9.         (q = ws[m & r & SQMASK]) != null && r != 0 && rs > 0 &&
  10.         U.compareAndSwapInt(q, QLOCK, 0, 1)) {
  11.         ForkJoinTask<?>[] a; int am, n, s;
  12.         //获取全部任务列表，判断是否为空，且数组长度大于队列长度
  13.         if ((a = q.array) != null &&
  14.             (am = a.length - 1) > (n = (s = q.top) - q.base)) {
  15.             int j = ((am & s) << ASHIFT) + ABASE;
  16.             //把该任务放在队列的top端
  17.             U.putOrderedObject(a, j, task);
  18.             //然后给Top向上移
  19.             U.putOrderedInt(q, QTOP, s + 1);
  20.             U.putIntVolatile(q, QLOCK, 0);
  21.             //如果队列小于1，提醒开始干活了
  22.             if (n <= 1)
  23.                 signalWork(ws, q);
  24.             return;
  25.         }
  26.         //CAS解锁
  27.         U.compareAndSwapInt(q, QLOCK, 1, 0);
  28.     }
  29.     //提交任务
  30.     externalSubmit(task);
  31. }