并发编程之ForkJoin框架原理分析
点赞再看，养成习惯，公众号搜一搜【一角钱技术】关注更多原创技术文章。本文 GitHub org_hejianhui/JavaStudy 已收录，有个人系列文章。java
前言
前面咱们介绍了线程池框架（ExecutorService）的两个具体实现：git
线程池为线程生命周期的开销和资源不足问题提供了解决方案。经过对多个任务重用线程，线程建立的开销被分摊到多个任务上。Java7 又提供了的一个用于并行执行的任务的框架 Fork/Join ，是一个把大任务分割成若干个小任务，最终汇总每一个小任务结果后获得大任务结果的框架。在介绍Fork/Join 框架以前咱们先了解几个概念：CPU密集型、IO密集型，再逐步深刻去认识Fork/Join 框架。github
任务性质类型
CPU密集型（CPU bound）
CPU密集型也叫计算密集型，指的是系统的硬盘、内存性能相对于CPU要好很好多，此时，系统运做大部分的情况是 CPU Loading 100%，CPU要读/写 I/O（硬盘/内存），I/O在很短的时间就能够完成，而CPU还有许多运算要处理，CPU Loading很高。web
在多重程序系统中，大部分时间用来作计算、逻辑判断等CPU动做的程序称之 CPU bound。例如一个计算圆周率至小数点一千位如下的程序，在执行的过程中绝大部分时间在用三角函数和开根号的计算，即是属于CPU bound的程序。算法
CPU bound的程序通常而言CPU占用率至关高。这多是由于任务自己不太须要访问I/O设备，也多是由于程序是多线程实现所以屏蔽了等待I/O的时间。编程
线程数通常设置为：线程数 = CPU核数 + 1（现代CPU支持超线程）数组
IO密集型（I/O bound）
I/O密集型指的是系统的CPU性能相对硬盘、内存要好不少，此时，系统运做，大部分的情况是 CPU 在等 I/O（硬盘/内存）的读/写操做，此时 CPU Loading 并不高。markdown
I/O bound的程序通常在达到性能极限时，CPU占用率仍然较低。这多是由于任务自己须要大量I/O操做，而 pipeline 作的不是很好，没有充分利用处理器能力。网络
线程数通常设置为：线程数 = ((线程等待时间 + 线程CPU时间) / 线程CPU时间) * CPU数目数据结构
CPU密集型 VS I/O密集型
咱们能够把任务分为计算密集型和I/O密集型。
计算密集型任务的特色是要进行大量的计算，消耗CPU资源，好比计算圆周率、对视频进行高清解码等等，全靠CPU的运算能力。这种计算密集型任务虽然也能够用多任务完成，可是任务越多，花在任务切换的时间就越多，CPU执行任务的效率就越低，因此，要最高效地利用CPU，计算密集型任务同时进行的数量应当等于CPU的核心数。
计算密集型任务因为主要消耗CPU资源，所以，代码运行效率相当重要。Python这样的脚本语言运行效率很低，彻底不适合计算密集型任务。对于计算密集型任务，最好用C语言编写。
第二种任务的类型是I/O密集型，涉及到网络、磁盘I/O的任务都是I/O密集型任务，这类任务的特色是CPU消耗不多，任务的大部分时间都在等待I/O操做完成（由于I/O的速度远远低于CPU和内存的速度）。对于I/O密集型任务，任务越多，CPU效率越高，但也有一个限度。常见的大部分任务都是I/O密集型任务，好比Web应用。
I/O密集型任务执行期间，99%的时间都花在I/O上，花在CPU上的时间不多，所以，用运行速度极快的C语言替换用Python这样运行速度极低的脚本语言，彻底没法提高运行效率。对于I/O密集型任务，最合适的语言就是开发效率最高（代码量最少）的语言，脚本语言是首选，C语言最差。
什么是 Fork/Join 框架？
Fork/Join 框架是 Java7 提供了的一个用于并行执行的任务的框架，是一个把大任务分割成若干个小任务，最终汇总每一个小任务结果后获得大任务结果的框架。
Fork 就是把一个大任务切分为若干个子任务并行的执行，Join 就是合并这些子任务的执行结果，最后获得这个大任务的结果。好比计算 1+2+......+10000，能够分割成10个子任务，每一个子任务对1000个数进行求和，最终汇总这10个子任务的结果。以下图所示： 
Fork/Join的特性：
- ForkJoinPool 不是为了替代 ExecutorService，而是它的补充，在某些应用场景下性能比 ExecutorService 更好。（见 Java Tip: When to use ForkJoinPool vs ExecutorService ）
- ForkJoinPool 主要用于实现“分而治之”的算法，特别是分治以后递归调用的函数，例如 quick sort 等；
- ForkJoinPool 最适合的是计算密集型的任务，若是存在 I/O、线程间同步、sleep() 等会形成线程长时间阻塞的状况时，最好配合 MangedBlocker。
关于“分而治之”的算法，能够查看《分治、回溯的实现和特性》
工做窃取算法
工做窃取（work-stealing）算法 是指某个线程从其余队列里窃取任务来执行。
咱们须要作一个比较大的任务，咱们能够把这个任务分割为若干互不依赖的子任务，为了减小线程间的竞争，因而把这些子任务分别放到不一样的队列里，并为每一个队列建立一个单独的线程来执行队列里的任务，线程和队列一一对应，好比A线程负责处理A队列里的任务。
可是有的线程会先把本身队列里的任务干完，而其余线程对应的队列里还有任务等待处理。干完活的线程与其等着，不如去帮其余线程干活，因而它就去其余线程的队列里窃取一个任务来执行。而在这时它们会访问同一个队列，因此为了减小窃取任务线程和被窃取任务线程之间的竞争，一般会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。
工做窃取算法的优势是充分利用线程进行并行计算，并减小了线程间的竞争，其缺点是在某些状况下仍是存在竞争，好比双端队列里只有一个任务时。而且消耗了更多的系统资源，好比建立多个线程和多个双端队列。
- ForkJoinPool 的每一个工做线程都维护着一个工做队列（WorkQueue），这是一个双端队列（Deque），里面存放的对象是任务（ForkJoinTask）。
- 每一个工做线程在运行中产生新的任务（一般是由于调用了 fork()）时，会放入工做队列的队尾，而且工做线程在处理本身的工做队列时，使用的是 LIFO 方式，也就是说每次从队尾取出任务来执行。
- 每一个工做线程在处理本身的工做队列同时，会尝试窃取一个任务（或是来自于刚刚提交到 pool 的任务，或是来自于其余工做线程的工做队列），窃取的任务位于其余线程的工做队列的队首，也就是说工做线程在窃取其余工做线程的任务时，使用的是 FIFO 方式。
- 在遇到 join() 时，若是须要 join 的任务还没有完成，则会先处理其余任务，并等待其完成。
- 在既没有本身的任务，也没有能够窃取的任务时，进入休眠。
Fork/Join的使用
使用场景示例
定义fork/join任务，以下示例，随机生成2000w条数据在数组当中，而后求和_
package com.niuh.forkjoin.recursivetask;
import java.util.concurrent.RecursiveTask;
/** * RecursiveTask 并行计算，同步有返回值 * ForkJoin框架处理的任务基本都能使用递归处理，好比求斐波那契数列等，但递归算法的缺陷是： * 一只会只用单线程处理， * 二是递归次数过多时会致使堆栈溢出； * ForkJoin解决了这两个问题，使用多线程并发处理，充分利用计算资源来提升效率，同时避免堆栈溢出发生。 * 固然像求斐波那契数列这种小问题直接使用线性算法搞定可能更简单，实际应用中彻底不必使用ForkJoin框架， * 因此ForkJoin是核弹，是用来对付你们伙的，好比超大数组排序。 * 最佳应用场景：多核、多内存、能够分割计算再合并的计算密集型任务 */
class LongSum extends RecursiveTask<Long> {
//任务拆分的最小阀值
static final int SEQUENTIAL_THRESHOLD = 1000;
static final long NPS = (1000L * 1000 * 1000);
static final boolean extraWork = true; // change to add more than just a sum
int low;
int high;
int[] array;
LongSum(int[] arr, int lo, int hi) {
array = arr;
low = lo;
high = hi;
}
/** * fork()方法：将任务放入队列并安排异步执行，一个任务应该只调用一次fork()函数，除非已经执行完毕并从新初始化。 * tryUnfork()方法：尝试把任务从队列中拿出单独处理，但不必定成功。 * join()方法：等待计算完成并返回计算结果。 * isCompletedAbnormally()方法：用于判断任务计算是否发生异常。 */
protected Long compute() {
if (high - low <= SEQUENTIAL_THRESHOLD) {
long sum = 0;
for (int i = low; i < high; ++i) {
sum += array[i];
}
return sum;
} else {
int mid = low + (high - low) / 2;
LongSum left = new LongSum(array, low, mid);
LongSum right = new LongSum(array, mid, high);
left.fork();
right.fork();
long rightAns = right.join();
long leftAns = left.join();
return leftAns + rightAns;
}
}
}
复制代码
执行fork/join任务
package com.niuh.forkjoin.recursivetask;
import com.niuh.forkjoin.utils.Utils;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
public class LongSumMain {
//获取逻辑处理器数量
static final int NCPU = Runtime.getRuntime().availableProcessors();
/** * for time conversion */
static final long NPS = (1000L * 1000 * 1000);
static long calcSum;
static final boolean reportSteals = true;
public static void main(String[] args) throws Exception {
int[] array = Utils.buildRandomIntArray(2000000);
System.out.println("cpu-num:" + NCPU);
//单线程下计算数组数据总和
long start = System.currentTimeMillis();
calcSum = seqSum(array);
System.out.println("seq sum=" + calcSum);
System.out.println("singgle thread sort:->" + (System.currentTimeMillis() - start));
start = System.currentTimeMillis();
//采用fork/join方式将数组求和任务进行拆分执行，最后合并结果
LongSum ls = new LongSum(array, 0, array.length);
ForkJoinPool fjp = new ForkJoinPool(NCPU); //使用的线程数
ForkJoinTask<Long> task = fjp.submit(ls);
System.out.println("forkjoin sum=" + task.get());
System.out.println("singgle thread sort:->" + (System.currentTimeMillis() - start));
if (task.isCompletedAbnormally()) {
System.out.println(task.getException());
}
fjp.shutdown();
}
static long seqSum(int[] array) {
long sum = 0;
for (int i = 0; i < array.length; ++i) {
sum += array[i];
}
return sum;
}
}
复制代码
Fork/Join框架原理
Fork/Join 其实就是指由ForkJoinPool做为线程池、ForkJoinTask(一般实现其三个抽象子类)为任务、ForkJoinWorkerThread做为执行任务的具体线程实体这三者构成的任务调度机制。
ForkJoinWorkerThread
ForkJoinWorkerThread 直接继承了Thread，可是仅仅是为了增长一些额外的功能，并无对线程的调度执行作任何更改。 ForkJoinWorkerThread 是被ForkJoinPool管理的工做线程，在建立出来以后都被设置成为了守护线程，由它来执行ForkJoinTasks。该类主要为了维护建立线程实例时经过ForkJoinPool为其建立的任务队列，与其余两个线程池整个线程池只有一个任务队列不一样，ForkJoinPool管理的全部工做线程都拥有本身的工做队列，为了实现任务窃取机制，该队列被设计成一个双端队列，而ForkJoinWorkerThread的首要任务就是执行本身的这个双端任务队列中的任务，其次是窃取其余线程的工做队列，如下是其代码片断：
public class ForkJoinWorkerThread extends Thread {
// 这个线程工做的ForkJoinPool池
final ForkJoinPool pool;
// 这个线程拥有的工做窃取机制的工做队列
final ForkJoinPool.WorkQueue workQueue;
//建立在给定ForkJoinPool池中执行的ForkJoinWorkerThread。
protected ForkJoinWorkerThread(ForkJoinPool pool) {
// Use a placeholder until a useful name can be set in registerWorker
super("aForkJoinWorkerThread");
this.pool = pool;
//向ForkJoinPool执行池注册当前工做线程，ForkJoinPool为其分配一个工做队列
this.workQueue = pool.registerWorker(this);
}
//该工做线程的执行内容就是执行工做队列中的任务
public void run() {
if (workQueue.array == null) { // only run once
Throwable exception = null;
try {
onStart();
pool.runWorker(workQueue); //执行工做队列中的任务
} catch (Throwable ex) {
exception = ex; //记录异常
} finally {
try {
onTermination(exception);
} catch (Throwable ex) {
if (exception == null)
exception = ex;
} finally {
pool.deregisterWorker(this, exception); //撤销工做
}
}
}
}
.....
}
复制代码
ForkJoinTask
ForkJoinTask ：与FutureTask同样， ForkJoinTask也是Future的子类，不过它是一个抽象类。 ForkJoinTask ：咱们要使用 ForkJoin 框架，必须首先建立一个 ForkJoin 任务。它提供在任务中执行 fork() 和 join() 操做的机制，一般状况下咱们不须要直接继承 ForkJoinTask 类，而只须要继承它的子类，Fork/Join框架提供类如下几个子类：
- RecursiveAction：用于没有返回结果的任务。（好比写数据到磁盘，而后就退出。一个 RecursiveAvtion 能够把直接的工做分割成更小的几块，这样它们能够由独立的线程或者 CPU 执行。咱们能够经过继承来实现一个 RecusiveAction）
- RescursiveTask：用于有返回结果的任务。（能够将本身的工做分割为若干更小任务，并将这些子任务的执行合并到一个集体结果。能够有几个水平的分割和合并）
- CountedCompleter ：在任务完成执行后会触发执行一个自定义的钩子函数。
常量介绍
ForkJoinTask 有一个int类型的status字段：
- 其高16位存储任务执行状态例如NORMAL、CANCELLED或EXCEPTIONAL
- 低16位预留用于用户自定义的标记。
任务未完成以前status大于等于0，完成以后就是NORMAL、CANCELLED或EXCEPTIONAL这几个小于0的值，这几个值也是按大小顺序的：0（初始状态） > NORMAL > CANCELLED > EXCEPTIONAL.
public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
/** 该任务的执行状态 */
volatile int status; // accessed directly by pool and workers
static final int DONE_MASK = 0xf0000000; // mask out non-completion bits
static final int NORMAL = 0xf0000000; // must be negative
static final int CANCELLED = 0xc0000000; // must be < NORMAL
static final int EXCEPTIONAL = 0x80000000; // must be < CANCELLED
static final int SIGNAL = 0x00010000; // must be >= 1 << 16
static final int SMASK = 0x0000ffff; // short bits for tags
// 异常哈希表
//被任务抛出的异常数组，为了报告给调用者。由于异常不多见，因此咱们不直接将它们保存在task对象中，而是使用弱引用数组。注意，取消异常不会出如今数组，而是记录在statue字段中
//注意这些都是 static 类属性，全部的ForkJoinTask共用的。
private static final ExceptionNode[] exceptionTable; //异常哈希链表数组
private static final ReentrantLock exceptionTableLock;
private static final ReferenceQueue<Object> exceptionTableRefQueue; //在ForkJoinTask被GC回收以后，相应的异常节点对象的引用队列
/** * 固定容量的exceptionTable. */
private static final int EXCEPTION_MAP_CAPACITY = 32;
//异常数组的键值对节点。
//该哈希链表数组使用线程id进行比较，该数组具备固定的容量，由于它只维护任务异常足够长，以便参与者访问它们，因此在持续的时间内不该该变得很是大。可是，因为咱们不知道最后一个joiner什么时候完成，咱们必须使用弱引用并删除它们。咱们对每一个操做都这样作(所以彻底锁定)。此外，任何ForkJoinPool池中的一些线程在其池变为isQuiescent时都会调用helpExpungeStaleExceptions
static final class ExceptionNode extends WeakReference<ForkJoinTask<?>> {
final Throwable ex;
ExceptionNode next;
final long thrower; // 抛出异常的线程id
final int hashCode; // 在弱引用消失以前存储hashCode
ExceptionNode(ForkJoinTask<?> task, Throwable ex, ExceptionNode next) {
super(task, exceptionTableRefQueue); //在ForkJoinTask被GC回收以后，会将该节点加入队列exceptionTableRefQueue
this.ex = ex;
this.next = next;
this.thrower = Thread.currentThread().getId();
this.hashCode = System.identityHashCode(task);
}
}
.................
}
复制代码
除了status记录任务的执行状态以外，其余字段主要是为了对任务执行的异常的处理，ForkJoinTask采用了哈希数组 + 链表的数据结构（JDK8之前的HashMap实现方法）存放全部（由于这些字段是static）的ForkJoinTask任务的执行异常。
fork 方法（安排任务异步执行）
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;
}
复制代码
该方法其实就是将任务经过push方法加入到当前工做线程的工做队列或者提交队列（外部非ForkJoinWorkerThread线程经过submit、execute方法提交的任务），等待被线程池调度执行，这是一个非阻塞的当即返回方法。
这里须要知道，ForkJoinPool线程池经过哈希数组+双端队列的方式将全部的工做线程拥有的任务队列和从外部提交的任务分别映射到哈希数组的不一样槽位上。
join 方法（等待执行结果）
join() 的工做则复杂得多，也是 join() 可使得线程免于被阻塞的缘由——不像同名的 Thread.join()。
- 检查调用 join() 的线程是不是 ForkJoinThread 线程。若是不是（例如 main 线程），则阻塞当前线程，等待任务完成。若是是，则不阻塞。
- 查看任务的完成状态，若是已经完成，直接返回结果。
- 若是任务还没有完成，但处于本身的工做队列内，则完成它。
- 若是任务已经被其余的工做线程偷走，则窃取这个小偷的工做队列内的任务（以 FIFO 方式），执行，以期帮助它早日完成欲 join 的任务。
- 若是偷走任务的小偷也已经把本身的任务所有作完，正在等待须要 join 的任务时，则找到小偷的小偷，帮助它完成它的任务。
- 递归地执行第5步。
将上述流程画成序列图的话就是这个样子：
源代码以下：
//当计算完成时返回计算结果。此方法与get()的不一样之处在于，异常完成会致使RuntimeException或Error，而不是ExecutionException，调用线程被中断不会经过抛出InterruptedException致使方法忽然返回。
public final V join() {
int s;
if ((s = doJoin() & DONE_MASK) != NORMAL)
reportException(s); //非正常结束，抛出相关的异常堆栈信息
return getRawResult(); //正常结束，返回结果
}
//等待任务执行结束并返回其状态status，该方法实现了join, get, quietlyJoin. 直接处理已经完成的，外部等待和unfork+exec的状况，其它状况转发到ForkJoinPool.awaitJoin
//若是 status < 0 则返回s；
//不然，若不是ForkJoinWorkerThread ，则等待 externalAwaitDone() 返回
//不然，若 (w = (wt = (ForkJoinWorkerThread)t).workQueue).tryUnpush(this) && (s = doExec()) < 0 则 返回s；
//不然，返回 wt.pool.awaitJoin(w, this, 0L)
private int doJoin() {
int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
return (s = status) < 0 ? s : //status为负数表示任务已经执行结束，直接返回status。
((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
(w = (wt = (ForkJoinWorkerThread)t).workQueue).
tryUnpush(this) && (s = doExec()) < 0 ? s : //调用pool的执行逻辑，并等待返回执行结果状态
wt.pool.awaitJoin(w, this, 0L) : //调用pool的等待机制
externalAwaitDone(); //不是ForkJoinWorkerThread，
}
//抛出与给定状态关联的异常(若是有),被取消是CancellationException。
private void reportException(int s) {
if (s == CANCELLED)
throw new CancellationException();
if (s == EXCEPTIONAL)
rethrow(getThrowableException());
}
public abstract V getRawResult();
//返回给定任务的执行异常（若是有的话），为了提供准确的异常堆栈信息，若异常不是由当前线程抛出的，将尝试以记录的异常为缘由建立一个与抛出异常类型相同的新异常。
//若是没有那样的构造方法将尝试使用无参的构造函数，并经过设置initCause方法以达到一样的效果，尽管它可能包含误导的堆栈跟踪信息。
private Throwable getThrowableException() {
if ((status & DONE_MASK) != EXCEPTIONAL)
return null;
//1. 经过当前任务对象的哈希值到哈希链表数组中找到相应的异常节点
int h = System.identityHashCode(this); //当前任务的hash值
ExceptionNode e;
final ReentrantLock lock = exceptionTableLock;
lock.lock(); //加锁
try {
expungeStaleExceptions(); //清理被GC回收的任务的异常节点
ExceptionNode[] t = exceptionTable;
e = t[h & (t.length - 1)]; //经过取模对应得索引获取哈希数组槽位中得节点
while (e != null && e.get() != this)
e = e.next; //遍历找到当前任务对应的异常节点
} finally {
lock.unlock();
}
Throwable ex;
if (e == null || (ex = e.ex) == null) //表示没有出现任何异常
return null;
if (e.thrower != Thread.currentThread().getId()) { //有异常可是不是由当前线程抛出的
Class<? extends Throwable> ec = ex.getClass();
try {
Constructor<?> noArgCtor = null;
Constructor<?>[] cs = ec.getConstructors();// public ctors only
//经过反射找到构造方法，并构造新异常
for (int i = 0; i < cs.length; ++i) {
Constructor<?> c = cs[i];
Class<?>[] ps = c.getParameterTypes();
if (ps.length == 0)
noArgCtor = c; //记录下无参构造方法，以备没有找到指望的构造方法时使用
else if (ps.length == 1 && ps[0] == Throwable.class) {
Throwable wx = (Throwable)c.newInstance(ex); //发现了咱们指望的Throwable类型的参数的构造方法
return (wx == null) ? ex : wx;
}
}
if (noArgCtor != null) { //没有找到指望的构造方法，只能经过无参构造方法建立新异常
Throwable wx = (Throwable)(noArgCtor.newInstance());
if (wx != null) {
wx.initCause(ex); //将原始异常设置进去
return wx;
}
}
} catch (Exception ignore) {
}
}
return ex;
}
//清除哈希链表数组中已经被GC回收掉的任务的异常节点。从exceptionTableRefQueue节点引用队列中获取异常节点并移除哈希链表数组中得对应节点
private static void expungeStaleExceptions() {
for (Object x; (x = exceptionTableRefQueue.poll()) != null;) {
if (x instanceof ExceptionNode) {
int hashCode = ((ExceptionNode)x).hashCode; //节点hash
ExceptionNode[] t = exceptionTable;
int i = hashCode & (t.length - 1); //取模获得哈希表索引
ExceptionNode e = t[i];
ExceptionNode pred = null;
while (e != null) {
ExceptionNode next = e.next;
if (e == x) { //找到了目标节点
if (pred == null)
t[i] = next;
else
pred.next = next;
break;
}
pred = e; //日后遍历链表
e = next;
}
}
}
}
//窃取任务的主要执行方法，除非已经完成了，不然调用exec()并记录完成时的状态。
final int doExec() {
int s; boolean completed;
if ((s = status) >= 0) { //任务还未完成
try {
completed = exec(); 调用exec()并记录完成时的状态。
} catch (Throwable rex) {
return setExceptionalCompletion(rex); //记录异常并返回相关状态，并唤醒经过join等待此任务的线程。
}
if (completed)
s = setCompletion(NORMAL); //更新状态为正常结束，并唤醒经过join等待此任务的线程。
}
return s;
}
//当即执行此任务的基本操做。返回true表示该任务已经正常完成，不然返回false表示此任务不必定完成(或不知道是否完成)。
//此方法还可能抛出(未捕获的)异常，以指示异常退出。此方法旨在支持扩展，通常不该以其余方式调用。
protected abstract boolean exec();
//等待未完成的非ForkJoinWorkerThread线程提交的任务执行结束，并返回任务状态status
private int externalAwaitDone() {
//如果CountedCompleter任务，等待ForkJoinPool.common.externalHelpComplete((CountedCompleter<?>)this, 0) 返回
//不然，若ForkJoinPool.common.tryExternalUnpush(this)，返回 doExec() 结果；
//不然，返回0
int s = ((this instanceof CountedCompleter) ? // try helping
ForkJoinPool.common.externalHelpComplete(
(CountedCompleter<?>)this, 0) : //辅助完成外部提交的CountedCompleter任务
ForkJoinPool.common.tryExternalUnpush(this) ? doExec() : 0); //辅助完成外部提交的非CountedCompleter任务
if (s >= 0 && (s = status) >= 0) { //表示任务还没结束，须要阻塞等待。
boolean interrupted = false;
do {
if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) { //标记有线程须要被唤醒
synchronized (this) {
if (status >= 0) {
try {
wait(0L); //任务还没结束，无限期阻塞直到被唤醒
} catch (InterruptedException ie) {
interrupted = true;
}
}
else
notifyAll(); //已经结束了唤醒全部阻塞的线程
}
}
} while ((s = status) >= 0);
if (interrupted)
Thread.currentThread().interrupt(); //恢复中断标识
}
return s;
}
//记录异常，更新status状态，唤醒全部等待线程
private int setExceptionalCompletion(Throwable ex) {
int s = recordExceptionalCompletion(ex);
if ((s & DONE_MASK) == EXCEPTIONAL)
internalPropagateException(ex); //调用钩子函数传播异常
return s;
}
/** * 对任务异常结束的异常传播支持的钩子函数 */
void internalPropagateException(Throwable ex) {
}
//记录异常并设置状态status
final int recordExceptionalCompletion(Throwable ex) {
int s;
if ((s = status) >= 0) {
int h = System.identityHashCode(this); //哈希值
final ReentrantLock lock = exceptionTableLock;
lock.lock(); //加锁
try {
expungeStaleExceptions();
ExceptionNode[] t = exceptionTable;
int i = h & (t.length - 1);
for (ExceptionNode e = t[i]; ; e = e.next) {
if (e == null) { //遍历完了都没找到，说明哈希链表数组中不存在该任务对于的异常节点
t[i] = new ExceptionNode(this, ex, t[i]); //建立一个异常节点用头插法插入哈希链表数组
break;
}
if (e.get() == this) // 哈希链表数组中已经存在相应的异常节点，退出
break;
}
} finally {
lock.unlock();
}
s = setCompletion(EXCEPTIONAL);
}
return s;
}
//标记任务完成标志，并唤醒经过join等待此任务的线程。
private int setCompletion(int completion) {
for (int s;;) {
if ((s = status) < 0)
return s;
if (U.compareAndSwapInt(this, STATUS, s, s | completion)) { //更新状态
if ((s >>> 16) != 0)
synchronized (this) { notifyAll(); } //唤醒全部等待线程
return completion;
}
}
}
复制代码
get 方法（获取异步任务结果）
既然ForkJoinTask也是Future的子类，那么Future最重要的获取异步任务结果的get方法也必然要实现：
//若是须要，等待计算完成，而后检索其结果。
public final V get() throws InterruptedException, ExecutionException {
int s = (Thread.currentThread() instanceof ForkJoinWorkerThread) ? doJoin() : //是ForkJoinWorkerThread，执行doJoin
externalInterruptibleAwaitDone(); //执行externalInterruptibleAwaitDone
Throwable ex;
if ((s &= DONE_MASK) == CANCELLED)
throw new CancellationException(); //被取消的抛出CancellationException
if (s == EXCEPTIONAL && (ex = getThrowableException()) != null)
throw new ExecutionException(ex); //执行中出现异常的抛出相应的异常
return getRawResult(); //返回正常结果
}
//阻塞非ForkJoinWorkerThread线程，直到完成或中断。
private int externalInterruptibleAwaitDone() throws InterruptedException {
int s;
if (Thread.interrupted())
throw new InterruptedException();
if ((s = status) >= 0 &&
(s = ((this instanceof CountedCompleter) ?
ForkJoinPool.common.externalHelpComplete(
(CountedCompleter<?>)this, 0) :
ForkJoinPool.common.tryExternalUnpush(this) ? doExec() :
0)) >= 0) { //根据不一样的任务类型 返回执行或暂时等待被执行的状态
while ((s = status) >= 0) { //须要阻塞等待
if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
synchronized (this) {
if (status >= 0)
wait(0L); //阻塞等待
else
notifyAll(); //唤醒全部等待线程
}
}
}
}
return s;
}
复制代码
get方法也是经过实现join方法的doJoin方法实现的，不一样的是，调用get方法的线程若是被中断的话，get方法会当即抛出InterruptedException异常，而join方法则不会；另外任务异常完成的的相关异常，get方法会将相关异常都封装成ExecutionException异常，而join方法则是原样抛出相关的异常不会被封装成ExecutionException异常。get方法采用的wait/notifyAll这种线程通讯机制来实现阻塞与唤醒。另外还有超时版本的get方法也相似，因而可知get支持可中断和/或定时等待完成。
invoke 方法（当即执行任务，并等待返回结果）
//开始执行此任务，若是须要等待其完成，并返回其结果，若是底层执行此任务时出现异常，则抛出相应的(未捕获的)RuntimeException或Error。
public final V invoke() {
int s;
if ((s = doInvoke() & DONE_MASK) != NORMAL)
reportException(s);
return getRawResult();
}
// invoke, quietlyInvoke的实现
private int doInvoke() {
int s; Thread t; ForkJoinWorkerThread wt;
return (s = doExec()) < 0 ? s : //执行此任务，完成返回其status
((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? //若未完成或须要等待就根据不一样任务类型执行不一样的等待逻辑
(wt = (ForkJoinWorkerThread)t).pool.
awaitJoin(wt.workQueue, this, 0L) :
externalAwaitDone();
}
复制代码
invoke的实现会利用当前调用invoke的线程当即执行exec方法，固然若是exec方法的实现使用了fork/join，其仍是会利用ForkJoinPool线程池的递归调度执行策略，等待子任务执行完成，一步步的合并成最终的任务结果，并返回。值得注意的是，该方法不会由于线程被中断而当即返回，而必须在等到任务执行有告终果以后才会对中断状态进行补偿。
invokeAll 方法（批量执行任务，并等待它们执行结束）
//执行两个任务
public static void invokeAll(ForkJoinTask<?> t1, ForkJoinTask<?> t2) {
int s1, s2;
t2.fork(); //t2任务交给线程池调度执行
if ((s1 = t1.doInvoke() & DONE_MASK) != NORMAL) //t1任务当即由当前线程执行
t1.reportException(s1); //若t1异常结束，则抛出异常，包括被取消的CancellationException
if ((s2 = t2.doJoin() & DONE_MASK) != NORMAL) //等待t2执行结束
t2.reportException(s2); //若t2异常结束，则抛出异常，包括被取消的CancellationException
}
//执行任务数组
public static void invokeAll(ForkJoinTask<?>... tasks) {
Throwable ex = null;
int last = tasks.length - 1;
for (int i = last; i >= 0; --i) {
ForkJoinTask<?> t = tasks[i];
if (t == null) {
if (ex == null) //都不能为null
ex = new NullPointerException();
}
else if (i != 0)
t.fork(); //除了第一个任务都交给线程池调度执行
else if (t.doInvoke() < NORMAL && ex == null) //由当前线程执行第一个任务
ex = t.getException(); //记录第一个任务的异常
}
for (int i = 1; i <= last; ++i) {
ForkJoinTask<?> t = tasks[i];
if (t != null) {
if (ex != null) //第一个任务异常结束，取消其余全部任务
t.cancel(false);
else if (t.doJoin() < NORMAL) //有任务异常结束，记录异常
ex = t.getException();
}
}
if (ex != null)
rethrow(ex); //如有任务异常结束，抛出数组最前面那个异常结束的任务的异常
}
//批量执行任务，返回每一个任务对应的ForkJoinTask实例，
public static <T extends ForkJoinTask<?>> Collection<T> invokeAll(Collection<T> tasks) {
if (!(tasks instanceof RandomAccess) || !(tasks instanceof List<?>)) {
invokeAll(tasks.toArray(new ForkJoinTask<?>[tasks.size()])); //将任务封装成ForkJoinTask，调用上面那个方法实现
return tasks;
}
//下面的逻辑与上面那个invokeAll也是同样的。
@SuppressWarnings("unchecked")
List<? extends ForkJoinTask<?>> ts = (List<? extends ForkJoinTask<?>>) tasks;
Throwable ex = null;
int last = ts.size() - 1;
for (int i = last; i >= 0; --i) {
ForkJoinTask<?> t = ts.get(i);
if (t == null) {
if (ex == null)
ex = new NullPointerException();
}
else if (i != 0)
t.fork();
else if (t.doInvoke() < NORMAL && ex == null)
ex = t.getException();
}
for (int i = 1; i <= last; ++i) {
ForkJoinTask<?> t = ts.get(i);
if (t != null) {
if (ex != null)
t.cancel(false);
else if (t.doJoin() < NORMAL)
ex = t.getException();
}
}
if (ex != null)
rethrow(ex);
return tasks;
}
复制代码
批量任务的执行其实现都是排在前面的任务（只有两个参数是，第一个参数就是排在前面的任务，是数组或者队列时，索引越小的就是排在越前面的）由当前线程执行，后面的任务交给线程池调度执行，若是有多个任务都出现异常，只会抛出排在最前面那个任务的异常。
quietlyInvoke、quietlyJoin 方法（不须要执行结果的invoke和join）
public final void quietlyJoin() {
doJoin();
}
public final void quietlyInvoke() {
doInvoke();
}
复制代码
quietlyInvoke()，quietlyJoin()这两个方法就仅仅了是调用了doInvoke和doJoin，而后就没有而后了，它们就是不关心执行结果版本的invoke和Join，固然异常结束的也不会将异常抛出来，当执行一组任务而且须要将结果或异常的处理延迟到所有任务完成时，这可能颇有用。
cancel 方法 (尝试取消任务的执行)
public boolean cancel(boolean mayInterruptIfRunning) {
return (setCompletion(CANCELLED) & DONE_MASK) == CANCELLED;
}
复制代码
其主要经过setCompletion标记还没有完成的任务的状态为CANCELLED，并唤醒经过join等待此任务的线程。已经执行完成的任务没法被取消，返回true表示取消成功。注意该方法传入的mayInterruptIfRunning并无使用，所以，ForkJoinTask不支持在取消任务时中断已经开始执行的任务，固然ForkJoinTask的子类能够重写实现。
tryUnfork 方法（取消fork，即从任务队列中移除任务）
//取消任务的执行计划。若是此任务是当前线程最近才刚刚经过fork安排执行，而且还没有在另外一个线程中开始执行，则此方法一般会成功，但也不是100%保证会成功。
public boolean tryUnfork() {
Thread t;
return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) : //针对ForkJoinWorkerThread的取消逻辑
ForkJoinPool.common.tryExternalUnpush(this)); //针对外部提交任务的取消逻辑
}
复制代码
tryUnfork尝试将该任务从任务队列中弹出，弹出以后线程池天然不会再调度该任务。该方法的实现只会在任务刚刚被推入任务队列，而且还处于任务队列的栈顶时才可能会成功，不然100%失败。
reinitialize 方法（从新初始化该任务）
public void reinitialize() {
if ((status & DONE_MASK) == EXCEPTIONAL) //有异常
clearExceptionalCompletion(); //从哈希链表数组中移除当前任务的异常节点，并将status重置为0
else
status = 0;
}
复制代码
若是任务异常结束，会从异常哈希表中清除该任务的异常记录，该方法仅仅是将任务状态status重置为0，使得该任务能够被从新执行。
isDone、isCompletedNormally、isCancelled、isCompletedAbnormally 方法（任务的完成状态查询）
任务的执行状态能够在多个级别上查询：
- 若是任务以任何方式完成(包括任务在未执行的状况下被取消)，则isDone为true。
- 若是任务在没有取消或没有遇到异常的状况下完成，则 isCompletedNormally 为true。
- 若是任务被取消（在这种状况下getException方法返回一个CancellationException），则 isCancelled 为true。
- 若是任务被取消或遇到异常，则isCompletedAbnormally异常为true，在这种状况下，getException将返回遇到的异常或java.util.concurrent.CancellationException。
ForkJoinTask 在执行的时候可能会抛出异常，可是咱们没办法在主线程里直接捕获异常，因此 ForkJoinTask 提供了 isCompletedAbnormally() 方法来检查任务是否已经抛出异常或已经被取消了，而且能够经过 ForkJoinTask 的 getException 方法获取异常。示例以下：
if(task.isCompletedAbnormally()){
System.out.println(task.getException());
}
复制代码
getException 方法返回 Throwable 对象，若是任务被取消了则返回CancellationException。若是任务没有完成或者没有抛出异常则返回 null。
为Runnable和Callable提供的adapt方法
adapt方法主要是为了兼容传统的Runnable和Callable任务，经过adapt方法能够将它们封装成ForkJoinTask任务，当将 ForkJoinTask 与其余类型的任务混合执行时，可使用这些方法。
其余一些方法
getPool能够返回执行该任务的线程所在的线程池实例，inForkJonPool能够断定当前任务是不是由ForkJoinWorkerThread线程提交的，通常来讲这意味着当前任务是内部拆分以后的子任务。
getQueuedTaskCount方法返回已经经过fork安排给当前工做线程执行，但尚未被执行的任务数量，该值是一个瞬间值。由于工做线程调度执行的任务经过fork提交的任务仍是进入的该工做线程的任务队列，所以能够经过该任务得知该值。
其它一些方法：
//可能会在承载当前任务的执行池处于静默（空闲）状态时执行任务。这个方法可能在有不少任务都经过fork被安排执行，可是一个显示的join调用都没有，直到它们都被执行完的设计中使用。
//其实就是若是有一批任务被安排执行，而且不知道它们何时结束，若是但愿在这些任务都执行结束以后再安排一个任务，就可使用helpQuiesce。
public static void helpQuiesce() {
Thread t;
//根据执行线程的不一样类型，调用不一样的静默执行逻辑
if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
wt.pool.helpQuiescePool(wt.workQueue);
}
else
ForkJoinPool.quiesceCommonPool();
}
//返回被当前工做线程持有的任务数a比其它可能窃取其任务的其它工做线程持有的任务数b多多少的估计值，就是 a - b 的差值。若当前工做线程不是在ForkJoinPool中，则返回0
//一般该值被恒定在一个很小的值3，若超过这个阈值，则就在本地处理。
public static int getSurplusQueuedTaskCount() {
return ForkJoinPool.getSurplusQueuedTaskCount();
}
//获取但不移除（即不取消执行计划）安排给当前线程的可能即将被执行的下一个任务。但不能保证该任务将在接下来实际被当即执行。该方法可能在即便任务存在但由于竞争而不可访问而返回null
//该方法主要是为了支持扩展，不然可能不会被使用。
protected static ForkJoinTask<?> peekNextLocalTask() {
Thread t; ForkJoinPool.WorkQueue q;
if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
q = ((ForkJoinWorkerThread)t).workQueue;
else
q = ForkJoinPool.commonSubmitterQueue();
return (q == null) ? null : q.peek();
}
//获取而且移除（即取消执行）安排给当前线程的可能即将被执行的下一个任务。
//该方法主要是为了支持扩展，不然可能不会被使用。
protected static ForkJoinTask<?> pollNextLocalTask() {
Thread t;
return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
((ForkJoinWorkerThread)t).workQueue.nextLocalTask() :
null;
}
//若是当前线程被ForkJoinPool运行，获取而且移除（即取消执行）当前线程即将可能执行的下一个任务。该任务多是从其它线程中窃取来的。
//返回nulll并不必定意味着此任务正在操做的ForkJoinPool处于静止状态。该方法主要是为了支持扩展，不然可能不会被使用。
protected static ForkJoinTask<?> pollTask() {
Thread t; ForkJoinWorkerThread wt;
return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
(wt = (ForkJoinWorkerThread)t).pool.nextTaskFor(wt.workQueue) :
null;
}
复制代码
小结
一般ForkJoinTask只适用于非循环依赖的纯函数的计算或孤立对象的操做，不然，执行可能会遇到某种形式的死锁，由于任务循环地等待彼此。可是，这个框架支持其余方法和技术(例如使用Phaser、helpQuiesce和complete)，这些方法和技术可用于构造解决这种依赖任务的ForkJoinTask子类，为了支持这些用法，可使用setForkJoinTaskTag或compareAndSetForkJoinTaskTag原子性地标记一个short类型的值，并使用getForkJoinTaskTag进行检查。ForkJoinTask实现没有将这些受保护的方法或标记用于任何目的，可是它们能够用于构造专门的子类，由此可使用提供的方法来避免从新访问已经处理过的节点/任务。
ForkJoinTask应该执行相对较少的计算，而且应该避免不肯定的循环。大任务应该被分解成更小的子任务，一般经过递归分解。若是任务太大，那么并行性就不能提升吞吐量。若是过小，那么内存和内部任务维护开销可能会超过处理开销。
ForkJoinTask是可序列化的，这使它们可以在诸如远程执行框架之类的扩展中使用。只在执行以前或以后序列化任务才是明智的，而不是在执行期间。
ForkJoinPool
ForkJoinPool：ForkJoinTask 须要经过 ForkJoinPool 来执行，任务分割出的子任务会添加到当前工做线程所维护的双端队列中，进入队列的头部。当一个工做线程的队列里暂时没有任务时，它会随机从其余工做线程的队列的尾部获取一个任务。
常量介绍
ForkJoinPool 与 内部类 WorkQueue 共享的一些常量
// Constants shared across ForkJoinPool and WorkQueue
// 限定参数
static final int SMASK = 0xffff; // 低位掩码，也是最大索引位
static final int MAX_CAP = 0x7fff; // 工做线程最大容量
static final int EVENMASK = 0xfffe; // 偶数低位掩码
static final int SQMASK = 0x007e; // workQueues 数组最多64个槽位
// ctl 子域和 WorkQueue.scanState 的掩码和标志位
static final int SCANNING = 1; // 标记是否正在运行任务
static final int INACTIVE = 1 << 31; // 失活状态 负数
static final int SS_SEQ = 1 << 16; // 版本戳，防止ABA问题
// ForkJoinPool.config 和 WorkQueue.config 的配置信息标记
static final int MODE_MASK = 0xffff << 16; // 模式掩码
static final int LIFO_QUEUE = 0; // LIFO队列
static final int FIFO_QUEUE = 1 << 16; // FIFO队列
static final int SHARED_QUEUE = 1 << 31; // 共享模式队列，负数 ForkJoinPool 中的相关常量和实例字段:
复制代码
ForkJoinPool 中的相关常量和实例字段
// 低位和高位掩码
private static final long SP_MASK = 0xffffffffL;
private static final long UC_MASK = ~SP_MASK;
// 活跃线程数
private static final int AC_SHIFT = 48;
private static final long AC_UNIT = 0x0001L << AC_SHIFT; //活跃线程数增量
private static final long AC_MASK = 0xffffL << AC_SHIFT; //活跃线程数掩码
// 工做线程数
private static final int TC_SHIFT = 32;
private static final long TC_UNIT = 0x0001L << TC_SHIFT; //工做线程数增量
private static final long TC_MASK = 0xffffL << TC_SHIFT; //掩码
private static final long ADD_WORKER = 0x0001L << (TC_SHIFT + 15); // 建立工做线程标志
// 池状态
private static final int RSLOCK = 1;
private static final int RSIGNAL = 1 << 1;
private static final int STARTED = 1 << 2;
private static final int STOP = 1 << 29;
private static final int TERMINATED = 1 << 30;
private static final int SHUTDOWN = 1 << 31;
// 实例字段
volatile long ctl; // 主控制参数
volatile int runState; // 运行状态锁
final int config; // 并行度|模式
int indexSeed; // 用于生成工做线程索引
volatile WorkQueue[] workQueues; // 主对象注册信息，workQueue
final ForkJoinWorkerThreadFactory factory;// 线程工厂
final UncaughtExceptionHandler ueh; // 每一个工做线程的异常信息
final String workerNamePrefix; // 用于建立工做线程的名称
volatile AtomicLong stealCounter; // 偷取任务总数，也可做为同步监视器
/** 静态初始化字段 */
//线程工厂
public static final ForkJoinWorkerThreadFactory defaultForkJoinWorkerThreadFactory;
//启动或杀死线程的方法调用者的权限
private static final RuntimePermission modifyThreadPermission;
// 公共静态pool
static final ForkJoinPool common;
//并行度，对应内部common池
static final int commonParallelism;
//备用线程数，在tryCompensate中使用
private static int commonMaxSpares;
//建立workerNamePrefix(工做线程名称前缀)时的序号
private static int poolNumberSequence;
//线程阻塞等待新的任务的超时值(以纳秒为单位)，默认2秒
private static final long IDLE_TIMEOUT = 2000L * 1000L * 1000L; // 2sec
//空闲超时时间，防止timer未命中
private static final long TIMEOUT_SLOP = 20L * 1000L * 1000L; // 20ms
//默认备用线程数
private static final int DEFAULT_COMMON_MAX_SPARES = 256;
//阻塞前自旋的次数，用在在awaitRunStateLock和awaitWork中
private static final int SPINS = 0;
//indexSeed的增量
private static final int SEED_INCREMENT = 0x9e3779b9;
复制代码
ForkJoinPool 的内部状态都是经过一个64位的 long 型 变量ctl来存储，它由四个16位的子域组成:
- AC: 正在运行工做线程数减去目标并行度，高16位
- TC: 总工做线程数减去目标并行度，中高16位
- SS: 栈顶等待线程的版本计数和状态，中低16位
- ID: 栈顶 WorkQueue 在池中的索引(poolIndex)，低16位
ForkJoinPool.WorkQueue 中的相关属性:
//初始队列容量，2的幂
static final int INITIAL_QUEUE_CAPACITY = 1 << 13;
//最大队列容量
static final int MAXIMUM_QUEUE_CAPACITY = 1 << 26; // 64M
// 实例字段
volatile int scanState; // Woker状态, <0: inactive; odd:scanning
int stackPred; // 记录前一个栈顶的ctl
int nsteals; // 偷取任务数
int hint; // 记录偷取者索引，初始为随机索引
int config; // 池索引和模式
volatile int qlock; // 1: locked, < 0: terminate; else 0
volatile int base; // 下一个poll操做的索引(栈底/队列头)
int top; // 一个push操做的索引(栈顶/队列尾)
ForkJoinTask<?>[] array; // 任务数组
final ForkJoinPool pool; // the containing pool (may be null)
final ForkJoinWorkerThread owner; // 当前工做队列的工做线程，共享模式下为null
volatile Thread parker; // 调用park阻塞期间为owner，其余状况为null
volatile ForkJoinTask<?> currentJoin; // 记录被join过来的任务
volatile ForkJoinTask<?> currentSteal; // 记录从其余工做队列偷取过来的任务
复制代码
内部数据结构
ForkJoinPool采用了哈希数组 + 双端队列的方式存听任务，但这里的任务分为两类：
- 一类是经过execute、submit 提交的外部任务
- 另外一类是ForkJoinWorkerThread工做线程经过fork/join分解出来的工做任务
ForkJoinPool并无把这两种任务混在一个任务队列中，对于外部任务，会利用Thread内部的随机probe值映射到哈希数组的偶数槽位中的提交队列中，这种提交队列是一种数组实现的双端队列称之为Submission Queue，专门存放外部提交的任务。
对于ForkJoinWorkerThread工做线程，每个工做线程都分配了一个工做队列，这也是一个双端队列，称之为Work Queue，这种队列都会被映射到哈希数组的奇数槽位，每个工做线程fork/join分解的任务都会被添加到本身拥有的那个工做队列中。
在ForkJoinPool中的属性 WorkQueue[] workQueues 就是咱们所说的哈希数组，其元素就是内部类WorkQueue实现的基于数组的双端队列。该哈希数组的长度为2的幂，而且支持扩容。以下就是该哈希数组的示意结构图：
如图，提交队列位于哈希数组workQueue的奇数索引槽位，工做线程的工做队列位于偶数槽位。
- 默认状况下，asyncMode为false时：
- 所以工做线程把工做队列当着栈同样使用（后进先出），将分解的子任务推入工做队列的top端，取任务的时候也从top端取（凡是双端队列都会有两个分别指向队列两端的指针，这里就是图上画出的base和top）；
- 而当某些工做线程的任务为空的时候，就会从其余队列（不限于workQueue，也会是提交队列）窃取（steal）任务，如图示拥有workQueue2的工做线程从workQueue1中窃取了一个任务，窃取任务的时候采用的是先进先出FIFO的策略(即从base端窃取任务)，这样不但能够避免在取任务的时候与拥有其队列的工做线程发生冲突，从而减少竞争，还能够辅助其完成比较大的任务。
- asyncMode为true的话，拥有该工做队列的工做线程将按照先进先出的策略从base端取任务，这通常只用于不须要返回结果的任务，或者事件消息传递框架。
ForkJoinPool构造函数
其完整构造方法以下
private ForkJoinPool(int parallelism, ForkJoinWorkerThreadFactory factory, UncaughtExceptionHandler handler, int mode, String workerNamePrefix) {
this.workerNamePrefix = workerNamePrefix;
this.factory = factory;
this.ueh = handler;
this.config = (parallelism & SMASK) | mode;
long np = (long)(-parallelism); // offset ctl counts
this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
}
复制代码
重要参数解释
- parallelism：并行度（ the parallelism level），默认状况下跟咱们机器的cpu个数保持一致，使用 Runtime.getRuntime().availableProcessors()能够获得咱们机器运行时可用的CPU个数。
- factory：建立新线程的工厂（ the factory for creating new threads）。默认状况下使用ForkJoinWorkerThreadFactory defaultForkJoinWorkerThreadFactory。
- handler：线程异常状况下的处理器（Thread.UncaughtExceptionHandler handler），该处理器在线程执行任务时因为某些没法预料到的错误而致使任务线程中断时进行一些处理，默认状况为null。
- asyncMode：这个参数要注意，在ForkJoinPool中，每个工做线程都有一个独立的任务队列
asyncMode表示工做线程内的任务队列是采用何种方式进行调度，能够是先进先出FIFO，也能够是后进先出LIFO。若是为true，则线程池中的工做线程则使用先进先出方式进行任务调度，默认状况下是false。
ForkJoinPool.submit 方法
public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
if (task == null)
throw new NullPointerException();
//提交到工做队列
externalPush(task);
return task;
}
复制代码
ForkJoinPool 自身拥有工做队列，这些工做队列的做用是用来接收由外部线程（非 ForkJoinThread 线程）提交过来的任务，而这些工做队列被称为 submitting queue 。 submit() 和 fork() 其实没有本质区别，只是提交对象变成了 submitting queue 而已（还有一些同步，初始化的操做）。submitting queue 和其余 work queue 同样，是工做线程”窃取“的对象，所以当其中的任务被一个工做线程成功窃取时，就意味着提交的任务真正开始进入执行阶段。
相关文章
PS：以上代码提交在 Github ：github.com/Niuh-Study/…
文章持续更新，能够公众号搜一搜「 一角钱技术 」第一时间阅读， 本文 GitHub org_hejianhui/JavaStudy 已经收录，欢迎 Star。
- 1. Forkjoin框架原理解析
- 2. 并发编程-ForkJoin【享学】
- 3. ForkJoin框架之ForkJoinTask
- 4. ForkJoin框架之CompletableFuture
- 5. ForkJoin框架之ForkJoinPool
- 6. Java并发编程与技术内幕：ForkJoin 框架初探
- 7. 并发编程之Disruptor并发框架
- 8. Java并发之ForkJoin
- 9. [转]Java7中的ForkJoin并发框架初探（下）—— ForkJoin的应用
- 10. Java7中的ForkJoin并发框架初探（下）—— ForkJoin的应用
- 更多相关文章...
- • Rust 并发编程 - RUST 教程
- • 高并发系统的分析和设计 - 红包项目实战
- • Git五分钟教程
- • 互联网组织的未来：剖析GitHub员工的任性之源
-
每一个你不满意的现在，都有一个你没有努力的曾经。