Java Fork/Join 框架

简介

从JDK1.7开始，Java提供Fork/Join框架用于并行执行任务，它的思想就是讲一个大任务分割成若干小任务，最终汇总每个小任务的结果得到这个大任务的结果。

这种思想和MapReduce很像（input --> split --> map --> reduce --> output）

主要有两步：

• 第一、任务切分；
• 第二、结果合并

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

工作窃取（work-stealing）

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

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

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

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

API介绍

ForkJoinPool

An ExecutorService for running ForkJoinTasks.

A ForkJoinPool differs from other kinds of ExecutorService mainly by virtue of employing work-stealing: all threads in the pool attempt to find and execute tasks submitted to the pool and/or created by other active tasks (eventually blocking waiting for work if none exist). This enables efficient processing when most tasks spawn other subtasks (as do most ForkJoinTasks), as well as when many small tasks are submitted to the pool from external clients. Especially when setting asyncMode to true in constructors, ForkJoinPools may also be appropriate for use with event-style tasks that are never joined.

ForkJoinPool与其它的ExecutorService区别主要在于它使用“工作窃取”：线程池中的所有线程都企图找到并执行提交给线程池的任务。当大量的任务产生子任务的时候，或者同时当有许多小任务被提交到线程池中的时候，这种处理是非常高效的。特别的，当在构造方法中设置asyncMode为true的时候这种处理更加高效。

ForkJoinTask

ForkJoinTask代表运行在ForkJoinPool中的任务。

主要方法：

• fork()    在当前线程运行的线程池中安排一个异步执行。简单的理解就是再创建一个子任务。
• join()    当任务完成的时候返回计算结果。
• invoke()    开始执行任务，如果必要，等待计算完成。

子类：

• RecursiveAction    一个递归无结果的ForkJoinTask（没有返回值）
• RecursiveTask    一个递归有结果的ForkJoinTask（有返回值）

ForkJoinWorkerThread

A thread managed by a ForkJoinPool, which executes ForkJoinTasks.

ForkJoinWorkerThread代表ForkJoinPool线程池中的一个执行任务的线程。

类图

    

代码分析

接下来，简略的看一下关键代码来加深对Fork/Join的理解。

ForkJoinPool

WorkQueue是一个ForkJoinPool中的内部类，它是线程池中线程的工作队列的一个封装，支持任务窃取。

什么叫线程的任务窃取呢？就是说你和你的一个伙伴一起吃水果，你的那份吃完了，他那份没吃完，那你就偷偷的拿了他的一些水果吃了。存在执行2个任务的子线程，这里要讲成存在A,B两个个WorkQueue在执行任务，A的任务执行完了，B的任务没执行完，那么A的WorkQueue就从B的WorkQueue的ForkJoinTask数组中拿走了一部分尾部的任务来执行，可以合理的提高运行和计算效率。

submit()

可以看到：

1. 同样是提交任务，submit会返回ForkJoinTask，而execute不会
2. 任务提交给线程池以后，会将这个任务加入到当前提交者的任务队列中。

前面我们说过，每个线程都有一个WorkQueue，而WorkQueue中有执行任务的线程（ForkJoinWorkerThread owner），还有这个线程需要处理的任务（ForkJoinTask<?>[] array）。那么这个新提交的任务就是加到array中。

ForkJoinWorkerThread

从代码中我们可以清楚地看到，ForkJoinWorkThread持有ForkJoinPool和ForkJoinPool.WorkQueue的引用，以表明该线程属于哪个线程池，它的工作队列是哪个

ForkJoinTask

fork()

可以看到，如果是ForkJoinWorkerThread运行过程中fork()，则直接加入到它的工作队列中，否则，重新提交任务。

join()和invoke()

可以看到它们都会等待计算完成

图形化处理过程

下面盗两张图

使用示例

批量发送消息

 package com.cjs.boot.demo;

 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.RecursiveAction;
 import java.util.concurrent.TimeUnit;

 public class ForkJoinPoolDemo {

     class SendMsgTask extends RecursiveAction {

         private final int THRESHOLD = 10;

         private int start;
         private int end;
         private List<String> list;

         public SendMsgTask(int start, int end, List<String> list) {
             this.start = start;
             this.end = end;
             this.list = list;
         }

         @Override
         protected void compute() {

             if ((end - start) <= THRESHOLD) {
                 for (int i = start; i < end; i++) {
                     System.out.println(Thread.currentThread().getName() + ": " + list.get(i));
                 }
             }else {
                 int middle = (start + end) / 2;
                 invokeAll(new SendMsgTask(start, middle, list), new SendMsgTask(middle, end, list));
             }

         }

     }

     public static void main(String[] args) throws InterruptedException {
         List<String> list = new ArrayList<>();
         for (int i = 0; i < 123; i++) {
             list.add(String.valueOf(i+1));
         }

         ForkJoinPool pool = new ForkJoinPool();
         pool.submit(new ForkJoinPoolDemo().new SendMsgTask(0, list.size(), list));
         pool.awaitTermination(10, TimeUnit.SECONDS);
         pool.shutdown();
     }

 }

