一丶 Fork/Join 框架产生背景:
随着并发需求的不断提高和硬件的不断发展, 程序并行执行仿佛就提上日程上来了, 伟大的毛主席就说过:"人多力量大", 所以如果一件事可以分配给多个人同时去做, 到最后再把完成的事情组合到一起去, 那么做事情的效率就会大大提升. 用下面的这张图啦感受一下:
一件大事被分为五个处理器来处理(当然分开的这些任务没有依赖性), 最终完成合并做成一件大事!
到这里, 大家应该对这个框架有一个简单的认识了吧!
Fork/Join 框架是 Java1.7 开始提供的一个并行执行框架, 结合上面的讲述, 再结合 Fork/Join 的字面意思我们可以知道, 先 Fork(分叉), 再结合!
二丶工作窃取算法
想必只要提到 Fork/Join 框架, 都要提到工作窃取算法, 工作窃取是指在分别完成分配的事情时, 如果工作结束早的线程可以窃取 (帮助) 其他线程来完成工作, 最终达到总的任务快速完成的目的, 就像老板布置一样任务, 项目经理将这个任务分给甲乙丙三个程序员来完成这样事情, 甲比较厉害, 或者分配的事情比较少, 他完成以后他主动帮助乙来完成工作.
如果这样子来完成任务, 那么怎么分配任务就是一个问题了:
首先将任务分为互不依赖的子任务, 将这些子任务放到不同的队列里, 并安排一个线程来处理这些任务, 线程和队列 (一般使用双端队列) 一一对应, 当有线程完成时, 就会去其他队列里窃取任务, 但是不同的是, 窃取任务是从队列的尾端窃取, 防止窃取过程中发生线程竞争.
优点: 充分利用并行计算, 提高运算效率.
缺点: 在某些情况下还是会发生竞争, 例如当队列中只剩下一个任务, 两个线程进行竞争.
三丶 Fork/Join 框架的设计思想
理解了 Fork/Join 框架的设计思想后, 我们理解这个框架就会变的非常容易, 我们用几行伪代码来直观的告诉你什么是 Fork/Join 思想:
if(任务足够小){
进行计算;
}else{
将任务分为两个部分;
结合两个子任务;
}
四丶 JDK 实现 Fork/Join 框架
(1)JDk 为 Fork/Join 框架提供了很好的支持, 我们想要用这个算法首先得创建一个 Fork/Join 任务, 在 JDK 中这个任务就叫做: ForJoinTask, 只要继承这个类就可以创建一个任务类, 但是实际使用中并不是直接继承 ForkJoinTask 类, 而是继承它的子类, 它有两个子类, 分别是 RecursiveAction 和 RecursiveTask, 它们之间的区别是是否返回任务结果, 前者用于没有返回结果的任务, 后者用于有返回结果的任务.
(2)有了 Fork/Join 任务后还需要执行任务, JDK 提供了 ForkJoinPool 来执行.
(3)下面我们来看一个用 ForkJoin 框架来进行累加计算例子:
- // 继承 RecursiveTask, 用于与结果返回的任务
- public class CountTask extends RecursiveTask<Integer>{
- // 阈值为 2, 当小于等于这个值的时候进行计算
- private static final int THRESHOLD = 2;
- // 累加的起始值
- private int start;
- private int end;
- public CountTask(int start,int end){
- this.start = start;
- this.end = end;
- }
- protected Integer compute() {
- int sum = 0;
- boolean canCompute = (end-start) <= THRESHOLD;
- // 如果 false, 表示 end-start 大于 2, 还要进行分割
- if(canCompute) {
- for(int i = start;i<=end; i++ ) {
- sum += i;
- }
- }
- else {
- int mid = (end + start) / 2;
- // 分为两个任务
- CountTask leftTask = new CountTask(start,mid);
- CountTask rightTask = new CountTask(mid + 1,end);
- // 执行子任务
- leftTask.fork();
- rightTask.fork();
- // 等待子任务完成
- int leftResult = leftTask.join();
- int rightResult = rightTask.join();
- // 合并子任务
- sum = leftResult+rightResult;
- }
- return sum;
- }
- public static void main(String[] args) {
- // 创建 ForkJoinPool 对象执行任务
- ForkJoinPool forkJoinPool = new ForkJoinPool();
- // 计算 1-5 累加
- CountTask task = new CountTask(1,5);
- // 执行
- Future<Integer> result = forkJoinPool.submit(task);
- try {
- System.out.println(result.get());
- } catch (InterruptedException e) {
- e.printStackTrace();
- } catch (ExecutionException e) {
- e.printStackTrace();
- }
- }
- }
五丶 ForkJoin 框架的实现原理
在看到例子后我的第一个想法是, 这个框架用到了递归, 这和我之前看到的归并排序有点类似, 那我们来看看框架到底是怎么实现的:
ForkJoinPool 由 ForkJoinTask 数组和 ForkJoinWorkThread 数组组成. ForkJoinTask 负责将存储的任务提交给 ForkJoinPool,ForkJoinWorkThread 负责处理这些任务.
但是我简单的翻了下源码, 发现 ForkJoinPool 里面已经没有 ForkJoinWorkThread 数组了, 取而代之的是 ForkJoinWorkThread 工厂(博主 Java8).
既然是实现原理, 就从它的核心方法开始看起:
5.1submit 方法:
一个任务执行的开端是 submit 方法, 即将任务提交给 ForkJoinPool, 让它唤醒一个 ForkJoinWorkThread 线程来执行这个任务:
- public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
- // 任务为空抛异常
- if (task == null)
- throw new NullPointerException();
- // 调用 externalPush 方法, 这个我们往下看, 看看它玩啥花样
- externalPush(task);
- return task;
- }
- final void externalPush(ForkJoinTask <?>task) {
- // 工作队列, 也就是之前介绍框架时所提到的偷窃队列
- // 创建队列数组的个数和线程数一致
- WorkQueue[] ws;
- WorkQueue q;
- int m;
- //getProbe 用来为当前线程生成一个不强制初始化的取样值
- int r = ThreadLocalRandom.getProbe();
- //runState 为 ForkJoinPool 的 volatile 字段, 表示锁的状态
- int rs = runState;
- //workQueues 为 ForkJoinPool 的数组 volatile 类型字段
- // 当这个队列不为空(如果为空肯定是要初始化一个队列的)
- // 并且队列长度大于等于 0
- // 额... 后面原谅我看不懂了
- // 但是大概的意思应该是要调用我们自定义的 task 对象的方法, 对这个任务进行分割
- if ((ws = workQueues) != null && (m = (ws.length - 1))>= 0 && (q = ws[m & r & SQMASK]) != null && r != 0 && rs> 0 && U.compareAndSwapInt(q, QLOCK, 0, 1)) {
- ForkJoinTask <?>[] a;
- int am,
- n,
- s;
- if ((a = q.array) != null && (am = a.length - 1)> (n = (s = q.top) - q.base)) {
- int j = ((am & s) <<ASHIFT) + ABASE;
- // 将 task 放入 ForkJoinTask 数组中
- U.putOrderedObject(a, j, task);
- U.putOrderedInt(q, QTOP, s + 1);
- U.putIntVolatile(q, QLOCK, 0);
- //signalWork 是用来创建线程来执行任务
- if (n <= 1) signalWork(ws, q);
- return;
- }
- U.compareAndSwapInt(q, QLOCK, 1, 0);
- }
- externalSubmit(task);
- }
5.2fork 方法:
我参考过并发编程的艺术, Java7 中的 fork 方法和 Java8 的 fork 方法有点小不同!
以下代码是 Java8 的:
- 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 方法是由 submit 方法触发的, 且 submit 方法里面唤醒过 ForkJoinWorkerThread 来处理队列中的任务, 所以通过 push 方法将当前分割后任务放到队列中, 同时调用 signalWork 唤醒线程对分割后的线程进行处理.
- final void push(ForkJoinTask<?> task) {
- ForkJoinTask<?>[] a; ForkJoinPool p;
- int b = base, s = top, n;
- if ((a = array) != null) { // ignore if queue removed
- int m = a.length - 1; // fenced write for task visibility
- U.putOrderedObject(a, ((m & s) <<ASHIFT) + ABASE, task);
- U.putOrderedInt(this, QTOP, s + 1);
- if ((n = s - b) <= 1) {
- if ((p = pool) != null)
- p.signalWork(p.workQueues, this);
- }
- else if (n>= m)
- growArray();
- }
- }
我对这个框架的理解是这样的(因为网上关于 1.8 的 fork/join 框架介绍比较少所以博主水平有限, 理解可能有偏差):
1)submit 提交 task
2)将 task 暂存到 ForkJoinTask 数组中
3)调用 signalWork 方法创建或唤醒线程来执行 ForkJoinTask 数组中的任务
4)线程拿到任务后调用 fork 方法
5)fork 方法将任务再分割, 放到 ForkJoinTask 数组中(分割前的总任务清 除), 然后再唤醒线程来处理 ForkJoinTask 数组中的任务(唤醒线程的个数取决任务个数).
6)fork 后通过调用 join 方法将结果合并.
7)任务继续分割, 直到不能再分割, 然后各个线程进行计算.
8)像递归一样, 将数据合并并返回, 最终返回到 submit 方法中.
以上就是博主读 fork/join 框架的理解, 可能偏差很大, 大的离谱, 如果有大神看见还请赐教!
分析完上面的, 博主还是不放心, 那就断点走起吧:
上图是我们要执行的主程序(完整程序上面有), 第一行和第二行没什么新建一些对象, 当执行到
Future<Integer> result = forkJoinPool.submit(task);
时我们看看都有哪些变量:
这个时候还没啥迹象, 且 result 都没有值, 只有这段代码执行结束才能看出点啥
- Future<Integer> result = forkJoinPool.submit(task);
- :
看看最终形态:
额, 告诉大家一个不幸的消息, 我没看出啥~(滑稽脸. jpg)
此次分析到此结束, 有空再来拜读源码!
恳请有大牛看见了此篇文章指点一下, 万谢!!!
2018.4.13 17:19
来源: https://www.cnblogs.com/MindMrWang/p/8822013.html