• 极客专栏正式上线!欢迎访问 https://www.jikewenku.com/topic.html
  • 极客专栏正式上线!欢迎访问 https://www.jikewenku.com/topic.html

Java7任务并行执行神器:Fork&Join框架

技术杂谈 勤劳的小蚂蚁 3个月前 (01-26) 89次浏览 已收录 0个评论 扫描二维码

Fork/Join是什么?

Fork/Join框架是Java7提供的并行执行任务框架,思想是将大任务分解成小任务,然后小任务又可以继续分解,然后每个小任务分别计算出结果再合并起来,最后将汇总的结果作为大任务结果。其思想和MapReduce的思想非常类似。对于任务的分割,要求各个子任务之间相互独立,能够并行独立地执行任务,互相之间不影响。
Fork/Join的运行流程图如下:
我们可以通过Fork/Join单词字面上的意思去理解这个框架。Fork是叉子分叉的意思,即将大任务分解成并行的小任务,Join是连接结合的意思,即将所有并行的小任务的执行结果汇总起来。

工作窃取算法

ForkJoin采用了工作窃取(work-stealing)算法,若一个工作线程的任务队列为空没有任务执行时,便从其他工作线程中获取任务主动执行。为了实现工作窃取,在工作线程中维护了双端队列,窃取任务线程从队尾获取任务,被窃取任务线程从队头获取任务。这种机制充分利用线程进行并行计算,减少了线程竞争。但是当队列中只存在一个任务了时,两个线程去取反而会造成资源浪费。
工作窃取的运行流程图如下:

Fork/Join核心类

Fork/Join框架主要由子任务、任务调度两部分组成,类层次图如下。
  • ForkJoinPool
ForkJoinPool是ForkJoin框架中的任务调度器,和ThreadPoolExecutor一样实现了自己的线程池,提供了三种调度子任务的方法:
  1. execute:异步执行指定任务,无返回结果;
  2. invoke、invokeAll:异步执行指定任务,等待完成才返回结果;
  3. submit:异步执行指定任务,并立即返回一个Future对象;
  • ForkJoinTask
Fork/Join框架中的实际的执行任务类,有以下两种实现,一般继承这两种实现类即可。
  1. RecursiveAction:用于无结果返回的子任务;
  2. RecursiveTask:用于有结果返回的子任务;

Fork/Join框架实战

下面实现一个Fork/Join小例子,从1+2+…10亿,每个任务只能处理1000个数相加,超过1000个的自动分解成小任务并行处理;并展示了通过不使用Fork/Join和使用时的时间损耗对比。
  1. import java.util.concurrent.ForkJoinPool;
  2. import java.util.concurrent.RecursiveTask;
  3. publicclassForkJoinTaskextendsRecursiveTask<Long> {
  4.    privatestaticfinallong MAX = 1000000000L;
  5.    privatestaticfinallong THRESHOLD = 1000L;
  6.    privatelong start;
  7.    privatelongend;
  8.    publicForkJoinTask(long start, longend) {
  9.        this.start = start;
  10.        this.end = end;
  11.    }
  12.    publicstaticvoid main(String[] args) {
  13.        test();
  14.        System.out.println("--------------------");
  15.        testForkJoin();
  16.    }
  17.    privatestaticvoid test() {
  18.        System.out.println("test");
  19.        long start = System.currentTimeMillis();
  20.        Long sum = 0L;
  21.        for (long i = 0L; i <= MAX; i++) {
  22.            sum += i;
  23.        }
  24.        System.out.println(sum);
  25.        System.out.println(System.currentTimeMillis() - start + "ms");
  26.    }
  27.    privatestaticvoid testForkJoin() {
  28.        System.out.println("testForkJoin");
  29.        long start = System.currentTimeMillis();
  30.        ForkJoinPool forkJoinPool = newForkJoinPool();
  31.        Long sum = forkJoinPool.invoke(newForkJoinTask(1, MAX));
  32.        System.out.println(sum);
  33.        System.out.println(System.currentTimeMillis() - start + "ms");
  34.    }
  35.    @Override
  36.    protectedLong compute() {
  37.        long sum = 0;
  38.        if (end - start <= THRESHOLD) {
  39.            for (long i = start; i <= end; i++) {
  40.                sum += i;
  41.            }
  42.            return sum;
  43.        } else {
  44.            long mid = (start + end) / 2;
  45.            ForkJoinTask task1 = newForkJoinTask(start, mid);
  46.            task1.fork();
  47.            ForkJoinTask task2 = newForkJoinTask(mid + 1, end);
  48.            task2.fork();
  49.            return task1.join() + task2.join();
  50.        }
  51.    }
  52. }
这里需要计算结果,所以任务继承的是RecursiveTask类。ForkJoinTask需要实现compute方法,在这个方法里首先需要判断任务是否小于等于阈值1000,如果是就直接执行任务。否则分割成两个子任务,每个子任务在调用fork方法时,又会进入compute方法,看看当前子任务是否需要继续分割成孙任务,如果不需要继续分割,则执行当前子任务并返回结果。使用join方法会阻塞并等待子任务执行完并得到其结果。
程序输出:
  1. test
  2. 500000000500000000
  3. 4992ms
  4. --------------------
  5. testForkJoin
  6. 500000000500000000
  7. 508ms
从结果看出,并行的时间损耗明显要少于串行的,这就是并行任务的好处。
尽管如此,在使用Fork/Join时也得注意,不要盲目使用。
  1. 如果任务拆解的很深,系统内的线程数量堆积,导致系统性能性能严重下降;
  2. 如果函数的调用栈很深,会导致栈内存溢出;


丨极客文库, 版权所有丨如未注明 , 均为原创丨
本网站采用知识共享署名-非商业性使用-相同方式共享 3.0 中国大陆许可协议进行授权
转载请注明原文链接:Java7任务并行执行神器:Fork&Join框架
喜欢 (0)
[247507792@qq.com]
分享 (0)
勤劳的小蚂蚁
关于作者:
温馨提示:本文来源于网络,转载文章皆标明了出处,如果您发现侵权文章,请及时向站长反馈删除。

您必须 登录 才能发表评论!

  • 精品技术教程
  • 编程资源分享
  • 问答交流社区
  • 极客文库知识库

客服QQ


QQ:2248886839


工作时间:09:00-23:00