[toc]
在前文中介绍了如何使用ForkJoinPool和ForkJoin的一些基本原理。现在继续来分析ForkJoin，原本计划从源码开始分析。但是ForkJoinPool的源码太过复杂。后续得分好几部分来讲解。今天先做一个总体的介绍。
1.ForkJoinPool总体介绍
在java中运行ForkJoinPool，经过对源码的分析，实际上，需要4个类来配合运行。这四个类分别是：
- ForkJoinPool 这是线程池的核心类，也是提供方法供我们使用的入口类。基本上forkJoin的核心操作及线程池的管理方法都由这个类提供。后面将详细介绍。
- ForkJoinPool.WorkQueue 这是ForkJoinPool类的内部类。也是线程池核心的组成部分。ForkJoinPool线程池将由WorkQueue数组组成。为了进一步提高性能，与ThreadPoolExecutor不一样的是，这没有采用外部传入的任务队列，而是作者自己实现了一个阻塞队列。
- ForkJoinWorkerThread 线程池中运行的thread也是作者重新定义的。这个类的目的是在于将外部的各种形式的task都转换为统一的ForkJoinTask格式。
- ForkJoinTask 这是ForkJoinPool支持运行的task抽象类，我们一般使用其子类如RecursiveTask或者RecursiveAction。这个在前文有介绍。这个类提供了任务拆分和结果合并的方法。
2 常量及成员变量
2.1 常量
在ForkJoin中，定义了非常多的常量。
//界限定义
static final int SMASK = 0xffff; // short bits == max index
static final int MAX_CAP = 0x7fff; // max #workers - 1
static final int EVENMASK = 0xfffe; // even short bits
static final int SQMASK = 0x007e; // max 64 (even) slots
// Masks and units for WorkQueue.scanState and ctl sp subfield
static final int SCANNING = 1; // false when running tasks
static final int INACTIVE = 1 << 31; // must be negative
static final int SS_SEQ = 1 << 16; // version count
// Mode bits for ForkJoinPool.config and WorkQueue.config
static final int MODE_MASK = 0xffff << 16; // top half of int
static final int LIFO_QUEUE = 0;
static final int FIFO_QUEUE = 1 << 16;
static final int SHARED_QUEUE = 1 << 31; // must be negative
上述这些界限用二进制的方式表示如下：
常量二进制表示
这些常量实际上是用于后续进行二进制位运算的范围常量。
在面还有一部分：
//定义IDLE的超时时间 默认位2秒
private static final long IDLE_TIMEOUT = 2000L * 1000L * 1000L; // 2sec
//idea的容忍度 20ms
private static final long TIMEOUT_SLOP = 20L * 1000L * 1000L; // 20ms
//静态初始化commonMaxSpares的初始值。这个值远远超出了正常的要求，但也远远低于MAX_CAP和典型的OS线程限制，因此允许jvm在耗尽所需资源之前捕获异常以避免滥用。
private static final int DEFAULT_COMMON_MAX_SPARES = 256;
//阻塞之前等待自旋的次数，这个默认值是0，以降低对CPU的使用。如果大于零，自旋值必须是2的幂次方，至少是4。值2048导致旋转的时间只有典型上下文切换时间的一小部分。
private static final int SPINS = 0;
//这个是产生随机性的魔数，用于scan的时候进行计算
private static final int SEED_INCREMENT = 0x9e3779b9;
下面是进行ctl字段控制的一些字段和掩码：
//ctl的计算字段和掩码
// Lower and upper word masks
private static final long SP_MASK = 0xffffffffL;
private static final long UC_MASK = ~SP_MASK;
// Active counts
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;
// Total counts
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); // sign
上述部分表示位如下二进制：
常量二进制表示
如下为线程池的运行状态。SHUTDOWN的时候为负数
// runState bits: SHUTDOWN must be negative, others arbitrary powers of two
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;
其用二级制的形式表示为如下：
线程池状态
2.2 成员变量
// Instance fields
volatile long ctl; // main pool control
volatile int runState; // lockable status
final int config; // parallelism, mode
int indexSeed; // to generate worker index
volatile WorkQueue[] workQueues; // main registry
final ForkJoinWorkerThreadFactory factory;
final UncaughtExceptionHandler ueh; // per-worker UEH
final String workerNamePrefix; // to create worker name string
volatile AtomicLong stealCounter; // also used as sync monitor
上面是最主要的成员变量，用下表来表示：
| 变量名 | 类型 | 说明 |
|---|---|---|
| ctl | volatile long | 线程池的主要控制字段 |
| runState | volatile int | 线程池的运行状态，值为常量中对应的值 |
| config | final int | 将并行度和mode放到了一个int中，便于后续通过位操作计算 |
| indexSeed | int | 随机种子，通过前面的魔数来实现 |
| workQueues | volatile WorkQueue[] | 组成workQueue的数组。是线程池的核心数据结构 |
| factory | final ForkJoinWorkerThreadFactory | 产生线程的工厂方法 |
| ueh | final UncaughtExceptionHandler | 每个worker出现异常之后的处理办法，类似于前面ThreadPoolExecutor的拒绝策略 |
| workerNamePrefix | final String | 创建线程名称的前缀 |
| stealCounter | volatile AtomicLong | 用于监控steal的计数器 |
3.构造函数
ForkJoinPool提供了3个构造函数：
3.1 ForkJoinPool(parallelism, factory,handler,asyncMode)
public ForkJoinPool(int parallelism,
ForkJoinWorkerThreadFactory factory,
UncaughtExceptionHandler handler,
boolean asyncMode) {
this(checkParallelism(parallelism),
checkFactory(factory),
handler,
asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
"ForkJoinPool-" + nextPoolId() + "-worker-");
checkPermission();
}
此方法没有进行初始化，但是有些字段需要进行说明：
| 属性 | 类型 | 说明 |
|---|---|---|
| parallelism | int | 表示ForkJoinPool支持的最大并行度 |
| factory | ForkJoinWorkerThreadFactory | 用于产生线程的工厂方法 |
| handler | UncaughtExceptionHandler | 用于异常处理的拒绝策略 |
| asyncMode | boolean | 异步模式，如果为ture，则队列将采用FIFO_QUEUE，实现先进先出，反之则LIFO_QUEUE 实现后进先出 |
这个方法中会对并行度和factory进程check，以确保系统能支持。
parallelism最大不允许大于MAX_CAP。最小必须大于0。
factory则只是判断是否为空，如果为空则这个地方会出现NPE异常。
3.2 ForkJoinPool(parallelism,factory,handler,mode,workerNamePrefix)
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 为7
config计算过程
然后再计算np和ctl
np及ctl计算过程
这样就完成了初始化，需要注意的是，这个位运算操作是ForkJoinPool的核心，需要扎实理解。
3.3 ForkJoinPool()
public ForkJoinPool() {
this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
defaultForkJoinWorkerThreadFactory, null, false);
}
无参的构造函数，将通过MAX_CAP和当前系统允许的最大并行度的最小值来指定并行度。使用默认的ThreadFactory。handler位空，默认采用asyncMode为false.
实际实这个方法最终还是调用的上面的两个方法，完成了初始化工作。
4. ForkJoinPool的基本组成
再了解了前面的变量之后，我们可以发现，ForkJoinPool的实际组成是，由一个WorkQueue的数组构成。但是这个数组比较特殊，在偶数位，存放外部调用提交的任务，如通过execute和submit等方法。这种队列称为SubmissionQueue。而另外一种任务是前者在执行过程种通过fork方法产生的新任务。这种队列称为workQueue。SubmissionQueue与WorkQueue的区别在于，WorkQueue的属性“final ForkJoinWorkerThread owner;”是有值的。也就是说，WorkQueue将有ForkJoinWorkerThread线程与之绑定。在运行过程中不断的从WorkQueue中获取任务。如果没有可执行的任务，则将从其他的SubmissionQueue和WorkQueue中窃取任务来执行。前面学习过了工作窃取算法，实际上载WorkQueue上的ForkJoinWorkerThread就是一个窃取者，它从SubmissionQueue中或者去偷的WorkQueue中，按FIFO的方式窃取任务。之后执行。也从自己的WorkQueue中安LIFO或者FIFO的方式执行任务。这取决于模式的设定。默认情况下是采用LIFO的方式在执行。组成如下图所示：
基本组成
5. ForkJoinPool外部任务的提交
ForkJoinPool在初始化之后，支持三种调用方式 invoke 、execute 和submit。我们来对这三种方式进行分析：
5.1 invoke
public <T> T invoke(ForkJoinTask<T> task) {
if (task == null)
throw new NullPointerException();
externalPush(task);
return task.join();
}
可以看到invoke是支持有返回值的调用方法，之后调用externalPush进行push。
5.2 execute
public void execute(ForkJoinTask<?> task) {
if (task == null)
throw new NullPointerException();
externalPush(task);
}
而exec方法，则是没有返回值的调用方法。适用于哪些不需要返回结果的计算。此外还有种变体，支持Runnable。实际上是将Runnable转换位ForkJoinTask。
public void execute(Runnable task) {
if (task == null)
throw new NullPointerException();
ForkJoinTask<?> job;
if (task instanceof ForkJoinTask<?>) // avoid re-wrap
job = (ForkJoinTask<?>) task;
else
job = new ForkJoinTask.RunnableExecuteAction(task);
externalPush(job);
}
5.3 submit
submit于invoke方法基本一致，唯一的不同是，方法内部没有join方法，需要自行定义。
public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
if (task == null)
throw new NullPointerException();
externalPush(task);
return task;
}
这样可能更加灵活。
此外也支持Runnable和Callable，而且Runnable也可以支持返回值。
对于Callable，通过AdaptedCallable进行适配。
public <T> ForkJoinTask<T> submit(Callable<T> task) {
ForkJoinTask<T> job = new ForkJoinTask.AdaptedCallable<T>(task);
externalPush(job);
return job;
}
对于Runnable,如果需要返回值，统一还是返回的ForkJoinTask，之后在外层join得到结果。
public <T> ForkJoinTask<T> submit(Runnable task, T result) {
ForkJoinTask<T> job = new ForkJoinTask.AdaptedRunnable<T>(task, result);
externalPush(job);
return job;
}
也可以采用这种方式：
public ForkJoinTask<?> submit(Runnable task) {
if (task == null)
throw new NullPointerException();
ForkJoinTask<?> job;
if (task instanceof ForkJoinTask<?>) // avoid re-wrap
job = (ForkJoinTask<?>) task;
else
job = new ForkJoinTask.AdaptedRunnableAction(task);
externalPush(job);
return job;
}
实际上通过这些提交方法，我们发现，最关键的入口时externalPush方法。下面我们就对这个方法进行分析。
6.外部提交过程分析
现在开始跟踪externalPush的源码，对外部提交过程进行分析。
6.1 externalPush
这是外部提交的唯一入口。
/**
* 尝试将任务添加到提交者当前的队列中，此方法只处理大多数情况，实际上是根据随机数指定一个workQueues的槽位，如果这个位置存在WorkQueue，则加入队列，然后调用signalWork通知其他工作线程来窃取。反之，则通过externalSubmit处理。这只适用于提交队列存在的普通情况。更复杂的逻辑参考externalSubmit。
*
* @param task the task. Caller must ensure non-null.
*/
final void externalPush(ForkJoinTask<?> task) {
WorkQueue[] ws; WorkQueue q; int m;
//通过ThreadLocalRandom产生随机数
int r = ThreadLocalRandom.getProbe();
//线程池的状态
int rs = runState;
//如果ws已经完成初始化，且根据随机数定位的index存在workQueue,且cas的方式加锁成功
if ((ws = workQueues) != null && (m = (ws.length - 1)) >= 0 &&
//此处先用随机数和wq的size取&，之后再取SQMASK，这些操作将多余的位的值去除
(q = ws[m & r & SQMASK]) != null && r != 0 && rs > 0 &&
//cas的方式加锁 将q中位于QLOCK位置的内存的值如果为0，则改为1，采用cas的方式进行
U.compareAndSwapInt(q, QLOCK, 0, 1)) {
ForkJoinTask<?>[] a; int am, n, s;
//判断q中的task数组是否为空，
if ((a = q.array) != null &&
//am为q的长度 这是个固定值，如果这个值大于n n就是目前队列中的元素，实际实这里是判断队列是否有空余的位置
(am = a.length - 1) > (n = (s = q.top) - q.base)) {
//j实际上是计算添加到workQueue中的index
int j = ((am & s) << ASHIFT) + ABASE;
//将task通过cas的方式插入a的index为j的位置
U.putOrderedObject(a, j, task);
//将队列q的QTOP位置的内存加1，实际上就是将TOP增加1
U.putOrderedInt(q, QTOP, s + 1);
//以可见的方式将q的QLOCK改为0
U.putIntVolatile(q, QLOCK, 0);
//此处，如果队列中的任务小于等于1则通知其他worker来窃取。为什么当任务大于1的时候不通知。而且当没有任务的时候发通知岂不是没有意义？此处不太理解
if (n <= 1)
//这是个重点方法，通知其他worker来窃取
signalWork(ws, q);
return;
}
U.compareAndSwapInt(q, QLOCK, 1, 0);
}
externalSubmit(task);
}
实际上externalPush的逻辑只能处理简单的逻辑，对于其他复杂的逻辑，则需要通过externalPush提供，而这些简单的逻辑，实际上就是添加到任务队列。这个任务队列的索引一定是偶数：i = m & r & SQMASK。
这个计算过程，实际上由于SQMASK最后一位为0，因此计算的index的最后一位一定为0，这样导致这个值为偶数。也就是说，workQueues的偶数位存放的是外部提交的任务队列。之后提交成功之后，调用signalWork方法让其他的worker来窃取。
6.2 externalSubmit
这是提交过程中的分支逻辑处理的方法。
/**
*externalPush的完整版本，处理哪些不常用的逻辑。如第一次push的时候进行初始化、此外如果索引队列为空或者被占用，那么创建一个新的任务队列。
*
* @param task the task. Caller must ensure non-null.
*/
private void externalSubmit(ForkJoinTask<?> task) {
//r是随机数，此处双重检测，确保r不为0
int r; // initialize caller's probe
if ((r = ThreadLocalRandom.getProbe()) == 0) {
ThreadLocalRandom.localInit();
r = ThreadLocalRandom.getProbe();
}
//死循环
for (;;) {
WorkQueue[] ws; WorkQueue q; int rs, m, k;
//move默认为false
boolean move = false;
//如果runstate小于0 则线程池处于SHUTDOWN状态，配合进行终止
if ((rs = runState) < 0) {
//终止的方法 并抛出异常，拒绝该任务
tryTerminate(false, false); // help terminate
throw new RejectedExecutionException();
}
//如果状态不为STARTED 说明此时线程池可用
else if ((rs & STARTED) == 0 || // initialize
//如果workQueues为null 或者其length小于1 则说明没用初始化
((ws = workQueues) == null || (m = ws.length - 1) < 0)) {
int ns = 0;
//对线程池以CAS的方式加锁，从RUNSTATE变为RSLOCK，如果不为RUNSTATE则自旋
rs = lockRunState();
try {
//如果状态为 RSIGNAL RSLOCK 说明加锁成功
if ((rs & STARTED) == 0) {
//用cas的方式初始化STEALCOUNTER
U.compareAndSwapObject(this, STEALCOUNTER, null,
new AtomicLong());
// create workQueues array with size a power of two
//创建workQueues的数组
//根据并行度计算得到config，此处确保p在SMASK范围内，即2个字节
int p = config & SMASK; // ensure at least 2 slots
//n判断p是否大于1，反之则默认按1处理
int n = (p > 1) ? p - 1 : 1;
//下列过程是找到大于n的最小的2的幂 这个过程之前在HashMap中演示过
n |= n >>> 1; n |= n >>> 2; n |= n >>> 4;
n |= n >>> 8; n |= n >>> 16; n = (n + 1) << 1;
//根据并行度计算得到了n,之后根据n确定workQueues的array的大小，这个数组的大小不会超过2^16
workQueues = new WorkQueue[n];
//将ns的值修改为STARTED
ns = STARTED;
}
} finally {
//最后将状态解锁 此时改为STARTED状态，这个计算过程有一点绕
unlockRunState(rs, (rs & ~RSLOCK) | ns);
}
//实际上这个分支只是创建了外层的workQueues数组，此时数组内的内容还是全部都是空的
}
//如果根据随机数计算出来的槽位不为空，即索引处的队列已经创建，这个地方是外层死循环再次进入的结果
//需要注意的是这个k的计算过程，SQMASK最低的位为0，这样就导致，无论随机数r怎么变化，得到的结果总是偶数。
else if ((q = ws[k = r & m & SQMASK]) != null) {
//如果这个槽位的workQueue未被锁定，则用cas的方式加锁 将其改为1
if (q.qlock == 0 && U.compareAndSwapInt(q, QLOCK, 0, 1)) {
//拿到这个队列中的array
ForkJoinTask<?>[] a = q.array;
//s为top索引
int s = q.top;
//初始化submitted状态
boolean submitted = false; // initial submission or resizing
try { // locked version of push
//与上面的externalPush一致，此处push到队列中
//先判断 数组不为空且数组中有空余位置，能够容纳这个task
if ((a != null && a.length > s + 1 - q.base) ||
//或者通过初始化的双端队列的数组不为null
(a = q.growArray()) != null) {
//计算数组的index
int j = (((a.length - 1) & s) << ASHIFT) + ABASE;
//在索引index处插入task
U.putOrderedObject(a, j, task);
//将队列的QTOP加1
U.putOrderedInt(q, QTOP, s + 1);
//将提交成功状态改为true
submitted = true;
}
} finally {
//最终采用cas的方式进行解锁 将队列的锁定状态改为0
U.compareAndSwapInt(q, QLOCK, 1, 0);
}
//如果submitted为true说明数据添加成功，此时调用其他worker来窃取
if (submitted) {
//调用窃取的方法
signalWork(ws, q);
//退出
return;
}
}
//move状态改为true
move = true; // move on failure
}
//如果状态不为RSLOCK 上面两个分支都判断过了，那么此处说明这个索引位置没有初始化
else if (((rs = runState) & RSLOCK) == 0) { // create new queue
/new一个新队列
q = new WorkQueue(this, null);
//hint 记录随机数
q.hint = r;
//计算config SHARED_QUEUE 将确保第一位为1 则这个计算出来的config是负数，这与初始化的方法是一致的
q.config = k | SHARED_QUEUE;
//将scan状态改为INCATIVE
q.scanState = INACTIVE;
//用cas的方式加锁
rs = lockRunState(); 将创建的workQueue push到workQueues的数组中
// publish index
if (rs > 0 && (ws = workQueues) != null &&
k < ws.length && ws[k] == null)
//赋值
ws[k] = q; // else terminated
//解锁
unlockRunState(rs, rs & ~RSLOCK);
}
else
//将move改为true
move = true; // move if busy
if (move)
//重新计算r
r = ThreadLocalRandom.advanceProbe(r);
}
}
这个地方将产生的无owoner的workQueue放置在索引k的位置，需要注意的是k的计算过程，k= r & m & SQMASK。r是随机数，m是数组的长度，而SQMASK：
SQMASK
最后一位不为1，这就导致不管r如何变化，得到的k最后一位都不为1，这就构造了一个偶数k最后一位为0，k不可能是奇数。
6.3 signalWork
上述两个方法如果提交成功，那么调用signalWork，通知工作线程运行。
/**
* 此处将激活worker Thread。如果工作线程太少则创建，反之则来进行窃取。
*
* @param ws the worker array to use to find signallees
* @param q a WorkQueue --if non-null, don't retry if now empty
*/
final void signalWork(WorkQueue[] ws, WorkQueue q) {
long c; int sp, i; WorkQueue v; Thread p;
//如果ctl为负数 ctl初始化的时候就会为负数 如果小于0 说明有任务需要处理
while ((c = ctl) < 0L) { // too few active
//c为long，强转int 32位的高位都丢弃，此时如果没有修改过ctl那么低位一定为0 可参考前面ctl的推算过程，所以此处sp 为0 sp为0则说明每月空闲的worker
if ((sp = (int)c) == 0) { // no idle workers
//还是拿c与ADD_WORKER取& 如果不为0 则说明worker太少，需要新增worker
if ((c & ADD_WORKER) != 0L) // too few workers
//通过tryAddWorker 新增worker
tryAddWorker(c);
break;
}
//再次缺认ws有没有被初始化 如果没有 退出
if (ws == null) // unstarted/terminated
break;
//如果ws的length小于sp的最低位 退出
if (ws.length <= (i = sp & SMASK)) // terminated
break;
//如果index处为空 退出
if ((v = ws[i]) == null) // terminating
break;
//将sp的低32位取出
int vs = (sp + SS_SEQ) & ~INACTIVE; // next scanState
//计算用sp减去 scanState
int d = sp - v.scanState; // screen CAS
long nc = (UC_MASK & (c + AC_UNIT)) | (SP_MASK & v.stackPred);
//采用cas的方式修改ctl 实际上就是加锁 由于ctl的修改可能会导致while循环退出
if (d == 0 && U.compareAndSwapLong(this, CTL, c, nc)) {
v.scanState = vs; // activate v
//如果p被park wait中
if ((p = v.parker) != null)
//将worker唤醒
U.unpark(p);
//退出
break;
}
//如果此队列为空或者没有task 也退出
if (q != null && q.base == q.top) // no more work
break;
}
}
可以看到，实际上signalWork就做了两件事情，第一，判断worker是否充足，如果不够，则创建新的worker。第二，判断worker的状态是否被park了，如果park则用unpark唤醒。这样worker就可以取scan其他队列进行窃取了。
6.4 tryAddWorker
此方法是在worker不足的时候，添加一个worker来执行的具体类。
/**
* 尝试新增一个worker，然后增加ctl中记录的worker的数量
*
* @param c incoming ctl value, with total count negative and no
* idle workers. On CAS failure, c is refreshed and retried if
* this holds (otherwise, a new worker is not needed).
*/
private void tryAddWorker(long c) {
//传入的c为外层调用方法的ctl add标记为false
boolean add = false;
do {
long nc = ((AC_MASK & (c + AC_UNIT)) |
(TC_MASK & (c + TC_UNIT)));
//如果此时ctl与外层传入的ctl相等 说明没有被修改
if (ctl == c) {
int rs, stop; // check if terminating
//用cas的方式加锁
if ((stop = (rs = lockRunState()) & STOP) == 0)
//增加ctl的数量，如果成功 add为ture
add = U.compareAndSwapLong(this, CTL, c, nc);
//解锁
unlockRunState(rs, rs & ~RSLOCK);
//如果stop不为0 则说明线程池停止 退出
if (stop != 0)
break;
//如果前面增加ctl中的数量成功，那么此处开始创建worker
if (add) {
createWorker();
break;
}
}
//这个while循环， 前半部分与ADD_WORKER取并，最终只会保留第48位，这个位置为1，同时c的低32为为0，
} while (((c = ctl) & ADD_WORKER) != 0L && (int)c == 0);
}
实际上这个类中，只是做了一些准备过程，增加count，以及加锁判断，最终还是通过createWorker来进行。
6.5 createWorker
创建worker的具体方法。
// Creating, registering and deregistering workers
/**
* 创建并启动一个worker，因为前面已经做了增加count，如果此处出现异常，创建worker不成功，则在deregisterWorker中会判断如果ex不为空，且当前为创建状态的话，会重新进入tryAddWorker方法。
*
* @return true if successful
*/
private boolean createWorker() {
//创建线程的工厂方法
ForkJoinWorkerThreadFactory fac = factory;
Throwable ex = null;
ForkJoinWorkerThread wt = null;
try {
//如果工厂方法不为空，则用这个工厂方法创建线程，之后再启动线程，此时newThread将与workQueue绑定
if (fac != null && (wt = fac.newThread(this)) != null) {
wt.start();
return true;
}
//如果创建失败，出现了异常 则ex变量有值
} catch (Throwable rex) {
ex = rex;
}
deregisterWorker(wt, ex);
return false;
}
此方法只是创建了一个ForkJoinThread,实际上worker还是没有创建。实际上这个创建过程是再newThread(this)中来进行的。
ForkJoinWorkerThread(ForkJoinPool pool, ThreadGroup threadGroup,
AccessControlContext acc) {
super(threadGroup, null, "aForkJoinWorkerThread");
U.putOrderedObject(this, INHERITEDACCESSCONTROLCONTEXT, acc);
eraseThreadLocals(); // clear before registering
this.pool = pool;
this.workQueue = pool.registerWorker(this);
}
再线程创建成功之后调用registerWorker与之绑定。
如果线程创建失败或者出现异常就要调用deregisterWorker对count进行清理或者解除绑定。
6.6 registerWorker
实际上是这个方法，完成worker的创建和绑定。
/**
* 创建线程，并建立线程与workQueue的关系，此处只会在workQueues数组的奇数位操作
*
* @param wt the worker thread
* @return the worker's queue
*/
final WorkQueue registerWorker(ForkJoinWorkerThread wt) {
UncaughtExceptionHandler handler;
//将线程设置为守护线程
wt.setDaemon(true); // configure thread
//如果没有handler则抛出异常
if ((handler = ueh) != null)
wt.setUncaughtExceptionHandler(handler);
//创建一个workQueue，此时owoner为当前输入的ForkJoinThread
WorkQueue w = new WorkQueue(this, wt);
//定义i为0
int i = 0; // assign a pool index
//定义mode
int mode = config & MODE_MASK;
//加锁
int rs = lockRunState();
try {
WorkQueue[] ws; int n; // skip if no array
//如果workQueues存在，且长度大于0
if ((ws = workQueues) != null && (n = ws.length) > 0) {
//通过魔数计算
int s = indexSeed += SEED_INCREMENT; // unlikely to collide
//m为n-1,而n为数组的初始化长度，第一次创建的时候，n=16，那么m为15
int m = n - 1;
//将s左移然后最后一位补上1，之后与奇数m求并集，那么得到的结果必然是奇数
i = ((s << 1) | 1) & m; // odd-numbered indices
//判断i位置是否为空 如果为空，出现了碰撞，则计算步长向后移动 这个步长一定是偶数
if (ws[i] != null) { // collision
int probes = 0; // step by approx half n
//最小步长为2 n是数组长度，比为偶数，那么如果n小于等于4，则步长为2，反之，则将n右移，将偶数最后一位的0清除，之后再和EVENMASK求并，这样就相当于将原来的长度缩小2倍，并确保是偶数。之后再加上2。那么假定n为16的话，此处计算的step就为10
int step = (n <= 4) ? 2 : ((n >>> 1) & EVENMASK) + 2;
//之后再通过while循环，继续判断增加步长之后是否碰撞，如果碰撞，则继续增加步长
while (ws[i = (i + step) & m] != null) {
//如果还是碰撞，且probes增加1之后大于长度n，则会触发扩容，workQueues会扩大2倍 这个probes感觉意义不大
if (++probes >= n) {
workQueues = ws = Arrays.copyOf(ws, n <<= 1);
m = n - 1;
//将probes置为0
probes = 0;
}
}
}
//设置随机数seed
w.hint = s; // use as random seed
//修改config
w.config = i | mode;
//修改scanState为i
w.scanState = i; // publication fence
//将w设置到i处
ws[i] = w;
}
} finally {
//cas的方式进行解锁
unlockRunState(rs, rs & ~RSLOCK);
}
//此处设置线程name
wt.setName(workerNamePrefix.concat(Integer.toString(i >>> 1)));
return w;
}
再registerWorker中，对工作线程和workQueue进行绑定，并设置到workQueues的奇数索引处。注意这里计算得到奇数索引的算法。
//将s左移然后最后一位补上1，之后与奇数m求并集，那么得到的结果必然是奇数。
i = ((s << 1) | 1) & m;
这个位操作，是我们非常值得注意的地方。worker创建成功，之后再将
6.7 deregisterWorker
如果线程创建没有成功，那么count需要回收。以及进行一些清理工作。
/**
*此方法的主要目的是在创建worker或者启动worker失败之后的回调方法，此时将之前的ctl中增加的count减去。
*
* @param wt the worker thread, or null if construction failed
* @param ex the exception causing failure, or null if none
*/
final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) {
WorkQueue w = null;
//如果workQueue和thread不为空
if (wt != null && (w = wt.workQueue) != null) {
WorkQueue[] ws; // remove index from array
//根据config计算index
int idx = w.config & SMASK;
//加锁
int rs = lockRunState();
如果ws不为空且length大于idx同时idx处的worker就是workerQueue 则将该idx处的worker移除
if ((ws = workQueues) != null && ws.length > idx && ws[idx] == w)
ws[idx] = null;
//解锁 修改rs状态
unlockRunState(rs, rs & ~RSLOCK);
}
//后续对count减少
long c; // decrement counts
//死循环 cas的方式将ctl修改
do {} while (!U.compareAndSwapLong
(this, CTL, c = ctl, ((AC_MASK & (c - AC_UNIT)) |
(TC_MASK & (c - TC_UNIT)) |
(SP_MASK & c))));
//入果workQueue不为空 将其中的task取消
if (w != null) {
w.qlock = -1; // ensure set
w.transferStealCount(this);
w.cancelAll(); // cancel remaining tasks
}
//死循环 如果为停止状态则配合停止
for (;;) { // possibly replace
WorkQueue[] ws; int m, sp;
if (tryTerminate(false, false) || w == null || w.array == null ||
(runState & STOP) != 0 || (ws = workQueues) == null ||
(m = ws.length - 1) < 0) // already terminating
break;
if ((sp = (int)(c = ctl)) != 0) { // wake up replacement
if (tryRelease(c, ws[sp & m], AC_UNIT))
break;
}
else if (ex != null && (c & ADD_WORKER) != 0L) {
tryAddWorker(c); // create replacement
break;
}
else // don't need replacement
break;
}
//异常处理
if (ex == null) // help clean on way out
ForkJoinTask.helpExpungeStaleExceptions();
else // rethrow
ForkJoinTask.rethrow(ex);
}
至此，我们通过外部线程push之后，将任务提交到submissionQueue队列之后，会根据并行度以及工作线程的需要创建workQueue。唤醒工作线程进行窃取的操作就完成了，外部线程的调用就结束了，会回到main方法中去等待结果。
6.8 外部线程提交过程总结
上述代码调用的过程，我们可以形象的用如下图进行表示：
最开始，workQueues是null状态。在第一次执行的时候，externalSubmit方法中会初始化这个数组。
数组初始化
在这之后，还是在externalSubmit方法的for循环中，完成对任务队列的创建，将任务队列创建在偶数索引处。之后将任务写入这个队列：
创建任务队列
此后，任务添加完成，但是没有工作队列进行工作。因此在这之后调用signalWork，通知工作队列开始干活。但是在这个方法执行的过程中，由于工作队列并不存在，没有worker，所以调用tryAddWorker开始创建worker。在createWorker会创建一个worker线程：
创建workThread
但是workQueue还没创建。这是在newthread之后，通过registerWorker创建的，在registerWorker方法中，会在奇数位创建一个workQueue，并将此前创建的线程与之绑定。这样一个worker就成功创建了。
绑定workQueue
这样就完成了worker创建的全过程。
7.workQueue的工作过程
在workQueue创建完成之后，下一步，这些线程的run方法调用后被启动。之后就进入了worker线程的生命周期了。实际上run方方法如下：
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);
}
}
}
}
重点执行的式runWorker，此时一旦出错，可以调用deregisterWorker方法进行清理。下面来看看runWorker的详细过程。
7.1 runWorker
这是worker工作线程的执行方法。通过死循环，不断scan是否有任务，之后窃取这个任务进行执行。
/**
* 通过调用线程的run方法，此时开始最外层的runWorker
*/
final void runWorker(WorkQueue w) {
//初始化队列，这个方法会根据任务进行判断是否需要扩容
w.growArray(); // allocate queue
//hint是采用的魔数的方式增加
int seed = w.hint; // initially holds randomization hint
//如果seed为0 则使用1
int r = (seed == 0) ? 1 : seed; // avoid 0 for xorShift
//死循环
for (ForkJoinTask<?> t;;) {
//调用scan方法 对经过魔数计算的r 之后开始进行窃取过程 如果能够窃取 则task不为空
if ((t = scan(w, r)) != null)
//运行窃取之后的task
w.runTask(t);
//反之则当前线程进行等待
else if (!awaitWork(w, r))
break;
r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift
}
}
通过scan方法从其他队列获得任务。
7.2 scan
/**
* 通过scan方法进行任务窃取，扫描从一个随机位置开始，如果出现竞争则通过魔数继续随机移动，反之则线性移动，直到所有队列上的相同校验连续两次出现为空，则说明没有任何任务可以窃取，因此worker会停止窃取，之后重新扫描，如果找到任务则重新激活，否则返回null，扫描工作应该尽可能少的占用内存，以减少对其他扫描线程的干扰。
*
* @param w the worker (via its WorkQueue)
* @param r a random seed
* @return a task, or null if none found
*/
private ForkJoinTask<?> scan(WorkQueue w, int r) {
WorkQueue[] ws; int m;
//如果workQueues不为空且长度大于1，当前workQueue不为空
if ((ws = workQueues) != null && (m = ws.length - 1) > 0 && w != null) {
//ss为扫描状态
int ss = w.scanState; // initially non-negative
//for循环 这是个死循环 origin将r与m求并，将多余位去除。然后赋值给k
for (int origin = r & m, k = origin, oldSum = 0, checkSum = 0;;) {
WorkQueue q; ForkJoinTask<?>[] a; ForkJoinTask<?> t;
int b, n; long c;
//如果k处不为空
if ((q = ws[k]) != null) {
//如果task大于0
if ((n = (b = q.base) - q.top) < 0 &&
(a = q.array) != null) { // non-empty
//计算i
long i = (((a.length - 1) & b) << ASHIFT) + ABASE;
//得到i处的task
if ((t = ((ForkJoinTask<?>)
U.getObjectVolatile(a, i))) != null &&
q.base == b) {
//如果扫描状态大于0
if (ss >= 0) {
//更改a中i的值为空 也就是此处将任务窃取走了
if (U.compareAndSwapObject(a, i, t, null)) {
//将底部的指针加1
q.base = b + 1;
//如果n小于-1 则通知工作线程工作
if (n < -1) // signal others
signalWork(ws, q);
//将窃取的task返回
return t;
}
}
//如果 scan状态小于0 则调用tryRelease方法唤醒哪些wait的worker
else if (oldSum == 0 && // try to activate
w.scanState < 0)
//调用tryRelease方法 后续详细介绍
tryRelease(c = ctl, ws[m & (int)c], AC_UNIT);
}
//如果ss小于0
if (ss < 0) // refresh
//更改ss
ss = w.scanState;
r ^= r << 1; r ^= r >>> 3; r ^= r << 10;
origin = k = r & m; // move and rescan
oldSum = checkSum = 0;
continue;
}
checkSum += b;
}
//此处判断k，k在此通过+1的方式完成对原有workQueues的遍历
if ((k = (k + 1) & m) == origin) { // continue until stable
if ((ss >= 0 || (ss == (ss = w.scanState))) &&
oldSum == (oldSum = checkSum)) {
if (ss < 0 || w.qlock < 0) // already inactive
break;
int ns = ss | INACTIVE; // try to inactivate
long nc = ((SP_MASK & ns) |
(UC_MASK & ((c = ctl) - AC_UNIT)));
w.stackPred = (int)c; // hold prev stack top
U.putInt(w, QSCANSTATE, ns);
if (U.compareAndSwapLong(this, CTL, c, nc))
ss = ns;
else
w.scanState = ss; // back out
}
checkSum = 0;
}
}
}
return null;
}
此方法最大的特点就是，根据随机数计算一个k，然后根据k去遍历workQueues，之后看看这个位置是否有数据，如果不为空，则检查checkSum，根据checkSum的状态缺认是否从这个队列中取数据。按之前约定的FIFO或者LIFO取数。这意味着，工作队列对窃取和是否获得本队列中的任务之间并没有优先级，而是根据随机数得到的index，之后对数组进行遍历。
7.3 tryRelease
对于执行完成的worker，则需要进行释放。
/**
* 如果worker处于空闲worker Stack的workQueue的顶部。则发信号对其进行释放。
*
* @param c incoming ctl value
* @param v if non-null, a worker
* @param inc the increment to active count (zero when compensating)
* @return true if successful
*/
private boolean tryRelease(long c, WorkQueue v, long inc) {
//sp取c的低位，计算vs
int sp = (int)c, vs = (sp + SS_SEQ) & ~INACTIVE; Thread p;
//如果v不为空 且v的sancState为sp
if (v != null && v.scanState == sp) { // v is at top of stack
//计算nc
long nc = (UC_MASK & (c + inc)) | (SP_MASK & v.stackPred);
//采用cas的方式 将ctl改为nc
if (U.compareAndSwapLong(this, CTL, c, nc)) {
//修改scanState的状态
v.scanState = vs;
//如果此时这线程为park状态，则调用unpark
if ((p = v.parker) != null)
U.unpark(p);
return true;
}
}
return false;
}
7.4 awaitWork
如果上述scan不到任务呢？也就是说，scan方法没有拿到task，则会调用awaitWork。将当前的线程进行阻塞。
/**
*如果不能窃取到任务，那么就将worker阻塞。如果停用导致线程池处于静止状态，则检查是否要关闭，如果这不是唯一的工作线程，则等待给定的持续时间，达到超时时间后，如果ctl没有更改，则将这个worker终止，之后唤醒另外一个其他的worker对这个过程进行重复。
*
* @param w the calling worker
* @param r a random seed (for spins)
* @return false if the worker should terminate
*/
private boolean awaitWork(WorkQueue w, int r) {
//如果w不为空切w的qlock小于0 则直接返回false
if (w == null || w.qlock < 0) // w is terminating
return false;
//for循环，这是个死循环，定义pred
for (int pred = w.stackPred, spins = SPINS, ss;;) {
//如果ss大于0 则返回
if ((ss = w.scanState) >= 0)
break;
//如果spins大于0
else if (spins > 0) {
//计算r
r ^= r << 6; r ^= r >>> 21; r ^= r << 7;
//如果r大于0 且spins为0
if (r >= 0 && --spins == 0) { // randomize spins
WorkQueue v; WorkQueue[] ws; int s, j; AtomicLong sc;
//如果pred不为0 且ws不为空
if (pred != 0 && (ws = workQueues) != null &&
//j<ws.length
(j = pred & SMASK) < ws.length &&
//j位置处不为空
(v = ws[j]) != null && // see if pred parking
//并且没有park
(v.parker == null || v.scanState >= 0))
//继续在for循环中自旋
spins = SPINS; // continue spinning
}
}
//如果qlock小于0 返回false
else if (w.qlock < 0) // recheck after spins
return false;
//如果被中断
else if (!Thread.interrupted()) {
long c, prevctl, parkTime, deadline;
int ac = (int)((c = ctl) >> AC_SHIFT) + (config & SMASK);
//如果线程池停止
if ((ac <= 0 && tryTerminate(false, false)) ||
(runState & STOP) != 0) // pool terminating
return false;
//最后的等待 获取不到任务 此时采用超时等待 park的方式进行
if (ac <= 0 && ss == (int)c) { // is last waiter
prevctl = (UC_MASK & (c + AC_UNIT)) | (SP_MASK & pred);
int t = (short)(c >>> TC_SHIFT); // shrink excess spares
if (t > 2 && U.compareAndSwapLong(this, CTL, c, prevctl))
return false; // else use timed wait
parkTime = IDLE_TIMEOUT * ((t >= 0) ? 1 : 1 - t);
deadline = System.nanoTime() + parkTime - TIMEOUT_SLOP;
}
else
prevctl = parkTime = deadline = 0L;
//获取线程
Thread wt = Thread.currentThread();
//设置PARKBLOCKER
U.putObject(wt, PARKBLOCKER, this); // emulate LockSupport
w.parker = wt;
//调用park方法
if (w.scanState < 0 && ctl == c) // recheck before park
U.park(false, parkTime);
U.putOrderedObject(w, QPARKER, null);
U.putObject(wt, PARKBLOCKER, null);
//如果scanState大于0 退出
if (w.scanState >= 0)
break;
//如果已经达到deadline的时间 则返回false
if (parkTime != 0L && ctl == c &&
deadline - System.nanoTime() <= 0L &&
U.compareAndSwapLong(this, CTL, c, prevctl))
return false; // shrink pool
}
}
return true;
}
上述这些方法，如果返回了false，则会导致外层的阻塞不能阻塞住，在runWorker方法中，会退出循环，这个线程就会退出。只有返回了true,那么这个线程就会一直运行，获得任务，循环。
7.5 workQueue的执行过程总结
当workQueue上的thread启动之后，这个线程就会调用runWorker方法。之后再runWorker方法中有一个死循环，不断的通过scan方法去扫描任务，实际上就是执行窃取过程。如下图所示，这样通过遍历外层workQueues的方式将会从任务队列中窃取任务进行执行。
窃取并执行
当执行之后，会通过fork产生新的任务，将这些任务任务添加到工作队列中去。其他线程继续scan，执行。这个过程不断循环，直到任务全部都执行完成，这样就完成了整个计算过程。
8.总结
本文对ForkJoinPool的部分源码进行分析，重点分析了外部提交过程和worker的执行过程。但是还有几个比较关键的地方，fork和join的过程，以及工作队列workQueue的组成都还没涉及到，限于篇幅限制，再后面的博客中接着补充。
网友评论