浅谈Java Fork/Join并行框架

初步了解Fork/Join框架

Fork/Join 框架是java7中加入的一个并行任务框架，可以将任务分割成足够小的小任务，然后让不同的线程来做这些分割出来的小事情，然后完成之后再进行join，将小任务的结果组装成大任务的结果。下面的图片展示了这种框架的工作模型：

使用Fork/Join并行框架的前提是我们的任务可以拆分成足够小的任务，而且可以根据小任务的结果来组装出大任务的结果，一个最简单的例子是使用Fork/Join框架来求一个数组中的最大/最小值，这个任务就可以拆成很多小任务，大任务就是寻找一个大数组中的最大/最小值，我们可以将一个大数组拆成很多小数组，然后分别求解每个小数组中的最大/最小值，然后根据这些任务的结果组装出最后的最大最小值，下面的代码展示了如何通过Fork/Join求解数组的最大值：

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
* Created by hujian06 on 2017/9/28.
*
* fork/join demo
*/
public class ForkJoinDemo {

 /**
  * how to find the max number in array by Fork/Join
  */
 private static class MaxNumber extends RecursiveTask<Integer> {

   private int threshold = 2;

   private int[] array; // the data array

   private int index0 = 0;
   private int index1 = 0;

   public MaxNumber(int[] array, int index0, int index1) {
     this.array = array;
     this.index0 = index0;
     this.index1 = index1;
   }

   @Override
   protected Integer compute() {
     int max = Integer.MIN_VALUE;
     if ((index1 - index0) <= threshold) {

       for (int i = index0;i <= index1; i ++) {
         max = Math.max(max, array[i]);
       }

     } else {
       //fork/join
       int mid = index0 + (index1 - index0) / 2;
       MaxNumber lMax = new MaxNumber(array, index0, mid);
       MaxNumber rMax = new MaxNumber(array, mid + 1, index1);

       lMax.fork();
       rMax.fork();

       int lm = lMax.join();
       int rm = rMax.join();

       max = Math.max(lm, rm);

     }

     return max;
   }
 }

 public static void main(String ... args) throws ExecutionException, InterruptedException, TimeoutException {

   ForkJoinPool pool = new ForkJoinPool();

   int[] array = {100,400,200,90,80,300,600,10,20,-10,30,2000,1000};

   MaxNumber task = new MaxNumber(array, 0, array.length - 1);

   Future<Integer> future = pool.submit(task);

   System.out.println("Result:" + future.get(1, TimeUnit.SECONDS));

 }

}

可以通过设置不同的阈值来拆分成小任务，阈值越小代表拆出来的小任务越多。

工作窃取算法

Fork/Join在实现上，大任务拆分出来的小任务会被分发到不同的队列里面，每一个队列都会用一个线程来消费，这是为了获取任务时的多线程竞争，但是某些线程会提前消费完自己的队列。而有些线程没有及时消费完队列，这个时候，完成了任务的线程就会去窃取那些没有消费完成的线程的任务队列，为了减少线程竞争，Fork/Join使用双端队列来存取小任务，分配给这个队列的线程会一直从头取得一个任务然后执行，而窃取线程总是从队列的尾端拉取task。

Frok/Join框架的实现细节

在上面的示例代码中，我们发现Fork/Join的任务是通过ForkJoinPool来执行的，所以框架的一个核心是任务的fork和join，然后就是这个ForkJoinPool。关于任务的fork和join，我们可以想象，而且也是由我们的代码自己控制的，所以要分析Fork/Join，那么ForkJoinPool最值得研究。

上面的图片展示了ForkJoinPool的类关系图，可以看到本质上它就是一个Executor。在ForkJoinPool里面，有两个特别重要的成员如下：

  volatile WorkQueue[] workQueues; 
  final ForkJoinWorkerThreadFactory factory;

workQueues 用于保存向ForkJoinPool提交的任务，而具体的执行有ForkJoinWorkerThread执行，而ForkJoinWorkerThreadFactory可以用于生产出ForkJoinWorkerThread。可以看一些ForkJoinWorkerThread，可以发现每一个ForkJoinWorkerThread会有一个pool和一个workQueue，和我们上面描述的是一致的，每个线程都被分配了一个任务队列，而执行这个任务队列的线程由pool提供。

下面我们看一下当我们fork的时候发生了什么：

  public final ForkJoinTask<V> fork() {
    Thread t;
    if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
      ((ForkJoinWorkerThread)t).workQueue.push(this);
    else
      ForkJoinPool.common.externalPush(this);
    return this;
  }

看上面的fork代码，可以看到首先取到了当前线程，然后判断是否是我们的ForkJoinPool专用线程，如果是，则强制类型转换（向下转换）成ForkJoinWorkerThread，然后将任务push到这个线程负责的队列里面去。如果当前线程不是ForkJoinWorkerThread类型的线程，那么就会走else之后的逻辑，大概的意思是首先尝试将任务提交给当前线程，如果不成功，则使用例外的处理方法，关于底层实现较为复杂，和我们使用Fork/Join关系也不太大，如果希望搞明白具体原理，可以看源码。

