大家好,又见面了,我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。
Jetbrains全系列IDE使用 1年只要46元 售后保障 童叟无欺
声明:本篇博客是在阅读了引用博客的两篇文章后做了简短的概括与归纳,只作为自己笔记
文章目录
一、思想
Fork/Join是Java7提供的并行执行任务的框架,是一个把大人物分割成若干小任务,最终汇总小任务的结果得到大任务结果的框架
小任务可以继续拆分为更小的任务
二、工作窃取算法
1、工作窃取会选择双端队列作为存储任务的数据结构,默认正常线程会选择LIFO(栈获取)的方式,从当前双端队列的尾部获取任务;窃取线程会选择FIFO(队列获取)方式,从当前双端队列的头部获取任务
默认添加元素是从双端队列的尾部添加元素
三、demo用例
通过Fork/Join并行计算1+2+3+4
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;
public class ForkjoinDemo extends RecursiveTask<Long> {
private long start;
private long end;
public static final int THRESHOLD = 2;
public ForkjoinDemo(long start, long end) {
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
long sum = 0;
boolean flag = (end - start) <= THRESHOLD;
if (flag) {
for (long i = start; i <= end; i++) {
sum += i;
}
} else {
long middle = (end + start) / 2;
// divide task
ForkjoinDemo leftTask = new ForkjoinDemo(start, middle);
ForkjoinDemo rightTask = new ForkjoinDemo(middle+1 , end);
// execute sub task
leftTask.fork();
rightTask.fork();
// get result sub task
long left = leftTask.join();
long right = rightTask.join();
sum = left + right;
}
return sum;
}
public static void main(String[] args) {
ForkJoinPool pool = new ForkJoinPool();
ForkjoinDemo task = new ForkjoinDemo(1, 4);
Future<Long> result = pool.submit(task);
try {
System.out.println(result.get());
} catch (InterruptedException | ExecutionException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
四、关键组件
ForkJoinPool
(一)线程池的作用可能是用来合理分配线程资源:
1、接受外部任务的提交(外部调用ForkJoinPool的invoke/execute/submit方法提交任务);
2、接受ForkJoinTask自身fork出的子任务的提交;
3、任务队列数组(WorkQueue[])的初始化和管理;
4、工作线程(Worker)的创建/管理
(二)提交方式的差异:
invoke:同步提交,有返回值,任务执行完成后返回
submit:异步提交,有返回值,调用线程立即返回
execute:异步提交,无返回值,调用线程立即返回
(三)初始化线程池方式
1、FIFO_QUEUE:先进先出,异步模式
2、LIFO_QUEUE:(默认)先进后出,同步模式
ForkJoinTask
(一)子类
1、RecursiveAction:用于没有返回结果的任务
2、RecursiveTask:用于有返回结果的任务
(二)函数
1、fork()拆分为子任务
2、join()合并子任务
ForkJoinWorkerThread
任务处理原则:首先根据同步/异步模式从任务队列选择任务,如果完成自身任务,通过窃取算法获取其他线程的任务
WorkQueue
底层是通过数组实现的双端队列,容量为2的幂次,任务队列在首次调用线程池外部方法提交任务之后初始化任务队列,通过ThreadLocalRandom.probe来计算出任务队列在数组中的索引位置(外部方法调用产生的索引一定是偶数),没有绑定工作线程
1、有工作线程(Worker)绑定的任务队列:数组下标始终是奇数,称为task queue,该队列中的任务均由工作线程调用产生(工作线程调用FutureTask.fork方法);
2、没有工作线程(Worker)绑定的任务队列:数组下标始终是偶数,称为submissions queue,该队列中的任务全部由其它线程提交(也就是非工作线程调用execute/submit/invoke或者FutureTask.fork方法)。
五、Fork/Join运行流程图
任务提交
(一)外部提交任务
1、提交任务命中,externalPush
2、提交任务未命中,externalSubmit
CASE1:线程池已经关闭,则执行终止操作,并拒绝该任务的提交;
CASE2:线程池未初始化,则进行初始化,主要就是初始化任务队列数组;
CASE3:命中了任务队列,则将任务入队,并尝试创建/唤醒一个工作线程(Worker);
CASE4:未命中任务队列,则在偶数索引处创建一个任务队列
(二)工作线程fork任务:直接push进当前任务队列
创建线程signalWork方法
1、工作线程数不足:
创建一个工作线程:
(1)通过createWorker方法通过线程池工厂创建线程,在线程创建的过程中,registerWorker会创建一个工作队列与工作线程绑定,创建成功,则start线程;
(2)创建失败:则deregisterWorker注销该工作线程,清除已关闭的工作线程或回滚创建线程之前的操作,并把传入的异常抛给 ForkJoinTask 来处理。
2、工作线程数足够:唤醒一个空闲(阻塞)的工作线程。
任务执行
1、runWorker执行任务
(1)scan()窃取任务:随机选择一个任务队列,获取底部任务,任务数大于1,则signalWork创建或唤醒其他工作线程。
(2)成功窃取runTask:
a.执行窃取任务:调用FutureTask.deExec()执行任务,其内部会调用FutureTask.exec()方法,该方法为抽象方法,由子类实现。
b.工作线程还会执行自己队列中的任务,即WorkQueue.execLocalTasks
(3)未成功窃取awaitWork:没有任务则阻塞
六、引用博客
极力推荐并发系列博客:https://segmentfault.com/a/1190000016781127
发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/191329.html原文链接:https://javaforall.cn
【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛
【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...