java8学习：ForkJoin

内容来自《 java8实战 》，本篇文章内容均为非盈利，旨为方便自己查询、总结备份、开源分享。如有侵权请告知，马上删除。
书籍购买地址：java8实战

• 这篇是接上一篇并行数据处理与性能余下的问题：forkjoin进行讲解的
• forkjoin的目的就是以递归的方式来拆分更小的任务，然后将每个小任务处理后的结果在合并，fork就是拆分，join就是合并

• RecursiveTask

  • 要把任务提交到这个池，就必须创建 RecursiveTask的子类，R就是你需要返回的结果类型，如果不返回结果就使用RecursiveAction类型，继承这个抽象类只需要实现一个方法

  protected abstract V compute();

  • 此方法定以了将任务拆分成子任务的逻辑，以及无法再才分或不方便拆分时生成单个子任务结果的逻辑，(它定义了咋拆分又定义了怎么把子任务聚合)

• 先实现一下，再来说原理

  • 还是实现1到一千万的累加和

  public class ForkJoinImpl extends java.util.concurrent.RecursiveTask<Long> { //临界值，就是结束值减开始值的结果如果小于这个值那么就不拆分了，大于这个值才会拆分 private final int MEDIAN_NUM = 100000; //从多少计算 private int start_num = 0; //计算到多少 private int end_num = 0; //构造 public ForkJoinImpl(int start_num, int end_num) { this.start_num = start_num; this.end_num = end_num; } @Override protected Long compute() { //结束值减开始值的结果 int temp = end_num - start_num; //判断结束值减开始值的结果是否小于上面定义的临界值 if (temp <= MEDIAN_NUM){ //如果小的话，那么就不进行拆分了，就直接调用方法开始计算 return sequentiallySum(); } //到这就代表结束值减开始值的结果是大于临界值的 //继续进行拆分 //start_num到start_num + temp / 2是把数据的左半部分形成一个新的task //比如0到10，那么就是 10-0=10，temp=10，start_num=0,所以形成的新task就是(0,10/2=5),也就是左半部分 ForkJoinImpl leftTask = new ForkJoinImpl(start_num,start_num + temp / 2); //利用ForkJoinPool中的线程异步执行新创建的子任务 leftTask.fork(); //这创建的就是数据的后半段，start_num + temp / 2 = 0+10/2 = 6,所以形成的新task就是(0+10/2=6,10),也就是右半部分 ForkJoinImpl rightTask = new ForkJoinImpl(start_num + temp / 2,end_num); //同时执行第二个子任务，有可能允许进一步划分 Long rightResult = rightTask.compute(); //读取第一个子任务的结果，如果没有完成就等待 Long leftResult = leftTask.join(); //该任务的结果是两个子任务结果的组合 return rightResult + leftResult; } //计算方法：在不能进行拆分的时候进行计算 private Long sequentiallySum(){ long sum = 0; for (int i = start_num; i <= end_num; i++) { sum += i; } return sum; } } @Test public void test() throws Exception { ForkJoinImpl forkJoin = new ForkJoinImpl(0, 10000000); Long invoke = new ForkJoinPool().invoke(forkJoin); System.out.println("invoke = " + invoke); }

  • 上面的流程的总结：当把ForkJoinImpl对象传给ForkJoinPool时，这个任务就由池中的一个线程执行，这个线程会调用任务的compute方法，该方法会检查任务是够孝道足以顺序执行，也就是我们上面定义的临界值，如果不够小就会要求再次拆分数据，并分给一个新的ForkJoinImpl，新的ForkJoinImpl也是由pool安排执行。因此这个过程是递归重复的，把缘任务拆分为更小的任务。这时候达到临街值要求后，会顺序计算每个任务的结果，然后由分支过程创建的任务二叉树遍历回到它的根。接下来会合并每个子任务的部分结果，从而得到总任务的结果。如下面的图

• 使用此框架的最佳做法

  • 对一个任务嗲用join方法会阻塞调用方，直到该任务做出结果，因此有必要在两个子任务的计算都开始之后在调用它，否则，你得到的版本会比原始的顺序算法更慢更复杂，因为每个子任务都必须去等待另一个任务完成才能启动
  • 对子任务调用fork方法可以把他排进pool，同时对左边和右边的子任务调用它似乎很自然，但这样做的效率比直接对其中一个调用compute低，这样做可以以为其中一个子任务重用同一个线程，从而避免在线程池中多分配一个任务造成的开销。比如有些写

  //执行子任务 left.fork(); right.fork(); //获取子任务结果 int lResult = left.join(); int rResult = right.join(); 而更高效的写法是 leftTask.fork(); ForkJoinImpl rightTask = new ForkJoinImpl(start_num + temp / 2,end_num); //自己理解的是它一直判断是否有大于临界值，如果有就左半部分进行加入pool，而右边一直去判断是否还有大于临界值的情况 //而不至于两方都在等待，并且上面的写法需要left返回后右边才会返回。如果理解的不对请评论指正 Long rightResult = rightTask.compute(); Long leftResult = leftTask.join();

• 工作窃取

  • 对于上面会产生很多的小任务然后分配到内核上进行计算，理想情况下是所有的小任务机会全部同时执行完毕，但是由于线程是随意切换的，这种情况会发生改变：可能A已经空闲了，但是B还是忙的要死，为了解决这样的一个问题，此框架就有工作窃取的技术了
  • 在应用中，这意味着这些任务拆不多被平均分配到pool中的所有的线程上。但每个线程都为分配给他的任务保存一个双向队列，每完成一个任务，就会从队列头上取出下一个任务开始执行，当A已经忙完，B还在忙的时候，A并不会闲下来，而是随机选一个别的线程，从队列的尾巴上偷走一个任务，这个过程一直继续下去，直到所有的任务执行完毕，所有队列都情况，这就是为什么不划分为几个大任务而是很多小任务，这样有助于更好的在工作线程之间平衡负载