当前位置:网站首页>并发编程系列之什么是ForkJoin框架?
并发编程系列之什么是ForkJoin框架?
2022-07-01 15:40:00 【51CTO】
并发编程系列之什么是ForkJoin框架?
1、什么是ForkJoin框架
ForkJoin框架是java的JUC包里提供的,用于处理一些比较繁重的任务,会将这个大任务分为多个小任务,多个小任务处理完成后会将结果汇总给Result,体现的是一种“分而治之”的思想。第一步,拆分fork任务,将大任务分为多个小任务;第二步,归并join,会将小任务的处理结果进行归并为一个结果。
在这里插入图片描述
2、ForkJoinTask
ForkJoinTask
是ForkJoin
框架的提供的任务API,ForkJoinTask
是一个抽象类,有两个主要的实现类,RecursiveTask
和RecursiveAction
,其中RecursiveTask
和RecursiveAction
的主要区别是,RecursiveAction
没有返回值,而RecursiveTask
是有返回值的
在这里插入图片描述
3、ForkJoinPool
ForkJoinPool类是forkjoin框架的线程池实现,基于ExecutorService接口。这个线程池是jdk1.7才加入的,用于管理线程,执行forkjoin的任务。对于线程池的使用,我们使用ThreadPoolExecutor比较多,可以在idea里看一下uml类图,可以看出ForkJoinPool和ThreadPoolExecutor实现差不多的。
在这里插入图片描述
几个重要的参数:
- parallelism:并行度,并行执行线程,可用指定,也可以不指定,不指定的情况,是根据cpu核数创建可用的线程
- ForkJoinWorkerThreadFactory:创建线程的工厂实现
- UncaughtExceptionHandler :因为未知异常中断的回调处理
- asyncMode:是否异步,默认情况是false
使用时候,可以直接创建ForkJoinPool,可以不传参,不传参的情况,默认指定的线程并行数为Runtime.getRunTime().availableProcessors();
,表示根据cpu核数创建可用线程数
也是可用传参,对并行度进行指定的public ForkJoinPool(int parallelism)
, parallelism并行度,并行执行几个线程
将forkjoin任务加入到FrokJoinPool线程池有几种方式
- execute():调用其 fork 方法在多个线程之间拆分工作。
- invoke():在ForkJoinPool线程池上调用invoke方法
- submit():返回一个Future对象,Future可以进行监控,任务完成后返回结果
4、打印斐波那契数列
ForkJoin框架可以用于一些递归遍历的场景,对于斐波那契数列,你可以比较熟悉,因为在面试中有时候经常问到,斐波那契数列的特点就是最后一项的结果等于前面两项的和
package com.example.concurrent.forkjoin;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
/**
* <pre>
* 斐波那契数列
* </pre>
* <p>
* <pre>
* @author nicky.ma
* 修改记录
* 修改后版本: 修改人: 修改日期: 2021/10/12 16:22 修改内容:
* </pre>
*/
public class Fibonacci extends RecursiveTask<Integer>{
private int n;
public Fibonacci(int n) {
this.n = n;
}
@Override
protected Integer compute() {
if (n <= 1)
return n;
Fibonacci f1 = new Fibonacci(n - 1);
f1.fork();
Fibonacci f2 = new Fibonacci(n - 2);
f2.fork();
return f1.join() + f2.join();
}
public static void main(String[] args) throws ExecutionException, InterruptedException {
ForkJoinPool pool = new ForkJoinPool();
for (int i = 0; i< 10; i++) {
ForkJoinTask task = pool.submit(new Fibonacci(i));
System.out.println(task.get());
}
}
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
- 27.
- 28.
- 29.
- 30.
- 31.
- 32.
- 33.
- 34.
- 35.
- 36.
- 37.
- 38.
- 39.
- 40.
- 41.
- 42.
- 43.
- 44.
- 45.
5、ForkJoin归并排序
面试题:快速实现对一个长度百万的数组的排序
难点:可以使用归并排序,多线程如何组织实现归并排序
package com.example.concurrent.forkjoin;
import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;
/**
* <pre>
* 大数组排序
* </pre>
* <p>
* <pre>
* @author mazq
* 修改记录
* 修改后版本: 修改人: 修改日期: 2021/10/12 17:04 修改内容:
* </pre>
*/
public class ArraySortTask extends RecursiveAction{
final long[] array; final int lo, hi;
ArraySortTask(long[] array, int lo, int hi) {
this.array = array; this.lo = lo; this.hi = hi;
}
ArraySortTask(long[] array) { this(array, 0, array.length); }
@Override
protected void compute() {
if (hi - lo < THRESHOLD)
// 少于阀值,使用Arrays.sort 快排
sortSequentially(lo, hi);
else {
/* 归并排序 */
// 取中间值
int mid = (lo + hi) >>> 1;
// 拆分任务
invokeAll(new ArraySortTask(array, lo, mid),
new ArraySortTask(array, mid, hi));
// 归并结果
merge(lo, mid, hi);
}
}
// implementation details follow:
static final int THRESHOLD = 1000;
void sortSequentially(int lo, int hi) {
Arrays.sort(array, lo, hi);
}
void merge(int lo, int mid, int hi) {
long[] buf = Arrays.copyOfRange(array, lo, mid);
for (int i = 0, j = lo, k = mid; i < buf.length; j++)
array[j] = (k == hi || buf[i] < array[k]) ?
buf[i++] : array[k++];
}
public static void main(String[] args) throws ExecutionException, InterruptedException {
int size = 10_000;
long[] array = new long[size];
Random random = new Random();
for (int i = 0; i< size; i++) {
array[i] = random.nextInt();
}
ForkJoinPool forkJoinPool = new ForkJoinPool();
ArraySortTask task = new ArraySortTask(array , 0 , size);
forkJoinPool.submit(task);
task.get();
for (long a : array) {
System.out.println(a);
}
}
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
- 27.
- 28.
- 29.
- 30.
- 31.
- 32.
- 33.
- 34.
- 35.
- 36.
- 37.
- 38.
- 39.
- 40.
- 41.
- 42.
- 43.
- 44.
- 45.
- 46.
- 47.
- 48.
- 49.
- 50.
- 51.
- 52.
- 53.
- 54.
- 55.
- 56.
- 57.
- 58.
- 59.
- 60.
- 61.
- 62.
- 63.
- 64.
- 65.
- 66.
- 67.
- 68.
- 69.
- 70.
- 71.
- 72.
- 73.
- 74.
- 75.
参考资料
边栏推荐
- Flink 系例 之 TableAPI & SQL 与 Kafka 消息获取
- 【目标跟踪】|模板更新 时间上下文信息(UpdateNet)《Learning the Model Update for Siamese Trackers》
- The last picture is seamlessly connected with the first picture in the swiper rotation picture
- 《QT+PCL第六章》点云配准icp系列6
- 有些能力,是工作中学不来的,看看这篇超过90%同行
- Redis秒杀demo
- Can I choose to open an account on Great Wall Securities? Is it safe?
- MySQL审计插件介绍
- Returning to the top of the list, the ID is still weak
- Qt+pcl Chapter 9 point cloud reconstruction Series 2
猜你喜欢
The difference between arrow function and ordinary function in JS
《性能之巅第2版》阅读笔记(五)--file-system监测
Introduction to MySQL audit plug-in
如何实现时钟信号分频?
Short Wei Lai grizzly, to "touch China" in the concept of stocks for a living?
Photoshop插件-HDR(二)-脚本开发-PS插件
Zhang Chi's class: several types and differences of Six Sigma data
硬件开发笔记(九): 硬件开发基本流程,制作一个USB转RS232的模块(八):创建asm1117-3.3V封装库并关联原理图元器件
Microservice tracking SQL (support Gorm query tracking under isto control)
Survey of intrusion detection systems:techniques, datasets and challenges
随机推荐
采集数据工具推荐,以及采集数据列表详细图解流程
【OpenCV 例程200篇】216. 绘制多段线和多边形
选择在长城证券上炒股开户可以吗?安全吗?
Wechat applet 01 bottom navigation bar settings
你TM到底几点下班?!!!
《性能之巅第2版》阅读笔记(五)--file-system监测
Task.Run(), Task.Factory.StartNew() 和 New Task() 的行为不一致分析
Hardware design guide for s32k1xx microcontroller
Implementation of wechat web page subscription message
使用swiper制作手机端轮播图
《QT+PCL第六章》点云配准icp系列5
做空蔚来的灰熊,以“碰瓷”中概股为生?
Go zero actual combat demo (I)
Stm32f4-tft-spi timing logic analyzer commissioning record
S32K1xx 微控制器的硬件设计指南
药品溯源夯实安全大堤
[Cloudera][ImpalaJDBCDriver](500164)Error initialized or created transport for authentication
Survey of intrusion detection systems:techniques, datasets and challenges
【STM32学习】 基于STM32 USB存储设备的w25qxx自动判断容量检测
雷神科技冲刺北交所,拟募集资金5.4亿元