参考:https://www.infoq.cn/article/fork-join-introduction算法
Fork/Join 框架是 Java7 提供了的一个用于并行执行任务的框架, 是一个把大任务分割成若干个小任务,最终汇总每一个小任务结果后获得大任务结果的框架。数组
咱们再经过 Fork 和 Join 这两个单词来理解下 Fork/Join 框架,Fork 就是把一个大任务切分为若干子任务并行的执行,Join 就是合并这些子任务的执行结果,最后获得这个大任务的结果。好比计算 1+2+。。+10000,能够分割成 10 个子任务,每一个子任务分别对 1000 个数进行求和,最终汇总这 10 个子任务的结果。Fork/Join 的运行流程图以下:bash
工做窃取(work-stealing)算法是指某个线程从其余队列里窃取任务来执行。工做窃取的运行流程图以下:框架
那么为何须要使用工做窃取算法呢?假如咱们须要作一个比较大的任务,咱们能够把这个任务分割为若干互不依赖的子任务,为了减小线程间的竞争,因而把这些子任务分别放到不一样的队列里,并为每一个队列建立一个单独的线程来执行队列里的任务,线程和队列一一对应,好比 A 线程负责处理 A 队列里的任务。可是有的线程会先把本身队列里的任务干完,而其余线程对应的队列里还有任务等待处理。干完活的线程与其等着,不如去帮其余线程干活,因而它就去其余线程的队列里窃取一个任务来执行。而在这时它们会访问同一个队列,因此为了减小窃取任务线程和被窃取任务线程之间的竞争,一般会使用双端队列,被窃取任务线程永远从双端队列的头部拿任务执行,而窃取任务的线程永远从双端队列的尾部拿任务执行。异步
工做窃取算法的优势是充分利用线程进行并行计算,并减小了线程间的竞争,其缺点是在某些状况下仍是存在竞争,好比双端队列里只有一个任务时。而且消耗了更多的系统资源,好比建立多个线程和多个双端队列。ui
咱们已经很清楚 Fork/Join 框架的需求了,那么咱们能够思考一下,若是让咱们来设计一个 Fork/Join 框架,该如何设计?这个思考有助于你理解 Fork/Join 框架的设计。this
第一步分割任务。首先咱们须要有一个 fork 类来把大任务分割成子任务,有可能子任务仍是很大,因此还须要不停的分割,直到分割出的子任务足够小。spa
第二步执行任务并合并结果。分割的子任务分别放在双端队列里,而后几个启动线程分别从双端队列里获取任务执行。子任务执行完的结果都统一放在一个队列里,启动一个线程从队列里拿数据,而后合并这些数据。线程
Fork/Join 使用两个类来完成以上两件事情:设计
让咱们经过一个简单的需求来使用下 Fork/Join 框架,需求是:计算 1+2+3+4 的结果。
使用 Fork/Join 框架首先要考虑到的是如何分割任务,若是咱们但愿每一个子任务最多执行两个数的相加,那么咱们设置分割的阈值是 2,因为是 4 个数字相加,因此 Fork/Join 框架会把这个任务 fork 成两个子任务,子任务一负责计算 1+2,子任务二负责计算 3+4,而后再 join 两个子任务的结果。
由于是有结果的任务,因此必须继承 RecursiveTask,实现代码以下:
经过这个例子让咱们再来进一步了解 ForkJoinTask,ForkJoinTask 与通常的任务的主要区别在于它须要实现 compute 方法,在这个方法里,首先须要判断任务是否足够小,若是足够小就直接执行任务。若是不足够小,就必须分割成两个子任务,每一个子任务在调用 fork 方法时,又会进入 compute 方法,看看当前子任务是否须要继续分割成孙任务,若是不须要继续分割,则执行当前子任务并返回结果。使用 join 方法会等待子任务执行完并获得其结果。
ForkJoinTask 在执行的时候可能会抛出异常,可是咱们没办法在主线程里直接捕获异常,因此 ForkJoinTask 提供了 isCompletedAbnormally() 方法来检查任务是否已经抛出异常或已经被取消了,而且能够经过 ForkJoinTask 的 getException 方法获取异常。使用以下代码:
if(task.isCompletedAbnormally())
{
System.out.println(task.getException());
}
复制代码
getException 方法返回 Throwable 对象,若是任务被取消了则返回 CancellationException。若是任务没有完成或者没有抛出异常则返回 null。
ForkJoinPool 由 ForkJoinTask 数组和 ForkJoinWorkerThread 数组组成,ForkJoinTask 数组负责存放程序提交给 ForkJoinPool 的任务,而 ForkJoinWorkerThread 数组负责执行这些任务。
ForkJoinTask 的 fork 方法实现原理。当咱们调用 ForkJoinTask 的 fork 方法时,程序会调用 ForkJoinWorkerThread 的 pushTask 方法异步的执行这个任务,而后当即返回结果。代码以下:
public final ForkJoinTask fork() { ((ForkJoinWorkerThread) Thread.currentThread()) .pushTask(this); return this; } 复制代码
pushTask 方法把当前任务存放在 ForkJoinTask 数组 queue 里。而后再调用 ForkJoinPool 的 signalWork() 方法唤醒或建立一个工做线程来执行任务。代码以下:
final void pushTask(ForkJoinTask t) {
ForkJoinTask[] q; int s, m;
if ((q = queue) != null) { // ignore if queue removed
long u = (((s = queueTop) & (m = q.length - 1)) << ASHIFT) + ABASE;
UNSAFE.putOrderedObject(q, u, t);
queueTop = s + 1; // or use putOrderedInt
if ((s -= queueBase) <= 2)
pool.signalWork();
else if (s == m)
growQueue();
}
}
复制代码
ForkJoinTask 的 join 方法实现原理。Join 方法的主要做用是阻塞当前线程并等待获取结果。让咱们一块儿看看 ForkJoinTask 的 join 方法的实现,代码以下:
public final V join() {
if (doJoin() != NORMAL)
return reportResult();
else
return getRawResult();
}
private V reportResult() {
int s; Throwable ex;
if ((s = status) == CANCELLED)
throw new CancellationException();
if (s == EXCEPTIONAL && (ex = getThrowableException()) != null)
UNSAFE.throwException(ex);
return getRawResult();
}
复制代码
首先,它调用了 doJoin() 方法,经过 doJoin() 方法获得当前任务的状态来判断返回什么结果,任务状态有四种:已完成(NORMAL),被取消(CANCELLED),信号(SIGNAL)和出现异常(EXCEPTIONAL)。
让咱们再来分析下 doJoin() 方法的实现代码:
private int doJoin() {
Thread t; ForkJoinWorkerThread w; int s; boolean completed;
if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
if ((s = status) < 0)
return s;
if ((w = (ForkJoinWorkerThread)t).unpushTask(this)) {
try {
completed = exec();
} catch (Throwable rex) {
return setExceptionalCompletion(rex);
}
if (completed)
return setCompletion(NORMAL);
}
return w.joinTask(this);
}
else
return externalAwaitDone();
}
复制代码
在 doJoin() 方法里,首先经过查看任务的状态,看任务是否已经执行完了,若是执行完了,则直接返回任务状态,若是没有执行完,则从任务数组里取出任务并执行。若是任务顺利执行完成了,则设置任务状态为 NORMAL,若是出现异常,则纪录异常,并将任务状态设置为 EXCEPTIONAL。