Fork/Join 框架框架使用
框架 Join fork 使用
2023-09-27 14:25:36 时间
1、介绍
Fork/Join 框架是 Java7 提供了的一个用于并行执行任务的框架, 是一个把大任务分割成若干个小任务,最终汇总每个小任务结果后得到大任务结果的框架。在多核计算机中正确使用可以很好的发挥cpu的作用,提高程序的执行效率。框架采用工作窃取算法,当有子任务线程处理完当前任务时,它会从其他线程执行的任务队列里窃取任务来执行,从而提高整体的执行效率。为了减少线程间的任务资源竞争,队列通常使用双端队列,别窃取任务线程永远从啥UN广大UN队列的呕吐不获取任务执行,而窃取任务的线程永远从双端队列的尾部获取任务执行。
2、使用
根据业务场景来考虑是否需要使用Fork/Join框架来进行任务的拆分和汇总操作。当需要时,比如说需要执行一个很大的业务计算之类的,此时使用Fork/Join框架分以下两步:
- 对任务进行分割 把大任务分割成子任务,有可能子任务还是很大,所以还需要不停的分割,直到分割出的子任务足够小
- 执行分割的子任务并汇总结果 分割的子任务分别放在双端队列里,然后几个启动线程分别从双端队列里获取任务执行。子任务执行完的结果都统一放在一个队列里,启动一个线程从队列里拿数据,然后合并这些数据
具体实现以上两步:
- 创建ForkJoinTask 它提供在任务中执行 fork() 和 join() 操作的机制,通常情况下我们不需要直接继承 ForkJoinTask 类,而只需要继承它的子类,Fork/Join 框架提供了以下两个子类: RecursiveAction:用于没有返回结果的任务 RecursiveTask :用于有返回结果的任务
- 使用ForkJoinPool执行ForkJoinTask 任务分割出的子任务会添加到当前工作线程所维护的双端队列中,进入队列的头部。当一个工作线程的队列里暂时没有任务时,它会随机从其他工作线程的队列的尾部获取一个任务
3、示例
3.1 使用fork
public class SumTask extends RecursiveTask <Integer>{ private static final Integer THRESHOLD = 1000; private int start; private int end; public SumTask(int start, int end) { this.start = start; this.end = end; } @Override protected Integer compute() { Integer sum = 0; boolean isOk = (end - start) <= THRESHOLD; if(isOk) { for(int i = start; i <= end; i ++) { sum += i; } return sum; } int middle = (end + start) / 2; //子任务递归 SumTask sumSubTask = new SumTask(start, middle); SumTask sumSubTask1 = new SumTask(middle + 1, end); //fork子任务 sumSubTask.fork(); sumSubTask1.fork(); //join子任务 Integer join = sumSubTask.join(); Integer join1 = sumSubTask1.join(); sum = join + join1; //计算结果 return sum; } }
3.2 使用invokeAll
public class SumTask2 extends RecursiveTask <Integer>{ private static final Integer THRESHOLD = 1000; private int start; private int end; public SumTask2(int start, int end) { this.start = start; this.end = end; } @Override protected Integer compute() { Integer sum = 0; boolean isOk = end - start <= THRESHOLD; if(isOk) { for(int i = start; i <= end; i ++) { sum += i; } // System.out.println(String.format("compute %d-%d = %d", start, end, sum)); return sum; } //除以2 int middle = (end + start) / 2; //子任务递归 // System.out.println(String.format("fork %d-%d => %d-%d&%d-%d", start, end, start, middle - 1, middle, end)); SumTask2 sumSubTask = new SumTask2(start, middle - 1); SumTask2 sumSubTask1 = new SumTask2(middle, end); //fork子任务 invokeAll(sumSubTask, sumSubTask1); //join子任务 Integer join = sumSubTask.join(); Integer join1 = sumSubTask1.join(); sum = join + join1; //计算结果 return sum; } }
测试
ForkJoinPool fjp2 = new ForkJoinPool(); SumTask2 sumTask2 = new SumTask2(start, end); long begin3 = System.currentTimeMillis(); Integer invoke = fjp2.invoke(sumTask2); long end3 = System.currentTimeMillis(); System.out.println("计算结果3为 sum = " + invoke + ",计算时长为" + begin3 + "-" + end3 + "--- " + (end3 - begin3) + "ms"); ForkJoinPool fjp = new ForkJoinPool(); long begin2 = System.currentTimeMillis(); SumTask sumTask = new SumTask(start, end); ForkJoinTask<Integer> submit = fjp.submit(sumTask); Integer join = submit.join(); long end2 = System.currentTimeMillis(); System.out.println("计算结果2为 sum = " + join + ",计算时长为" + begin2 + "-" + end2 + "--- " + (end2 - begin2) + "ms");
结果
从结果(可以多次运行测试)可以看出,使用invokeAll方式效率比使用单独fork方式高,所以在使用时尽量采用invokeAll方式,这样可以充分利用线程池中的线程去执行任务。
源码参照Github
相关文章
- flask框架不同路由之间传递变量
- Selenium+Pytest自动化测试框架实战
- Fork/Join框架(四)异步运行任务
- 跟常用框架Spring Security`和shiro比 ,微服务到底胜在哪?
- thinkphp 3.2.3+Admin LTE后台框架
- 罗列几个Android插件化开发框架
- Pytest单元测试框架实战之setup_function和teardown_function
- AWTK 全称为 Toolkit AnyWhere,是 ZLG 倾心打造的一套基于 C 语言开发的 GUI 框架(三平台+2个手机平台+嵌入式)
- ubuntu16.04 源码安装Python3.7 (可以在此基础上安装Tensorflow) (确保Tensorflow计算框架与系统的彻底隔离)
- java 中的fork join框架
- 初学SSM框架感觉一团糟,希望这篇文章能帮到你!(Mybatis篇)
- vue项目中使用了vw适配方案,引入第三方ui框架mint-ui时,适配问题解决
- AI学习---TensorFlow框架介绍[图+会话+张量+变量OP+API]
- 怎样更好的设计你的REST API之基于REST架构的Web Service设计及REST框架实现
- 框架设计之ADO.NET Command的ExecuteScalar误用情景及底层解说
- element框架中表格的筛选功能使用说明(转载)
- C++高性能服务器框架——日志系统