求和

 package com.cjs.boot.demo;

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

 public class ForkJoinTaskDemo {

     private class SumTask extends RecursiveTask<Integer> {

         private static final int THRESHOLD = 20;

         private int arr[];
         private int start;
         private int end;

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

         /**
          * 小计
          */
         private Integer subtotal() {
             Integer sum = 0;
             for (int i = start; i < end; i++) {
                 sum += arr[i];
             }
             System.out.println(Thread.currentThread().getName() + ": ∑(" + start + "~" + end + ")=" + sum);
             return sum;
         }

         @Override
         protected Integer compute() {

             if ((end - start) <= THRESHOLD) {
                 return subtotal();
             }else {
                 int middle = (start + end) / 2;
                 SumTask left = new SumTask(arr, start, middle);
                 SumTask right = new SumTask(arr, middle, end);
                 left.fork();
                 right.fork();

                 return left.join() + right.join();
             }
         }
     }

     public static void main(String[] args) throws ExecutionException, InterruptedException {
         int[] arr = new int[100];
         for (int i = 0; i < 100; i++) {
             arr[i] = i + 1;
         }

         ForkJoinPool pool = new ForkJoinPool();
         ForkJoinTask<Integer> result = pool.submit(new ForkJoinTaskDemo().new SumTask(arr, 0, arr.length));
         System.out.println("最终计算结果: " + result.invoke());
         pool.shutdown();
     }

 }

ForkJoinPool.commonPool-worker-2: ∑(50~62)=678
ForkJoinPool.commonPool-worker-2: ∑(62~75)=897
ForkJoinPool.commonPool-worker-2: ∑(75~87)=978
ForkJoinPool.commonPool-worker-2: ∑(87~100)=1222
ForkJoinPool-1-worker-1: ∑(0~12)=78
ForkJoinPool-1-worker-1: ∑(12~25)=247
ForkJoinPool-1-worker-1: ∑(25~37)=378
ForkJoinPool-1-worker-1: ∑(37~50)=572
ForkJoinPool-1-worker-2: ∑(75~87)=978
ForkJoinPool-1-worker-3: ∑(50~62)=678
ForkJoinPool-1-worker-5: ∑(62~75)=897
ForkJoinPool.commonPool-worker-7: ∑(0~12)=78
ForkJoinPool.commonPool-worker-3: ∑(37~50)=572
ForkJoinPool-1-worker-4: ∑(87~100)=1222
ForkJoinPool.commonPool-worker-2: ∑(25~37)=378
ForkJoinPool.commonPool-worker-5: ∑(12~25)=247
最终计算结果: 5050

api文档中的两个示例

 package com.cjs.boot.demo;

 import java.util.Arrays;
 import java.util.concurrent.*;

 public class RecursiveActionDemo {

     private static class SortTask extends RecursiveAction {

         static final int THRESHOLD = 100;

         final long[] array;
         final int lo, hi;

         public SortTask(long[] array, int lo, int hi) {
             this.array = array;
             this.lo = lo;
             this.hi = hi;
         }

         public SortTask(long[] array) {
             this(array, 0, array.length);
         }

         public void sortSequentially(int lo, int hi) {
             Arrays.sort(array, lo, hi);
         }

         public void merge(int lo, int mid, int hi) {
             long[] buf = Arrays.copyOfRange(array, lo, mid);
             for (int i = 0, j = lo, k = mid; i < buf.length; j++) {
                 array[j] = (k == hi || buf[i] < array[k]) ? buf[i++] : array[k++];
             }
         }

         @Override
         protected void compute() {
             if (hi - lo < THRESHOLD) {
                 sortSequentially(lo, hi);
             }else {
                 int mid = (lo + hi) >>> 1;
                 invokeAll(new SortTask(array, lo, mid), new SortTask(array, mid, hi));
                 merge(lo, mid, hi);
             }
         }
     }

     public static void main(String[] args) throws ExecutionException, InterruptedException {
         long[] array = new long[120];
         for (int i = 0; i < array.length; i++) {
             array[i] = (long) (Math.random() * 1000);
         }
         System.out.println(Arrays.toString(array));

         ForkJoinPool pool = new ForkJoinPool();
         pool.submit(new SortTask(array));
         pool.awaitTermination(5, TimeUnit.SECONDS);
         pool.shutdown();

     }

 }

 package com.cjs.boot.demo;

 import java.util.concurrent.*;

 public class RecursiveTaskDemo {

     private static class Fibonacci extends RecursiveTask<Integer> {

         final int n;

         public Fibonacci(int n) {
             this.n = n;
         }

         @Override
         protected Integer compute() {
             if (n <= 1) {
                 return n;
             }else {
                 Fibonacci f1 = new Fibonacci(n - 1);
                 f1.fork();
                 Fibonacci f2 = new Fibonacci(n - 1);
                 return f2.compute() + f1.join();
             }
         }
     }

     public static void main(String[] args) throws InterruptedException, ExecutionException {
         ForkJoinPool pool = new ForkJoinPool();
         Future<Integer> future = pool.submit(new Fibonacci(10));
         System.out.println(future.get());
         pool.shutdown();
     }

 }

参考

http://gee.cs.oswego.edu/dl/papers/fj.pdf

http://ifeve.com/talk-concurrency-forkjoin/

https://www.cnblogs.com/senlinyang/p/7885964.html

https://blog.csdn.net/u012403290/article/details/70917810