下面看一下join的流程：

 public final V join() {
    int s;
    if ((s = doJoin() & DONE_MASK) != NORMAL)
      reportException(s);
    return getRawResult();
  }
  
    private int doJoin() {
    int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
    return (s = status) < 0 ? s :
      ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
      (w = (wt = (ForkJoinWorkerThread)t).workQueue).
      tryUnpush(this) && (s = doExec()) < 0 ? s :
      wt.pool.awaitJoin(w, this, 0L) :
      externalAwaitDone();
  }
  
    final int doExec() {
    int s; boolean completed;
    if ((s = status) >= 0) {
      try {
        completed = exec();
      } catch (Throwable rex) {
        return setExceptionalCompletion(rex);
      }
      if (completed)
        s = setCompletion(NORMAL);
    }
    return s;
  }
  
   /**
   * Implements execution conventions for RecursiveTask.
   */
  protected final boolean exec() {
    result = compute();
    return true;
  }

上面展示了主要的调用链路，我们发现最后落到了我们在代码里编写的compute方法，也就是执行它，所以，我们需要知道的一点是，fork仅仅是分割任务，只有当我们执行join的时候，我们的额任务才会被执行。

如何使用Fork/Join并行框架

前文首先展示了一个求数组中最大值得例子，然后介绍了“工作窃取算法”，然后分析了Fork/Join框架的一些细节，下面才是我们最关心的，怎么使用Fork/Join框架呢？

为了使用Fork/Join框架，我们只需要继承类RecursiveTask或者RecursiveAction。前者适用于有返回值的场景，而后者适合于没有返回值的场景。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。

# java  # 并行框架  # 并行计算框架 

相关文章： 如何通过免费商城建站系统源码自定义网站主题与功能？  如何在阿里云域名上完成建站全流程？  C#怎么创建控制台应用 C# Console App项目创建方法  如何通过万网虚拟主机快速搭建网站？  如何通过VPS建站无需域名直接访问？  广州网站制作的公司,现在专门做网站的公司有没有哪几家是比较好的，性价比高，模板也多的？  Python lxml的etree和ElementTree有什么区别  制作网站的公司有哪些,做一个公司网站要多少钱？  广州网站建站公司选择指南：建站流程与SEO优化关键词解析  如何通过虚拟主机空间快速建站？  c++怎么编写动态链接库dll_c++ __declspec(dllexport)导出与调用【方法】  如何用花生壳三步快速搭建专属网站？  如何通过wdcp面板快速创建网站？  如何通过多用户协作模板快速搭建高效企业网站？  个人网站制作流程图片大全,个人网站如何注销？  如何在云服务器上快速搭建个人网站？  如何在宝塔面板中创建新站点？  深圳网站制作培训,深圳哪些招聘网站比较好？  实现虚拟支付需哪些建站技术支撑？  制作网站的基本流程,设计网站的软件是什么？  在线流程图制作网站手机版,谁能推荐几个好的CG原画资源网站么？  香港代理服务器配置指南：高匿IP选择、跨境加速与SEO优化技巧  企业宣传片制作网站有哪些,传媒公司怎么找企业宣传片项目？  建站之星如何取消后台验证码生成？  如何在万网自助建站平台快速创建网站？  宝塔建站教程：一键部署配置流程与SEO优化实战指南  如何快速搭建高效WAP手机网站？  微网站制作教程,我微信里的网站怎么才能复制到浏览器里？  ,南京靠谱的征婚网站？  如何在香港服务器上快速搭建免备案网站？  Android滚轮选择时间控件使用详解  建站中国必看指南：CMS建站系统+手机网站搭建核心技巧解析  淘宝制作网站有哪些,淘宝网官网主页？  北京企业网站设计制作公司,北京铁路集团官方网站？  焦点电影公司作品,电影焦点结局是什么？  建站之星导航如何优化提升用户体验？  制作电商网页,电商供应链怎么做？  如何通过FTP服务器快速搭建网站？  如何零基础开发自助建站系统？完整教程解析  中山网站制作网页,中山新生登记系统登记流程？  历史网站制作软件,华为如何找回被删除的网站？  建站主机选购指南：核心配置与性价比推荐解析  建站之星3.0如何解决常见操作问题？  相亲简历制作网站推荐大全,新相亲大会主持人小萍萍资料？  简历在线制作网站免费,免费下载个人简历的网站是哪些？  jQuery 常见小例汇总  如何在橙子建站中快速调整背景颜色？  香港网站服务器数量如何影响SEO优化效果？  ,交易猫的商品怎么发布到网站上去？  在线制作视频网站免费,都有哪些好的动漫网站？