Fork/Join 框架
本文部分摘自《Java 并发编程的艺术》
Fork/Join 框架概述
Fork/Join 框架是 Java7 提供的一个用于并行执行任务的框架,是把一个大任务分割成若干个小任务,最终汇总每个小任务结果后得到大任务结果的框架,其运行流程如图所示:
工作窃取算法
工作窃取算法是指某个线程从其他队列里窃取任务来执行,为什么要这样做呢?假如我们需要做一个比较大的任务,可以把这个任务分割为若干个互不依赖的子任务,为了减少线程间的竞争,把这些子任务分别放到不同的队列里,并为每个队列创建一个单独的线程来执行队列里的任务,线程和队列一一对应。然而,如果某一线程先把自己队列的任务干完了,而其他线程对应的队列里还有任务等待处理,干完活的线程与其等着,不如去帮其他线程干活,这就是工作窃取算法的动机。
为了减少窃取任务线程和被窃取任务线程之间的竞争,通常会使用双端队列,被窃取任务线程永远从队列的头部拿任务执行,而窃取任务的线程永远从双端队列的尾部拿任务执行
使用 Fork/Join 框架
首先思考一下,如果让我们来设计一个 Fork/Join 框架,该如何设计呢?
分割任务
首先我们需要一个有 fork 类来把大任务分割成子任务,有可能子任务还是很大,所以需要不停地分割,直到分割出来的子任务足够小
执行任务并合并结果
分割的子任务分别放在双端队列里,然后几个启动线程分别从双端队列里获取任务执行。子任务执行完的结果都统一放在一个队列里,启动一个线程从队列里拿数据,然后进行合并
Fork/Join 使用两个类来完成以上两件事情:
ForkJoinTask
我们使用 ForkJoin 框架,必须首先创建 ForkJoin 任务,它提供在任务中执行 fork() 和 join() 操作的机制。通常情况下,我们不需要直接继承 ForkJoinTask 类,只需要继承它的子类即可,Fork/Join 框架提供了以下两个子类:
- RecursiveAction:用于没有返回结果的任务
- RecursiveTask:用于有返回结果待任务
ForkJoinPool
ForkJoinTask 需要通过 ForkJoinPool 来执行
我们通过一个简单的需求来使用 Fork/Join 框架,需求是:计算 1+2+3+4 的结果
使用 Fork/Join 框架把这个任务 fork 成两个子任务,子任务一负责计算 1+2,子任务而负责计算 3+4,然后再 join 两个子任务的结果,因为是有结果的任务,所以必须继承 RecursiveTask,代码实现如下:
public class CountTask extends RecursiveTask<Integer> {
// 阈值
private static final int THRESHOLD = 2;
private final int start;
private final int end;
public CountTask(int start, int end) {
this.start = start;
this.end = end;
}
@Override
protected Integer compute() {
int sum = 0;
boolean canCompute = (end - start) <= THRESHOLD;
// 如果任务足够小就计算任务
if (canCompute) {
for (int i = start; i <= end; i++) {
sum += i;
}
} else {
// 如果任务大于阈值,就分裂成两个子任务计算
int middle = (start + end) / 2;
CountTask leftTask = new CountTask(start, middle);
CountTask rightTask = new CountTask(middle + 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 = new ForkJoinPool();
// 生成一个计算任务,负责计算 1+2+3+4
CountTask task = new CountTask(1, 4);
// 执行一个任务
Future<Integer> result = forkJoinPool.submit(task);
try {
System.out.println(result.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
}
上面的例子中是通过 new ForkJoinPool(),然而这并不是其作者 Doug Lea 推荐的方式。ForkJoinPool 类有一个静态方法commonPool(),它所获得的 ForkJoinPool 实例是由整个应用进程共享的,可以帮助应用程序中多个需要进行归并计算的任务共享计算资源
ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();
ForkJoinTask 在执行的时候可能会抛出异常,但我们没办法在主线程直接捕获线程,所以 ForkJoinTask 提供了 isCompletedAbnormally() 方法来检查任务是否已经抛出异常或已经被取消,并可以通过 ForkJoinTask 的 getException 方法获取异常
if(task.isCompletedAbnormally()) {
System.out.println(task.getException());
}
Fork/Join 框架的实现原理
ForkJoinPool 中用来处理任务的工作线程采用的是 ForkJoinWorkerThread,它继承了 Thread 类,拥有两个非常关键的变量
final ForkJoinPool pool;
final ForkJoinPool.WorkQueue workQueue;
pool 是这个工作线程所属的 ForkJoinPool 实例,workQueue 是一个双端队列,可以发现,它是 ForkJoinPool 的一个内部类,其结构如下(省略部分代码)
static final class WorkQueue {
...
ForkJoinTask<?>[] array;
final ForkJoinPool pool;
final ForkJoinWorkerThread owner;
...
}
WorkQueue 里维护一个 ForkJoinTask 数组,用来存放待执行的任务(ForkJoinTask)。所以 Fork/Join 框架的基本思想就是:ForkJoinPool 的每个工作线程都维护着一个工作队列(WorkQueue),里面存放的对象是任务,每个工作线程处理自己的工作队列里的任务
fork() 方法做的工作只有一件事,既是把任务推入当前工作线程的工作队列里
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;
}
join() 方法的工作则复杂一些,首先会判断线程是否为 ForkJoinThread 线程,如果不是,阻塞当前线程,等待任务完成,如果是,则不阻塞。接着查看任务的完成状态,如果已经完成,直接返回结果,否则从队列中取出任务执行
最新文章
- NetMQ(三): 发布订阅模式 Publisher-Subscriber
- Effective C++ 1.让自己习惯C++
- Oracle的Export用法
- 【JS】Beginner7:Functions
- ROW_NUMBER() OVER函数的基本用法用法
- OSChina 其中很重要的一类——RequestContext
- 一、ASP.NET Routing路由(深入解析路由系统架构原理)
- JS控制台打印星星,总有你要的那一款~
- Java并发编程-各种锁
- 详解BLE 空中包格式—兼BLE Link layer协议解析
- abstract class VS interface
- Shell文本处理四剑客
- cmd 查看端口
- 最长公共子序列与最长公共字串 (dp)转载http://blog.csdn.net/u012102306/article/details/53184446
- C# Monitor的Wait和Pulse方法使用详解
- 1-9-假期训练心得(dp+bfs)
- 《mysql必知必会》学习_第八章_20180730_欢
- vmware虚拟机三种网络连接方式
- 如何打开chrome中flash debug player
- Java动态代理探讨
热门文章
- HTML5 drag &; drop &; H5 DnD
- 如何用 js 实现一个 call 函数
- Stack Overflow &; Segment Fault
- HTTP cache in depth
- React Native &; CodePush &; App Center
- Dyno-queues 分布式延迟队列 之 生产消费
- java高并发编程基础之AQS
- 通过const app = getApp()实现在 page 页面获取 app.js 定义的属性globalData,即获取全局数据
- nacos服务注册之服务器端Raft
- IDEA SVN 使用