Java7之forkjoin简介_动力节点Java学院整理
作者:mrr 发布时间:2023-08-31 14:08:28
Java7引入了Fork Join的概念,来更好的支持并行运算。顾名思义,Fork Join类似与流程语言的分支,合并的概念。也就是说Java7 SE原生支持了在一个主线程中开辟多个分支线程,并且根据分支线程的逻辑来等待(或者不等待)汇集,当然你也可以fork的某一个分支线程中再开辟Fork Join,这也就可以实现Fork Join的嵌套。
有两个核心类ForkJoinPool和ForkJoinTask。
ForkJoinPool实现了ExecutorService接口,起到线程池的作用。所以他的用法和Executor框架的使用时一样的,当然Fork Join本身就是Executor框架的扩展。ForkJoinPool有3个关键的方法,来启动线程,execute(…),invoke(…),submit(…)。
客户端非fork/join调用 内部调用fork/join
异步执行 execute(ForkJoinTask) ForkJoinTask.fork
等待获取结果 invoke(ForkJoinTask) ForkJoinTask.invoke
执行,获取Futrue submit(ForkJoinTask) ForkJoinTask.fork(ForkJoinTasks are Futures)
ForkJoinTask是分支合并的执行任何,分支合并的业务逻辑使用者可以再继承了这个抽先类之后,在抽象方法exec()中实现。其中exec()的返回结果和ForkJoinPool的执行调用方(execute(…),invoke(…),submit(…)),共同决定着线程是否阻塞,具体请看下面的测试用例。
首先,用户需要创建一个自己的ForkJoinTask。代码如下:
public class MyForkJoinTask extends ForkJoinTask {
/**
*
*/
private static final long serialVersionUID = 1L;
private V value;
private boolean success = false;
@Override
public V getRawResult() {
return value;
}
@Override
protected void setRawResult(V value) {
this.value = value;
}
@Override
protected boolean exec() {
System.out.println("exec");
return this.success;
}
public boolean isSuccess() {
return success;
}
public void setSuccess(boolean isSuccess) {
this.success = isSuccess;
}
}
测试ForkJoinPool.invoke(…):
@Test
public void testForkJoinInvoke() throws InterruptedException, ExecutionException {
ForkJoinPool forkJoinPool = new ForkJoinPool();
MyForkJoinTask task = new MyForkJoinTask();
task.setSuccess(true);
task.setRawResult("test");
String invokeResult = forkJoinPool.invoke(task);
assertEquals(invokeResult, "test");
}
@Test
public void testForkJoinInvoke2() throws InterruptedException, ExecutionException {
final ForkJoinPool forkJoinPool = new ForkJoinPool();
final MyForkJoinTask task = new MyForkJoinTask();
new Thread(new Runnable() {
public void run() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
task.complete("test");
}
}).start();
// exec()返回值是false,此处阻塞,直到另一个线程调用了task.complete(...)
String result = forkJoinPool.invoke(task);
System.out.println(result);
}
@Test
public void testForkJoinSubmit() throws InterruptedException, ExecutionException {
final ForkJoinPool forkJoinPool = new ForkJoinPool();
final MyForkJoinTask task = new MyForkJoinTask();
task.setSuccess(true); // 是否在此任务运行完毕后结束阻塞
ForkJoinTask result = forkJoinPool.submit(task);
result.get(); // 如果exec()返回值是false,在此处会阻塞,直到调用complete
}
测试ForkJoinPool.submit(…):
@Test
public void testForkJoinSubmit() throws InterruptedException, ExecutionException {
final ForkJoinPool forkJoinPool = new ForkJoinPool();
final MyForkJoinTask task = new MyForkJoinTask();
task.setSuccess(true); // 是否在此任务运行完毕后结束阻塞
ForkJoinTask result = forkJoinPool.submit(task);
result.get(); // 如果exec()返回值是false,在此处会阻塞,直到调用complete
}
@Test
public void testForkJoinSubmit2() throws InterruptedException, ExecutionException {
final ForkJoinPool forkJoinPool = new ForkJoinPool();
final MyForkJoinTask task = new MyForkJoinTask();
forkJoinPool.submit(task);
Thread.sleep(1000);
}
@Test
public void testForkJoinSubmit3() throws InterruptedException, ExecutionException {
final ForkJoinPool forkJoinPool = new ForkJoinPool();
final MyForkJoinTask task = new MyForkJoinTask();
new Thread(new Runnable() {
public void run() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
task.complete("test");
}
}).start();
ForkJoinTask result = forkJoinPool.submit(task);
// exec()返回值是false,此处阻塞,直到另一个线程调用了task.complete(...)
result.get();
Thread.sleep(1000);
}
测试ForkJoinPool.execute(…):
@Test
public void testForkJoinExecute() throws InterruptedException, ExecutionException {
ForkJoinPool forkJoinPool = new ForkJoinPool();
MyForkJoinTask task = new MyForkJoinTask();
forkJoinPool.execute(task); // 异步执行,无视task.exec()返回值。
}
在实际情况中,很多时候我们都需要面对经典的“分治”问题。要解决这类问题,主要任务通常被分解为多个任务块(分解阶段),其后每一小块任务被独立并行计算。一旦计算任务完成,每一快的结果会被合并或者解决(解决阶段)。ForkJoinTask天然就是为了支持“分治”问题的。
分支/合并的完整过程如下:
下面列举一个分治算法的实例。
import java.util.Random;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
public class MaximumFinder extends RecursiveTask<Integer> {
private static final int SEQUENTIAL_THRESHOLD = 5;
private final int[] data;
private final int start;
private final int end;
public MaximumFinder(int[] data, int start, int end) {
this.data = data;
this.start = start;
this.end = end;
}
public MaximumFinder(int[] data) {
this(data, 0, data.length);
}
@Override
protected Integer compute() {
final int length = end - start;
if (length < SEQUENTIAL_THRESHOLD) {
return computeDirectly();
}
final int split = length / 2;
final MaximumFinder left = new MaximumFinder(data, start, start + split);
left.fork();
final MaximumFinder right = new MaximumFinder(data, start + split, end);
return Math.max(right.compute(), left.join());
}
private Integer computeDirectly() {
System.out.println(Thread.currentThread() + ' computing: ' + start
+ ' to ' + end);
int max = Integer.MIN_VALUE;
for (int i = start; i < end; i++) {
if (data[i] > max) {
max = data[i];
}
}
return max;
}
public static void main(String[] args) {
// create a random data set
final int[] data = new int[1000];
final Random random = new Random();
for (int i = 0; i < data.length; i++) {
data[i] = random.nextInt(100);
}
// submit the task to the pool
final ForkJoinPool pool = new ForkJoinPool(4);
final MaximumFinder finder = new MaximumFinder(data);
System.out.println(pool.invoke(finder));
}
}
以上所示是小编给大家介绍的Java7之forkjoin简介_动力节点Java学院整理,希望对大家有所帮助
猜你喜欢
- 一、前言在Spring中,事务有两种实现方式:编程式事务管理: 编程式事务管理使用TransactionTemplate可实现更细
- 前言Android提供了很多种保存应用程序数据的方法。其中一种就是用SharedPreferences对象来保存我们私有的键值(key-va
- 前言服务消费者调用服务提供者的时候使用RestTemplate技术存在不便之处:拼接urlrestTmplate.getForObJect这
- 背景实际开发中,常常需要将比较复杂的 JSON 字符串转换为对应的 Java 对象。这里记录下解决方案。如下所示,是入侵事件检测得到的 JS
- 一、IO流的分类字符流ReaderInputStreamReader(节点流)BufferedReader(处理流)WriterOutput
- 1)首先启动hadoop2个进程,进入hadoop/sbin目录下,依次启动如下命令[root@node02 sbin]# pwd/usr/
- 前言在电商的应用中,最常见的就是在首页或完成某事件之后,弹出一堆的活动/广告。假如重叠弹出,很丑,给用户的体验也不好,所以一般都会依次依条件
- Executor接口基于以下方法可以完成增,删,改查以及事务处理等操作。事实上,mybatis中的所有数据库操作是通过调用这些方法实现的。p
- 背景在接口请求过程中,传递json对象,springboot转换为实体VO对象后,所有属性都为null。post请求:后台接收请求:当时就懵
- 静态代理: 由我们开发者自己手动创建或者在程序运行前就已经存在的代理类,静态代理通常只代理一个类, * 是代理一个接口下的多个实现类。动态
- 几个月前写过一篇博客《xUtils3.0框架学习笔记》 ,上面也有记录通过xUtils实现文件上传的使用方法,代码如下:private vo
- Android中的异步消息机制分为四个部分:Message、Handler、MessageQueue和Looper。其中,Message是线
- 解决方法有如下两种:第一种如果你 repo sync 了 android 的整个源码,那么可以直接把你的 app 放到 /packages/
- 本文实例为大家分享了iOS新浪微博分享功能的具体代码,供大家参考,具体内容如下做新浪分享 需先去http://open.weibo.com/
- FFmpeg是一套可以用来记录、转换数字音频、视频,并能将其转化为流的开源计算机程序。采用LGPL或GPL许可证。它提供了录制、转换以及流化
- /// <summary> /// 遍历Co
- 先上代码新建一个Thread,代码如下:package com.thread.test;public class MyThread exte
- 该工具包含是封装了jedis,包含redis.properties和jedisPool,序列化使用的是protostuff,map类型操作使
- 这篇文章主要介绍了spring boot如何实现切割分片上传,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需
- 背景传说里玉皇大帝派龙王马上降雨到共光一带,龙王接到玉皇大帝命令,立马从海上调水,跑去共光施云布雨,但粗心又着急的龙王不小心把海里的鲸鱼随着