本人花费半年的时间总结的《Java面试指南》已拿腾讯等大厂offer，已开源在github ，欢迎star！
前言
上节我们讲了阻塞队列，Java中的并发容器就算有了个基本的认识，今天我们来介绍一种线程工作模式，叫Fork/Join，他是JDK7之后提供的一个并行执行框架，主要的思想我觉得是分而治之，将一个大的任务分成多个小的任务并行执行，然后等所有小任务执行完毕后汇总结果，完成任务执行，是不是觉得这种模式挺好的，也有不少类似场景可以适用吧，那还等什么，开始进入我们今天的并发之旅吧。
什么是Fork/Join？
Fork/Join从上图很容易理解他是一种分合思想，Fork/Join最常用的场景就用在工作窃取算法上，那么什么是工作窃取算法呢？我们来对这个算法有个简单的认识：
窃取算法指一个线程从其他队列中窃取任务来执行，当然前提是本线程自己队列中的任务已经执行完毕，就比如，我们在工作过程中，有两个工人，每人分配了10个任务，其中某一个人干活效率比较高，当他干活自己的工作后，发现同事还有任务要做，就去帮他做，但是你又不能从第一个任务开始，因为前面的任务你并不知道你同事已经完成到第几个了，所以你就索性从最后一个任务开始往前执行，这就是工作窃取算法，当然我讲的比较通俗，那在我们Java中呢，最好是使用双向队列来做这件事情，线程1和线程2分别执行队列1和队列2中的任务，无论谁先执行完，都从队列尾部去窃取任务执行，当2个队列都空了，则任务全部执行完毕。应用Fork/Join中场景如下：
窃取算法优点：充分利用线程的并行计算，减少线程间的竞争；
窃取算法缺点：在极端情况下，队列中只有一个任务时，可能会发生两个线程或者多个线程同时执行，发生竞争，当然这种情况发生概率很低，此外该算法还会消耗更多的系统资源；
Fork/Join的设计？
Fork/Join从上面我们已经知道该框架的核心主要就是2步，那我们来看看如何将这简单的2步设计好，并为我们所用吧，
第一步：分割任务，首先我们需要有一个fork类来把大任务分割成子任务，有可能子任务还是很大，所以还需要不停的分割，直到分割出的子任务足够小（具体有多小就要根据我们的实际情况和系统资源，合适的配置执行子任务的线程数，并不是越小越好，也不是越多线程执行就会越快，因为我们知道，线程上下文切换是要占用CPU调度时间的）；
第二步：执行任务并合并结果，分割的子任务分别放在双向队列里，然后几个启动线程分别从双端队列里获取任务执行。子任务执行完的结果都统一放在一个队列里，启动一个线程从队列里拿数据，然后合并这些数据，进行汇总计算；
Fork/Join的使用？
了解了如何设计Fork/Join，接下来我们再来看看如何使用呢，我们要知道要使用这种模式，Fork/Join提供了下面两个类分别来完成第一步和第二步要做的事：
- RecursiveAction：于没有返回结果的任务
- RecursiveTask ：用于有返回结果的任务
RecursiveAction：于没有返回结果的任务
ForkJoinPool：ForkJoinTask需要通过ForkJoinPool来执行，任务分割出的子任务会添加到当前工作线程所维护的双端队列中，进入队列的头部。当一个工作线程的队列里暂时没有任务时，它会随机从其他工作线程的队列的尾部获取一个任务
下面我们以一个计算1+2+3+…的值：
public class ForkJoin extends RecursiveTaskLong {
// 计算1+2+3+...的值
private int start;
private int end;
private final int DEFAULT_CAPACITY = 20;
public ForkJoin(int start, int end) {
this.start = start;
this.end = end;
}
/**
* 重载RecursiveTask的方法
*
* @return
*/
@Override
protected Long compute() {
long sum = 0;
//如果任务量在最大容量之内
if (end - start DEFAULT_CAPACITY) {
for (int i = start; i = end; i++) {
sum += i;
}
} else {
//如果超过了最大容量，那么就进行拆分处理
//计算容量中间值
int middle = (start + end) / 2;
//进行递归
ForkJoin fockJoinTest1 = new ForkJoin(start, middle);
ForkJoin fockJoinTest2 = new ForkJoin(middle + 1, end);
//执行任务
fockJoinTest1.fork();
fockJoinTest2.fork();
//等待任务执行并返回结果
sum = fockJoinTest1.join() + fockJoinTest2.join();
}
return sum;
}
public static void main(String[] args) {
ForkJoinPool forkJoinPool = new ForkJoinPool();
ForkJoin forkJoin = new ForkJoin(1, 100);
//前面我们说过，任务提交中invoke可以直接返回结果
long result = forkJoinPool.invoke(forkJoin);
System.out.println("fock/join计算结果:" +result);
}
}
Fork/Join的实现原理
Fork/Join的底层，要根据JDK版本来论，1.8之前和1.8及之后的底层实现不同，不过这些对于开发者来说都是透明的。
下面我们看下JDK1.7及之前版本的底层实现：
ForkJoinPool由ForkJoinTask数组和ForkJoinWorkerThread数组组成，ForkJoinTask数组负责存放程序提交给ForkJoinPool的任务，而ForkJoinWorkerThread数组负责执行这些任务。
public ForkJoinPool(int parallelism,
ForkJoinWorkerThreadFactory factory,
Thread.UncaughtExceptionHandler handler,
boolean asyncMode) {
checkPermission();
...
this.submissionQueue = new ForkJoinTask?[INITIAL_QUEUE_CAPACITY];
...
workers = new ForkJoinWorkerThread[n + 1];
this.submissionLock = new ReentrantLock();
this.termination = submissionLock.newCondition();
StringBuilder sb = new StringBuilder("ForkJoinPool-");
sb.append(poolNumberGenerator.incrementAndGet());
sb.append("-worker-");
this.workerNamePrefix = sb.toString();
}
new ForkJoinTask?，我们再看下这个数组：
public final ForkJoinTaskV fork() {
// 当我们调用ForkJoinTask的fork方法时，
// 程序会调用ForkJoinWorkerThread的pushTask方法异步的执行这个任务，然后立即返回结果
((ForkJoinWorkerThread) Thread.currentThread())
.pushTask(this);
return this;
}
再看下pushTask方法：
// 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();
}
}
我们再看另一个方法Join方法：
// Join方法的主要作用是阻塞当前线程并等待获取结果
public final V join() {
// 它调用了doJoin()方法，通过doJoin()方法得到当前任务的状态来判断返回什么结果
if (doJoin() != NORMAL)
return reportResult();
else
return getRawResult();
}
//已完成（NORMAL），被取消（CANCELLED），信号（SIGNAL）和出现异常（EXCEPTIONAL）
// 如果任务状态是已完成，则直接返回任务结果。
// 如果任务状态是被取消，则直接抛出CancellationException。
// 如果任务状态是抛出异常，则直接抛出对应的异常
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方法：
// 首先通过查看任务的状态，看任务是否已经执行完了
// 如果执行完了，则直接返回任务状态
// 如果没有执行完，则从任务数组里取出任务并执行
// 如果任务顺利执行完成了，则设置任务状态为NORMAL
// 如果出现异常，则纪录异常，并将任务状态设置为EXCEPTIONAL
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();
}
对于Fork/Join在jdk8中的底层，我们下次有时间再分析
相关文章：
原文始发于微信公众号（Justin的后端书架）：
