首页 / JAVA / Java Fork/Join 框架

Java Fork/Join 框架

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了Java Fork/Join 框架，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含8875字，纯文字阅读大概需要13分钟。

内容图文

简介

从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()

技术分享图片

技术分享图片

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

图形化处理过程

下面盗两张图

技术分享图片

技术分享图片

 

技术分享图片

技术分享图片

使用示例

批量发送消息 

             1
            package
             com.cjs.boot.demo;

             2
             3
            import
             java.util.ArrayList;

             4
            import
             java.util.List;

             5
            import
             java.util.concurrent.ForkJoinPool;

             6
            import
             java.util.concurrent.RecursiveAction;

             7
            import
             java.util.concurrent.TimeUnit;

             8
             9
            public
            class
             ForkJoinPoolDemo {

            10
            11
            class SendMsgTask extends RecursiveAction {
1213privatefinalint THRESHOLD = 10;
1415privateint start;
16privateint end;
17private List<String> list;
1819public SendMsgTask(int start, int end, List<String> list) {
20this.start = start;
21this.end = end;
22this.list = list;
23        }
2425        @Override
26protectedvoid compute() {
2728if ((end - start) <= THRESHOLD) {
29for (int i = start; i < end; i++) {
30                     System.out.println(Thread.currentThread().getName() + ": " + list.get(i));
31                }
32             }else {
33int middle = (start + end) / 2;
34                 invokeAll(new SendMsgTask(start, middle, list), new SendMsgTask(middle, end, list));
35            }
3637        }
3839    }
4041publicstaticvoid main(String[] args) throws InterruptedException {
42         List<String> list = new ArrayList<>();
43for (int i = 0; i < 123; i++) {
44             list.add(String.valueOf(i+1));
45        }
4647         ForkJoinPool pool = new ForkJoinPool();
48         pool.submit(new ForkJoinPoolDemo().new SendMsgTask(0, list.size(), list));
49         pool.awaitTermination(10, TimeUnit.SECONDS);
50        pool.shutdown();
51    }
5253 }

求和 

             1
            package
             com.cjs.boot.demo;

             2
             3
            import
             java.util.concurrent.ExecutionException;

             4
            import
             java.util.concurrent.ForkJoinPool;

             5
            import
             java.util.concurrent.ForkJoinTask;

             6
            import
             java.util.concurrent.RecursiveTask;

             7
             8
            public
            class
             ForkJoinTaskDemo {

             9
            10
            private
            class SumTask extends RecursiveTask<Integer> {
1112privatestaticfinalint THRESHOLD = 20;
1314privateint arr[];
15privateint start;
16privateint end;
1718public SumTask(int[] arr, int start, int end) {
19this.arr = arr;
20this.start = start;
21this.end = end;
22        }
2324/**25         * 小计
26*/27private Integer subtotal() {
28             Integer sum = 0;
29for (int i = start; i < end; i++) {
30                 sum += arr[i];
31            }
32             System.out.println(Thread.currentThread().getName() + ": ∑(" + start + "~" + end + ")=" + sum);
33return sum;
34        }
3536        @Override
37protected Integer compute() {
3839if ((end - start) <= THRESHOLD) {
40return subtotal();
41             }else {
42int middle = (start + end) / 2;
43                 SumTask left = new SumTask(arr, start, middle);
44                 SumTask right = new SumTask(arr, middle, end);
45                left.fork();
46                right.fork();
4748return left.join() + right.join();
49            }
50        }
51    }
5253publicstaticvoid main(String[] args) throws ExecutionException, InterruptedException {
54int[] arr = newint[100];
55for (int i = 0; i < 100; i++) {
56             arr[i] = i + 1;
57        }
5859         ForkJoinPool pool = new ForkJoinPool();
60         ForkJoinTask<Integer> result = pool.submit(new ForkJoinTaskDemo().new SumTask(arr, 0, arr.length));
61         System.out.println("最终计算结果: " + result.invoke());
62        pool.shutdown();
63    }
6465 }
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文档中的两个示例 

             1
            package
             com.cjs.boot.demo;

             2
             3
            import
             java.util.Arrays;

             4
            import java.util.concurrent.*;
 5 6publicclass RecursiveActionDemo {
 7 8privatestaticclass SortTask extends RecursiveAction {
 910staticfinalint THRESHOLD = 100;
1112finallong[] array;
13finalint lo, hi;
1415public SortTask(long[] array, int lo, int hi) {
16this.array = array;
17this.lo = lo;
18this.hi = hi;
19        }
2021public SortTask(long[] array) {
22this(array, 0, array.length);
23        }
2425publicvoid sortSequentially(int lo, int hi) {
26            Arrays.sort(array, lo, hi);
27        }
2829publicvoid merge(int lo, int mid, int hi) {
30long[] buf = Arrays.copyOfRange(array, lo, mid);
31for (int i = 0, j = lo, k = mid; i < buf.length; j++) {
32                 array[j] = (k == hi || buf[i] < array[k]) ? buf[i++] : array[k++];
33            }
34        }
3536        @Override
37protectedvoid compute() {
38if (hi - lo < THRESHOLD) {
39                sortSequentially(lo, hi);
40             }else {
41int mid = (lo + hi) >>> 1;
42                 invokeAll(new SortTask(array, lo, mid), new SortTask(array, mid, hi));
43                merge(lo, mid, hi);
44            }
45        }
46    }
4748publicstaticvoid main(String[] args) throws ExecutionException, InterruptedException {
49long[] array = newlong[120];
50for (int i = 0; i < array.length; i++) {
51             array[i] = (long) (Math.random() * 1000);
52        }
53        System.out.println(Arrays.toString(array));
5455         ForkJoinPool pool = new ForkJoinPool();
56         pool.submit(new SortTask(array));
57         pool.awaitTermination(5, TimeUnit.SECONDS);
58        pool.shutdown();
5960    }
6162 }
             1
            package
             com.cjs.boot.demo;

             2
             3
            import java.util.concurrent.*;
 4 5publicclass RecursiveTaskDemo {
 6 7privatestaticclass Fibonacci extends RecursiveTask<Integer> {
 8 9finalint n;
1011public Fibonacci(int n) {
12this.n = n;
13        }
1415        @Override
16protected Integer compute() {
17if (n <= 1) {
18return n;
19             }else {
20                 Fibonacci f1 = new Fibonacci(n - 1);
21                f1.fork();
22                 Fibonacci f2 = new Fibonacci(n - 1);
23return f2.compute() + f1.join();
24            }
25        }
26    }
2728publicstaticvoid main(String[] args) throws InterruptedException, ExecutionException {
29         ForkJoinPool pool = new ForkJoinPool();
30         Future<Integer> future = pool.submit(new Fibonacci(10));
31        System.out.println(future.get());
32        pool.shutdown();
33    }
3435 }

 

参考

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

 

原文：https://www.cnblogs.com/cjsblog/p/9078341.html

内容总结

以上是互联网集市为您收集整理的Java Fork/Join 框架全部内容，希望文章能够帮你解决Java Fork/Join 框架所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

扫描二维码推送至手机访问。

本文链接：https://qyyshop.com/info/1164412.html

来源：【匿名】