ForkJoinPool的整体逻辑其实相对于AQS来说简单多了，但是它的实现里面用了很多二进制的逻辑运算，导致整个实现看起来非常难，所以在正式的看ForkJoinPool的代码之前，先看一下二进制的一些玩法，这些玩法是我在ForkJoinPool的代码中摘出来的，明白了这些二进制的玩法后就能轻松的看动ForkJoinPool的逻辑代码了。详情见：二进制的一些玩法

1.初始化

先看下ForkJoinPool构造方法：

    /****parallelism： 并行度factory： 创建工作线程的工厂实现handler： 内部工作线程因为未知异常而终止的回调处理asyncMode： 异步模式（对于任务的处理顺序采用何种模式），true表示采用FIFO模式，false表示采用LIFO模式***/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();}

再看下内部重载的构造方法：

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 countsthis.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
}

这里我们需要重点分析一下config，np和ctl。
先看config，它等于(parallelism & SMASK) | mode;其中SMASK=0xffff;没有任务的业务含义，而并行度parallelism 与SMASK进行逻辑与运算，其实就是保证parallelism 不大于SMASK，作者这里有点多此一举了，因为私有的这个构造方法在传入parallelism之前都会进行parallelism的大小判断，都会保证parallelism不大于MAX_CAP(0x7fff),而MAX_CAP肯定是比SMASK小的。所以最终(parallelism & SMASK) | mode 可以简化为parallelism | mode。
我们看下mode有两个值LIFO和FIFO：

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
//其中LIFO_QUEUE为0这个很简单，而FIFO_QUEUE为1 << 16,转换成二进制表示法就是：
0000000000000001 0000000000000000（第17位为1）
而MAX_CAP(0x7fff)的二进制表示为：
0000000000000000 0111111111111111
所以parallelism | mode结果为两种：即17位是否为1用来表示模式，而低15位用来表示并行度。
当我们需要从config中取出模式的时候只需要用掩码MODE_MASK与config进行逻辑与运算（这是掩码的玩法），
因为MODE_MASK = 0xffff << 16二进制表示为：1111111111111111 0000000000000000，逻辑与运算
就取到了高16位，我们只用关注高16位是否为0（LIFO），或者1（FIFO），或者最高位为1（SHARED）

所以config可以总结表示为：

再来看下np = (long)(-parallelism)，即并行度补码转换为long型（64位），这里关于补码的运算不深入讲，自行百度，最终的结果就是：以MAX_CAP为例，则np为：

1111111111111111 1111111111111111 1111111111111111 1000000000000001

如果并行度为1，则np为：

1111111111111111 1111111111111111 1111111111111111 1111111111111111

那这个np有啥用？我们再看下ctl。

((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
//我们先看AC_SHIFT和TC_SHIFT// 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
//从上面的简单注释我们可以看出，AC即Active counts，即活跃线程数，而TC即Total counts，
//即总的线程数
其中np << AC_SHIFT代表np左移动48位，即低16位变成了高16位，所以ctl的64位中，高49~64位代表活跃线程数的负数，
同理np<<TC_SHIFT是将np左移32位，即低16位移动到了高33~48位，所以ctl的高33~48位代表线程总数的负数。
最后通过逻辑或运算合并到一起，这里还经过了掩码运算，例如& AC_MASK，就是为了取对应的位数。

所以那并行度MAX_CAP为例，初始化的ctl为：

1000000000000001 1000000000000001 0000000000000000 0000000000000000

暂且先不管ctl这样设计的用处，后续用到再分析。

2.核心方法

通过代码发现FrokJoinPool提供给外界的核心方法中，提交任务的有三类：submit、invoke、execute，然后还有一个shutdown方法，接下来一个一个分析。

2.1 invoke方法

/***
运行给定的任务，任务完成后返回结果。
如果运算期间遇到未受检查的异常或者错误，这些异常和错误将被当作是本次调用的结果重新抛出，
重新抛出的异常与普通的异常无异，但是，在可能的情况下，包含当前线程和真实遇到异常的线程
的栈信息，只有后者才能办到。
****/
public <T> T invoke(ForkJoinTask<T> task) {if (task == null)throw new NullPointerException();externalPush(task);return task.join();
}

该方法结构很简单，首先调用externalPush方法，最后返回任务join的结果。接下来看下核心的externalPush方法。

externalPush

/**
尝试将给定的任务添加到提交者的当前队列（其中一个submission queue).
在筛选externalSubmit需求时，只有(大部分)最常见的路径在此方法中被直接处理
**/
final void externalPush(ForkJoinTask<?> task) {WorkQueue[] ws; WorkQueue q; int m;int r = ThreadLocalRandom.getProbe();  //线程随机数int rs = runState; //runState运行状态，初始化为0//workQueues为整个线程池的工作者队列（其实就是一个数组）//当工作队列不为空，长度至少为1（并且m这里表示ws工作数组当前能够表示的最大下标）//其中m&r表示随机数不大于m，然后&SQMASK（SQMASK = 0x007e;）相当于只取偶数，并且偶数不大于0x7e(126),这里从随机的偶数槽位取出WorkQueue不为空，//r!=0, 说明不是第一个//rs>0 说明运行状态被初始化过//CAS:并且并发控制所取得的WorkQueue成功(qlock字段1表示锁定）if ((ws = workQueues) != null && (m = (ws.length - 1)) >= 0 &&(q = ws[m & r & SQMASK]) != null && r != 0 && rs > 0 &&U.compareAndSwapInt(q, QLOCK, 0, 1)) {ForkJoinTask<?>[] a; int am, n, s;if ((a = q.array) != null && //WorkQueue中的任务数组不为空(am = a.length - 1) > (n = (s = q.top) - q.base)) { //数组未装满？int j = ((am & s) << ASHIFT) + ABASE; //获取本次要放入元素的偏移量U.putOrderedObject(a, j, task);  //放入任务U.putOrderedInt(q, QTOP, s + 1); //top+1U.putIntVolatile(q, QLOCK, 0);   //释放任务队列(WorkQueue)的锁if (n <= 1) //放入任务成功后，如果发现放入任务前最多只有一个任务在队列中，当前放入任务成功后需要手动唤醒工作者，避免新加入的任务无法运行。？？signalWork(ws, q);return;}U.compareAndSwapInt(q, QLOCK, 1, 0); //释放锁}//在队列未进行初始化等条件下，代码直接来到此处，这里未初始化包括workQueues[]未初始化//或者对应下标的WorkQueue未初始化，或者WorkQueue中的ForkJoinTask<?>[]未初始化或者满externalSubmit(task);
}

从上面代码可以看到：

1.externalPush方法主要的目标是将用户提交的任务放入到线程池的下标为偶数的工作队列（workQueues[]的下标为偶数的工作队列）的任务列表中去。
2.当线程次中工作队列数组未初始，或者获取到的工作队列中的任务数组未初始化或者容量满，都转为调用externalSubmit方法，看来该方法包含了所有的初始化和扩容逻辑。

接下来重点看一下externalSubmit方法的执行逻辑。

externalSubmit

该方法的是一个完整的外部提交任务入任务队列的逻辑，大致的流程入下图：

详细的代码注释如下：

private void externalSubmit(ForkJoinTask<?> task) {int r;                                    // initialize caller's probeif ((r = ThreadLocalRandom.getProbe()) == 0) {ThreadLocalRandom.localInit();r = ThreadLocalRandom.getProbe(); //获取线程随机数}for (;;) {WorkQueue[] ws; WorkQueue q; int rs, m, k;boolean move = false;if ((rs = runState) < 0) { //线程池的状态为终止状态tryTerminate(false, false);     // help terminatethrow new RejectedExecutionException();}//线程池为初始化状态else if ((rs & STARTED) == 0 ||     // initialize((ws = workQueues) == null || (m = ws.length - 1) < 0)) {int ns = 0;rs = lockRunState();  //获取锁，这个方法一会儿单独解析try {if ((rs & STARTED) == 0) {//再判断一次状态是否为初始化，因为在lockRunState过程中有可能状态被别的线程更改了//初始化stealcounter的值（任务窃取计数器）U.compareAndSwapObject(this, STEALCOUNTER, null,new AtomicLong());// create workQueues array with size a power of two//这里英语注释已经很明白了，初始化任务队列数组，大小是2的k次幂int p = config & SMASK; // ensure at least 2 slots //这里取config的低16位（有值的只有低15位），表示并行度（见前面的config初始化）。//这里通过前面的代码我们知道，并行度默认是跟cpu核数相同，但是极端也可以到达MAX_CAP这么多。int n = (p > 1) ? p - 1 : 1;n |= n >>> 1; n |= n >>> 2;  n |= n >>> 4;n |= n >>> 8; n |= n >>> 16; n = (n + 1) << 1;workQueues = new WorkQueue[n];//关于n，大家可以看一下， 假如p=1，或者2，n最终就等于4，//当p=MAX_CAP的时候，n最终等于2的16次方//更有趣的是这个位移，一个数n通过→移动1，2，4，8，16，然后分别于自己取或运算，//其实的目的就是把数n二进制表示法的从非零位开始的低位全部变为1，然后最后一步n+1，其实是进1，低位清0，然后再左移1位（容量翻倍）////所以最终的结论就是：工作队列数组的大小，与并行度二进制表示后低位有效位数（k）有关，大小等于2的k+1次方。//具体对比可以看下面的**表格2-1**ns = STARTED; //}} finally {unlockRunState(rs, (rs & ~RSLOCK) | ns); //解锁单独说}}else if ((q = ws[k = r & m & SQMASK]) != null) { //这里取偶数槽位（或者0），并且低于126（SQMASK确定的，所以这里我们也可以看到，偶数槽位最多64个）if (q.qlock == 0 && U.compareAndSwapInt(q, QLOCK, 0, 1)) {//判断这个工作队列的锁定状态ForkJoinTask<?>[] a = q.array;int s = q.top;boolean submitted = false; // initial submission or resizingtry {                      // locked version of push//如果任务数组不为空，并且数组长度大于已经放入的元素+1（说明至少还能再放一个）//否则对数组进行扩容if ((a != null && a.length > s + 1 - q.base) ||(a = q.growArray()) != null) {//这里的逻辑非常有意思，请见后面的详细讲解//这里的目的是计算本次应该放入的任务在内存中的偏移位置，其实取到的就是top<<ASHIFT + ABASEint j = (((a.length - 1) & s) << ASHIFT) + ABASE;U.putOrderedObject(a, j, task); //放入任务U.putOrderedInt(q, QTOP, s + 1); //top+1submitted = true; //任务提交成功标志}} finally {U.compareAndSwapInt(q, QLOCK, 1, 0);}if (submitted) { //提交任务成功，唤醒工作线程signalWork(ws, q);return;  //唯一的任务提交自旋出口，任务提交成功返回}}//标记任务未成功提交，需要再次计算随机数，然后再尝试move = true;                   // move on failure}//如果找到的槽为空，则需要初始化WorkQueueelse if (((rs = runState) & RSLOCK) == 0) { // create new queueq = new WorkQueue(this, null);  //初始化q.hint = r; //设置工作队列的任务窃取线索q.config = k | SHARED_QUEUE; //将工作队列所在池的位置和任务队列模式记录到config中q.scanState = INACTIVE; //工作队列状态为未活动，小于0rs = lockRunState();           // publish index //锁定if (rs > 0 &&  (ws = workQueues) != null &&k < ws.length && ws[k] == null)ws[k] = q;                 // else terminated //将q放入工作队列池unlockRunState(rs, rs & ~RSLOCK);}elsemove = true;                   // move if busyif (move)r = ThreadLocalRandom.advanceProbe(r);}
}

以下是并行度与容量的关系，注意并行度p需要减1
表格2-1

并行度p	并行度-1后二进制表示	容量
1/2	01	4
3/4	1x	8
5/6/7/8	1xx	16
9~16	1xxx	32

从表格我们也能看出，为什么int n = (p > 1) ? p - 1 : 1;参与运行的n需要减去1，其实是为了2的k次方位减1降低1位，这样才有1与2一组，3与4一组，5到8一组，否则就会变为，1为一组，2与3一组，4到7一组。

计算任务数组放入元素的内存偏移量

int j = (((a.length - 1) & s) << ASHIFT) + ABASE;

由代码得知，a.length - 1大于s,所以最后可以写为int j = s<<ASHIFT +ABASE
其中ASHIFT和ABASE的代码如下：

private static final sun.misc.Unsafe U;
private static final int  ABASE;
private static final int  ASHIFT;
private static final long QTOP;
private static final long QLOCK;
private static final long QCURRENTSTEAL;
static {try {U = sun.misc.Unsafe.getUnsafe();Class<?> wk = WorkQueue.class;Class<?> ak = ForkJoinTask[].class;QTOP = U.objectFieldOffset(wk.getDeclaredField("top"));QLOCK = U.objectFieldOffset(wk.getDeclaredField("qlock"));QCURRENTSTEAL = U.objectFieldOffset(wk.getDeclaredField("currentSteal"));ABASE = U.arrayBaseOffset(ak); //获取ForkJoinTask[]数组对象头的偏移位置int scale = U.arrayIndexScale(ak); //获取ForkJoinTask[]数组中每个元素引用所占的大小if ((scale & (scale - 1)) != 0)  //确保scale为2的k次方throw new Error("data type scale not a power of two");//Integer.numberOfLeadingZeros(scale)表示scale的高位连续0的位数//然后31减去Integer.numberOfLeadingZeros(scale)，其实就是表示scale低位有多少位//例如scale=4, 二进制表示为：0000000000000000 0000000000000100，所以ASHIFT=2ASHIFT = 31 - Integer.numberOfLeadingZeros(scale); } catch (Exception e) {throw new Error(e);}
}

那么int j = s<<ASHIFT +ABASE到底表示什么呢？我们先不管它，我们来想一下，如果我们正常用加减乘除来计算数组元素便宜位置应该如何就算，答案是：scaletop+ABASE,其中top表示下一次要放置元素的下标，那么这个公式又怎么和int j = s<<ASHIFT +ABASE扯上关系了（其中s就算top哈）？我们看到ABASE都可以干掉，然后就剩下scale * top 与 top<<ASHIFT了，由二进制的一些玩法可以知道，一个数mn，如果n是2的k次方，那么mn可以直接表示为m<<k，那么再看scale * top, 其中scale就算2的k次方，所以完全就可以表示为top << k, 那么这个k，其实就是ASHIFT，所以scale * top 就等于 top<<ASHIFT，最终总结就是：int j = s（top）<<ASHIFT +ABASE 与 scaletop+ABASE等价，看到这里，很想说一句cao。

线程池加锁 lockRunState/unlockRunState

由上面的逻辑我们可以看到，在初始化线程池或者在向线程池中添加任务队列时都需要锁定线程池，这里不由自主的就产生了一个疑问：为什么不用ReentrantLock呢？
我们先看lockRunState的代码：


//在开始看代码之前，我们要来看一下runState的各个位代表的含义
// 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; //线程池关闭
//二进制表示为
1110000000000000 0000000000000111//其中RSLOCK=1，如果runState & RSLOCK == 0,说明没有加锁，则通过CAS快速尝试加锁，如果失败则进入awaitRunStateLock()
private int lockRunState() {int rs;return ((((rs = runState) & RSLOCK) != 0 ||!U.compareAndSwapInt(this, RUNSTATE, rs, rs |= RSLOCK)) ?awaitRunStateLock() : rs);
}//该方法保证加锁一定成功
private int awaitRunStateLock() {Object lock;boolean wasInterrupted = false; //线程中断标记for (int spins = SPINS, r = 0, rs, ns;;) {if (((rs = runState) & RSLOCK) == 0) { //判断是否加锁（==0表示未加锁）//未加锁的情况下， 通过CAS加锁（即runState最低位变为1）if (U.compareAndSwapInt(this, RUNSTATE, rs, ns = rs | RSLOCK)) {if (wasInterrupted) {try {Thread.currentThread().interrupt(); //重置线程中断标记} catch (SecurityException ignore) {}}return ns;//加锁成功返回最新的runState值，这也是整个自旋的唯一出口}}else if (r == 0) //当前锁被其它线程占有，更新线程随机数r = ThreadLocalRandom.nextSecondarySeed();else if (spins > 0) { //这里是随机的空转，初始SPINS是0，这里不会进入r ^= r << 6; r ^= r >>> 21; r ^= r << 7; // xorshiftif (r >= 0)--spins;}//如果是初始化状态加锁，锁被其它线程占有，则让出CPU，让其它线程加快初始else if ((rs & STARTED) == 0 || (lock = stealCounter) == null)Thread.yield();   // initialization race//如果其它线程持有锁，并且线程池已经初始化，则将需要唤醒位标记为1else if (U.compareAndSwapInt(this, RUNSTATE, rs, rs | RSIGNAL)) {synchronized (lock) { //通过该实例的属性加锁if ((runState & RSIGNAL) != 0) {//这里加锁后再判断了一次，如果RSIGNAL位已经为0，//则说明在加锁前刚好有线程进行了唤醒动作，所以这里不再等待直接到else里唤醒，否则说明RSIGNAL位==1，则让当前线程等待。try {lock.wait();} catch (InterruptedException ie) {if (!(Thread.currentThread() instanceofForkJoinWorkerThread))wasInterrupted = true;}}elselock.notifyAll();}}}
}

总体来看加锁的逻辑：
1.首先是看其它线程是否已经持有锁，没有则通过CAS改变RSLOCK位，完成加锁，如果加锁不成功则再自旋。
2.如果其它线程持有了锁，看下其它线程是不是在初始化线程池，如果是，则放弃本次CPU，否则再自旋。
3.如果其它线程持有了锁，并且线程池已经初始化，则设置唤醒标志位RSIGNAL,让当前竞争锁的线程睡眠，这个睡眠的动作是同步执行的，如果在睡眠前发现RSIGNAL位已经被释放，则执行唤醒操作。

再来看下开头提出的问题：为什么不用ReentrantLock？我们可以看到，这里自行实现的锁逻辑要比用ReentrantLock更细，这里在竞争锁的过程中会判断线程池状态，如果是初始化状态，则会放弃CPU，减少竞争；其次是作者将锁定标志位，唤醒位等一同设计到了runState这个字段中，如果用ReentrantLock，则就不需要这两个字段。总的来说如果用ReentrantLock是肯定能达到线程池加锁的目的，只是略微粗一点而已。

再看看unlockRunState的实现:

/*** Unlocks and sets runState to newRunState.** @param oldRunState a value returned from lockRunState* @param newRunState the next value (must have lock bit clear).*///从注释可以看出，newRunState一定是将锁定标志位清零了的
private void unlockRunState(int oldRunState, int newRunState) {//直接尝试CAS更新//如果更新失败，说明RSIGNAL位从0变成了1（oldRunState中的RSIGNAL为0），又因为newRunState没有改变RSIGNAL，所以newRunState中的RSIGNAL为0.//当然，如果这里更新成功，则说明RSIGNAL没变，有可能是持有锁的过程中没有其它线程竞争，所以不需要唤醒动作，如果有其它线程竞争，在前一个持有者释放锁的时候一定会把RSGINAL位强制设置为0，所以被唤醒的线程重新获取到锁的时候RSIGNAL位一定为0，这样就形成了一个良性循环。if (!U.compareAndSwapInt(this, RUNSTATE, oldRunState, newRunState)) {//CAS失败，代表oldRunState中的RSIGNAL位改变了（0变成了1）Object lock = stealCounter;//强制替换，更新了RSIGNAL位和RSLOCK位，都变为0runState = newRunState;              // clears RSIGNAL bitif (lock != null)//由于RSIGNAL位变为0，所以要唤醒全部的等待线程synchronized (lock) { lock.notifyAll(); }}
}

这里代码虽短，但是逻辑很有趣，通常的逻辑是释放锁再唤醒其它线程，但是这里的逻辑刚好是CAS失败才唤醒，为什么？
因为失败的唯一原因可能是有线程竞争而导致等待需要唤醒，如果CAS成功，则不需要做任何多余动作。

signalWork

在任务提交到工作队列成功后会调用signalWork方法进行工作线程唤醒，下面就来详细的看一下signalWork做了什么事情。

/*** Tries to create or activate a worker if too few are active.** @param ws the worker array to use to find signallees* @param q a WorkQueue --if non-null, don't retry if now empty*///注释：如果太少的活动线程，则该方法会创建或者激活一个工作者
/*****
并行度：MAX_CAP
ctl:1000000000000001(AC) 1000000000000001(TC) 00000000000000000000000000000000 （SP）
ADD_WORKER : 0x0001L << 47;  二进制表示为： 0000000000000000 1000000000000000 0000000000000000 0000000000000000
***/
public void signalWork(WorkQueue[] ws, WorkQueue q) {long c; int sp, i; WorkQueue v; Thread p;//如果ctl小于0，则说明活跃工作者太少，原因一会儿看while ((c = ctl) < 0L) {                       // too few active//将c取低32位（强转int）为SP，如果它等于0，则说明没有发呆的工作者if ((sp = (int)c) == 0) {                  // no idle workers//取TC的高位，如果不等于0，则说明目前的工作着还没有达到并行度if ((c & ADD_WORKER) != 0L)            // too few workerstryAddWorker(c); //添加工作者break;}if (ws == null)                            // unstarted/terminatedbreak;//工作队列数组长度小于SP的低16位。 （？）if (ws.length <= (i = sp & SMASK))         // terminatedbreak;//？if ((v = ws[i]) == null)                   // terminatingbreak;//---------------以下逻辑是：有发呆的worker，需要激活,逻辑确实没看懂，有过路的大神看到指点下---------------- //sp为ctl的低32位//SS_SEQ：第17位为1，后面16位为0，则SS_SEQ为2的16次方， //INACTIVE = 1 << 31，第32位为1，其它均为0，~INACTIVE则反过来，低1到31位为1.//vs代表：sp高17进行+1操作后，取低1到31位。int vs = (sp + SS_SEQ) & ~INACTIVE;        // next scanState//sp是pool的ctl里面的属性，scanState初始化为WorkQueue在数组中的下标（奇数）int d = sp - v.scanState;                  // screen CAS//ctl中对AC+1，取高32位，即AC和TC,long nc = (UC_MASK & (c + AC_UNIT)) | (SP_MASK & v.stackPred);//这里scanState, nc, stackPred， vs这几个值是怎么用的？if (d == 0 && U.compareAndSwapLong(this, CTL, c, nc)) {v.scanState = vs;                      // activate vif ((p = v.parker) != null)U.unpark(p);break;}if (q != null && q.base == q.top)          // no more workbreak;}
}

在signalWork方法中有很多没有看明白的地方，尤其是唤醒一个worker的逻辑，这里暂且先放一边，先来考虑添加worker的情况。在以开始的逻辑里，ctl<0则说明没有足够多的活跃worker，则是因为ctl的高16位，也就是49到64位为AC的表示，而由前面的初始化环节我们知道，AC是表示并行度的负数，高位肯定为1，如果按照每激活一个worker则对AC加一的逻辑，则当活跃worker数于并行度相等时AC则为0，再加就为正，所以能通过ctl是否小于0来判断活跃worker是否足够。再看(sp = (int)c) == 0，即ctl取低32位，用它是否等于0来判断是否有发呆的worker，如果它等于0，则说明没有发呆的，这里我们可以反推得出sp里藏着发呆worker的数量，但是现在还不知道是怎么存的，需要后续考量。然后在活跃数不够，并且没有发呆的情况下（说明创建出来的工作者都投入到了工作中），再判断worker总数是不是太少。(c & ADD_WORKER) != 0L，其中c & ADD_WORKER即为了取第48位，即TC的高位，而TC的初始化逻辑和AC是一样的，所以如果与运算后不等于0，则说明高位不等于0（为1），所以同理得出TC<0，因此说明总的worker数还未达到并行度，所以这个时候就需要创建工作者了。

创建工作者

/*** Tries to add one worker, incrementing ctl counts before doing* so, relying on createWorker to back out on failure.** @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).*///尝试新增一个worker，在新增之前首先对ctl进行增加，如果异常失败，则createWorker 方法会返回错误//c:即ctl，TC为负，并且没有发呆的worker。如果CAS失败，则会刷新c重新尝试。
private void tryAddWorker(long c) {boolean add = false;do {//激活数和总数分别+1long nc = ((AC_MASK & (c + AC_UNIT)) |(TC_MASK & (c + TC_UNIT)));//如果ctl未发生变化if (ctl == c) {int rs, stop;                 // check if terminating//获取线程池运行状态，并取STOP标志位，==0说明未停止if ((stop = (rs = lockRunState()) & STOP) == 0)add = U.compareAndSwapLong(this, CTL, c, nc);//更新CTLunlockRunState(rs, rs & ~RSLOCK); //解锁//说明stop位为1，说明线程池停止if (stop != 0)break;//ctl中的计数新增成功后，再创建真正的工作者if (add) {createWorker();break;}}} while (((c = ctl) & ADD_WORKER) != 0L && (int)c == 0); //如果ctl发生变化， 重新计算TC高位，不等于0说明TC总数还未达到并行度，并且无发呆的线程
}

tryAddWorker的主要逻辑是对ctl的AC和TC加1，然后尝试更新ctl，如果更新成功则执行真正的创建工作者，如果更新失败，则刷新ctl继续尝试。

/*** Tries to construct and start one worker. Assumes that total* count has already been incremented as a reservation.  Invokes* deregisterWorker on any failure.* 尝试创建并启动一个worker。ctl中的TC在之前已经完成了+1的动作，当创建失败的时候，需要调用deregisterWorker 方法* @return true if successful*/
private boolean createWorker() {ForkJoinWorkerThreadFactory fac = factory;Throwable ex = null;ForkJoinWorkerThread wt = null;try {//创建线程并启动线程if (fac != null && (wt = fac.newThread(this)) != null) {wt.start();return true;  //如果创建工作者成功，则返回}} catch (Throwable rex) {ex = rex;}//创建工作线程出现任务异常都必须执行“解除注册”deregisterWorker(wt, ex);return false;
}

这里我们可以看到createWorker的核心逻辑很简单，就是调用ForkJoinWorkerThreadFactory 来创建一个新的ForkJoinWorkerThread ，然后启动这个线程，如果启动成功则返回true，否则在创建和启动过程中出现仍和错误都需要执行deregisterWorker方法，然后返回false。所以这里的核心我们先看下创建工作线程是怎么创建的。
这里会涉及一些ForkJoinWorkerThreadFactory 的代码，在这个地方我们只是对这些代码简单的看下，后续会对ForkJoinWorkerThreadFactory ，ForkJoinWorkerThread 等进行详细的分析。

//ForkJoinWorkerThreadFactory 是ForkJoinPool里面的一个内部接口，同时也有内部实现类DefaultForkJoinWorkerThreadFactory
//在默认的不带有ForkJoinWorkerThreadFactory 参数的ForkJoinPool构造方法中，会默认DefaultForkJoinWorkerThreadFactory作为实现，否则用户将自定义实现然后传入。
public static interface ForkJoinWorkerThreadFactory {public ForkJoinWorkerThread newThread(ForkJoinPool pool);
}
//默认实现
static final class DefaultForkJoinWorkerThreadFactoryimplements ForkJoinWorkerThreadFactory {public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {return new ForkJoinWorkerThread(pool);}
}
//初始化一个WorkerThread
//这里我们可以看出来，所谓的创建一个worker，其实就是创建一个ForkJoinWorkerThread对象
protected ForkJoinWorkerThread(ForkJoinPool pool) {// Use a placeholder until a useful name can be set in registerWorkersuper("aForkJoinWorkerThread");this.pool = pool; //记录该线程所属的池this.workQueue = pool.registerWorker(this); //将工作线程注册到池
}

整个逻辑看下来也很简单，就是创建ForkJoinWorkerThread对象，然后将它注册到池中，这里的核心就在于如何将线程注册到池中，或者说这个注册动作是什么含义，线程与ForkJoinPool是一个什么样的关系，它们之间怎么联系的。接下来分析一下registerWorker方法就明白了。

/*** Callback from ForkJoinWorkerThread constructor to establish and* record its WorkQueue.* 从ForkJoinWorkerThread 的构造方法中回调回来用于确定并记录它的WorkQueue* @param wt the worker thread   即pool.registerWorker(this)中的this* @return the worker's queue，线程所使用的WorkQueue*/
final WorkQueue registerWorker(ForkJoinWorkerThread wt) {UncaughtExceptionHandler handler;wt.setDaemon(true); //设置守护线程                          // configure threadif ((handler = ueh) != null)wt.setUncaughtExceptionHandler(handler); //设置异常回调WorkQueue w = new WorkQueue(this, wt);       //创建一个新的WorkQueue，并将工作线程记录到WorkQueue中，int i = 0;     //新增WorkQueue将放入池中数组的下标    // assign a pool indexint mode = config & MODE_MASK; //获取池的模式，忘记了的往回看int rs = lockRunState();       //锁定池try {/**以下一段逻辑主要完成的任务是：在workQueues中找到一个空的奇数槽，为什么要是奇数呢？前面在提交任务那一节我们讲过，外部任务的提交是放到池的数组的偶数下标（包括0），并且下标值不大于SQMASK（126），***/WorkQueue[] ws; int n;                    // skip if no arrayif ((ws = workQueues) != null && (n = ws.length) > 0) {//数组不未空且有值//indexSeed 默认是0，SEED_INCREMENT = 0x9e3779b9，十进制为-1640531527int s = indexSeed += SEED_INCREMENT;  // unlikely to collide  //这样计算是为了减少冲突，至于为什么， 我也没看懂。应该是一个数学算法int m = n - 1;  //数组长度-1就是最大的下标//由前面逻辑知道n一定是2的k次幂，也就是一定是偶数，所以m一定是奇数//所以这里((s << 1) | 1) & m就得到一个小于m的奇数，i = ((s << 1) | 1) & m;               // odd-numbered indicesif (ws[i] != null) { //判断该槽位是否有值，有值则冲突   // collisionint probes = 0;                   // step by approx half n//计算一个增量，大约是n的一半int step = (n <= 4) ? 2 : ((n >>> 1) & EVENMASK) + 2;//将i每次递增step，然后再判断是否冲突while (ws[i = (i + step) & m] != null) {if (++probes >= n) {//每尝试一次则probes 加1，当尝试n次后如果还是冲突，则尝试扩容（这里的这个算法确实每看懂），扩容后又重新计算。workQueues = ws = Arrays.copyOf(ws, n <<= 1);m = n - 1;probes = 0;}}}//当找到一个空的槽位后，将对应的值付给WorkQueuew.hint = s;                           // use as random seed（随机数种子）w.config = i | mode;  //WorkQueue的config,由它在池中下标以及模式组成w.scanState = i;  //初始化scanState为池中下标，奇数                   // publication fencews[i] = w; //WorkQueue放入池中}} finally {unlockRunState(rs, rs & ~RSLOCK);}//设置线程名称为自定义前缀加上加标的一半（为什么是下标的一半而不是下标？不知道）。wt.setName(workerNamePrefix.concat(Integer.toString(i >>> 1)));return w;
}

看了下registerWorker的逻辑，整体的理解是没有问题的，但是要想抠细节，很多都每看明白。总体来说它是将工厂创建出来的ForkJoinWorkerThread 注册到池中，这个注册过程其实就是创建一个新的WorkQueue，并把这个线程关联到WorkQueue中，然后在池中找到一个奇数下标的槽位将WorkQueue放入。
这其中不明白的细节点是：
1.为什么计算初始下标是indexSeed += SEED_INCREMENT，然后左移位
2.冲突尝试的过程为什么step是取n的大概一半的位置开始，并且尝试的次数到达n才扩容。
3.赋值给WorkQueue的hint和scanState是怎么使用的?
4.线程名字为什么不是设置为前缀加池中下标，反而是下标的一半？

看完了注册动作，我们再回头看下，在createWorker中，如果创建新的Worker过程中出现异常，则回执行deregisterWorker。可以先不看代码想一下deregisterWorker可能会做些什么事情。大概会有：销毁线程，销毁刚才创建WorkQueue。

/**最终是从终止中的以及在构造或者启动时发生异常的worker回调回来的。该方法会清理池中数组的worker引用，修改总数。如果池将关闭，则尝试完成终止。* Final callback from terminating worker, as well as upon failure* to construct or start a worker.  Removes record of worker from* array, and adjusts counts. If pool is shutting down, tries to* complete termination.** @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;//1.清空池中的该工作线程对应的WorkQueue引用if (wt != null && (w = wt.workQueue) != null) {WorkQueue[] ws;                           // remove index from arrayint idx = w.config & SMASK;int rs = lockRunState();if ((ws = workQueues) != null && ws.length > idx && ws[idx] == w)ws[idx] = null;unlockRunState(rs, rs & ~RSLOCK);}//2.池的ctl标记里面的AC和TC分别-1long c;                                       // decrement countsdo {} while (!U.compareAndSwapLong(this, CTL, c = ctl, ((AC_MASK & (c - AC_UNIT)) |(TC_MASK & (c - TC_UNIT)) |(SP_MASK & c))));//3.清空WorkQueue ，取消未进行的任务                                  if (w != null) {w.qlock = -1;                             // ensure setw.transferStealCount(this);w.cancelAll();                            // cancel remaining tasks}//4.尝试for (;;) {                                    // possibly replaceWorkQueue[] ws; int m, sp;//4.1首先尝试终止，如果失败则查看工作队列、队列中的数组是否为空，最后再runState中的STOP标志护照workQueues是否为空//如果有之一满足，则说明线程池已经停止if (tryTerminate(false, false) || w == null || w.array == null ||(runState & STOP) != 0 || (ws = workQueues) == null ||(m = ws.length - 1) < 0)              // already terminatingbreak;//4.2 SP不等于0（说明啥 ？） if ((sp = (int)(c = ctl)) != 0) {         // wake up replacementif (tryRelease(c, ws[sp & m], AC_UNIT))break;}//4.3SP==0，有异常，并且线程总数还未超过并行度，则添加一个工作者else if (ex != null && (c & ADD_WORKER) != 0L) {tryAddWorker(c);                      // create replacementbreak;}else                                      // don't need replacementbreak;}//4.4重新处理异常if (ex == null)                               // help clean on way outForkJoinTask.helpExpungeStaleExceptions();else                                          // rethrowForkJoinTask.rethrow(ex);
}

在deregisterWorker的过程中，有两个方法，一个是tryTerminate，另外一个是tryRelease, 这两个方法究竟是干嘛的，先看下前者。
在介绍下面的代码前，我们有必要先对线程池的几个状态再做一下描述：
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 //不接收新任务，但是继续处理已接收任务

/**可能启动或者完成终止能够终止的前提条件是runState变为SHUTDOWN（1 << 31）now:是否立即终止，now为true代表立即终止，如果为false，会延迟到等所有的worker都结束后再终止enable: 是否开启SHUTDOWN,在进入tryTerminate方法后，如果runState状态没有标记SHUTDOWN，则本次是否需要标记，如果为true，则修改runState，否则退出方法结束终止操作。Possibly initiates and/or completes termination.** @param now if true, unconditionally terminate, else only* if no work and no active workers* @param enable if true, enable shutdown when next possible* @return true if now terminating or terminated*/
private boolean tryTerminate(boolean now, boolean enable) {int rs;//1.公共线程池不能终止（这个公共线程池在jdk1.8之后的并行处理接口中被使用）if (this == common)                       // cannot shut downreturn false;//2.在runState没有被标记SHUTDOWN的情况下，根据enbale参数判断本次是否要执行标记SHUTDOWN操作 //runState >=0 说明高32位为0，即runState没有SHUTDOWNif ((rs = runState) >= 0) {if (!enable) //这里用到了enable,如果是false，这里就直接返回了，即进入tryTerminate方法时，如果是runState里面没有标记SHUTDOWN,则直接返回false，//如果enable为true则修改runState的值，添加SHUTDOWN位。return false;rs = lockRunState();                  // enter SHUTDOWN phaseunlockRunState(rs, (rs & ~RSLOCK) | SHUTDOWN);}//3.判断线程池是否要停止处理已接收任务，如果STOP为0，表示还能继续处理已接收的任务if ((rs & STOP) == 0) {if (!now) {//检查是否要立即停止，now为true代表立即停止 for (long oldSum = 0L;;) {        // repeat until stableWorkQueue[] ws; WorkQueue w; int m, b; long c;long checkSum = ctl;//如果AC加上并行度大于0，大家还记得AC么， 是根据并行度取负数进行初始      化，后续只要增加一个跃工作者，则AC加一，如此以来，只要有活跃工作者，则AC+并行度就大于0//在不需要立即停止线程的情况下，有活跃的工作者，则本次终止失败，返回falseif ((int)(checkSum >> AC_SHIFT) + (config & SMASK) > 0)return false;             // still active workers//如果工作队列数组为空，则不需要继续处理，这里为什么是<=0,而不是小于0？if ((ws = workQueues) == null || (m = ws.length - 1) <= 0)break; //如果工作队列数组不为空，则遍历每一个工作队列          // check queuesfor (int i = 0; i <= m; ++i) {if ((w = ws[i]) != null) {//获取工作队列//工作队列中有待处理的任务或者正在扫描/或者有窃取来的任务，则本次终止返回falseif ((b = w.base) != w.top || w.scanState >= 0 ||w.currentSteal != null) {tryRelease(c = ctl, ws[m & (int)c], AC_UNIT);return false;     // arrange for recheck}//如果该工作队列已经没有任务，则增加校验和的值checkSum += b;//判断该工作队列是否为外部提交任务（i为偶数），如果是，则关闭外部任务if ((i & 1) == 0)w.qlock = -1;     // try to disable external}}//判断校验和是否跟上次循环的相等， 如果相等了， 说明本次循环过程中ws中已经没有任务任务队列if (oldSum == (oldSum = checkSum))break;}//for}//经过一番处理后，表示已有任务都处理完成，这里就给状态打上STOP的标志，表示不再处理已有任务if ((runState & STOP) == 0) {rs = lockRunState();              // enter STOP phaseunlockRunState(rs, (rs & ~RSLOCK) | STOP);}}//以上这段代码主要完成在now==false（不是立即终止线程池）时，这种情况下，如果工作队列数组中还有还有工作队列并且队列中还有未完成的任务，或者还有活跃的工作线程，则本次终止动作失败。//接下来是完成终止动作int pass = 0;                             // 3 passes to help terminatefor (long oldSum = 0L;;) {                // or until done or stableWorkQueue[] ws; WorkQueue w; ForkJoinWorkerThread wt; int m;long checkSum = ctl;//已经没有工作者或者有工作者但是工作队列数组为空，给状态打上TERMINATED标记if ((short)(checkSum >>> TC_SHIFT) + (config & SMASK) <= 0 ||(ws = workQueues) == null || (m = ws.length - 1) <= 0) {if ((runState & TERMINATED) == 0) {rs = lockRunState();          // doneunlockRunState(rs, (rs & ~RSLOCK) | TERMINATED);synchronized (this) { notifyAll(); } // for awaitTermination}break; //如果已经没有工作者或者workQueues为空，则说明线程池已经终止， 这里直接结束循环}//遍历工作队列数组for (int i = 0; i <= m; ++i) {if ((w = ws[i]) != null) {checkSum += w.base;  //下标有对应的工作队列， 则该表校验和w.qlock = -1; //disable掉工作队列，这里qlock在WorkQueue中介绍if (pass > 0) {w.cancelAll();            // clear queueif (pass > 1 && (wt = w.owner) != null) {if (!wt.isInterrupted()) {try {             // unblock joinwt.interrupt();} catch (Throwable ignore) {}}if (w.scanState < 0)U.unpark(wt);     // wake up}}}}if (checkSum != oldSum) {             // unstableoldSum = checkSum;pass = 0;}else if (pass > 3 && pass > m)        // can't further helpbreak;else if (++pass > 1) {                // try to dequeuelong c; int j = 0, sp;            // bound attemptswhile (j++ <= m && (sp = (int)(c = ctl)) != 0)tryRelease(c, ws[sp & m], AC_UNIT);}}return true;
}

咬牙看完tryTerminate的代码，留下了一堆的疑问：
1.ctl的低32位到底是个什么玩意儿？也就是代码中经常看到的（int)ctl，或是叫sp，为什么sp&m就是工作队列在数组中的位置了？？？
2.代码中多次出现m = ws.length - 1) <= 0这样的判断，为什么这里要减去1，判断数组是否为空难道不应该直接判断是否小于等于0么？？
3.STOP， TERMINATE, SHUTDOWN这几个状态标记到底咋个玩的???
暂且先放过这一段，先看下WorkQueue 和 ForkJoinWorkerThread 的实现看能否弄清楚这几个问题。

更新中…

ForkJoinPool源码深度解析相关推荐

dubbo源码深度解析_Spring源码深度解析：手把手教你搭建Spring开发环境
Spring环境搭建流程,如果是第一次接触spring源码的环境搭建,确实还是比较麻烦的. 作者使用的编译器为目前流行的lntelliJ IDEA,版本为2018旗舰版.Eclipse用户还需要自己揣 ...
《Spring源码深度解析》 PDF
Spring源码深度解析 PDF 下载下载地址:https://pan.baidu.com/s/1o9qEwXW 密码:vwyo 转载:http://download.csdn.net/detail ...
Go netpoll I/O 多路复用构建原生网络模型之源码深度解析
原文 Go netpoll I/O 多路复用构建原生网络模型之源码深度解析导言 Go 基于 I/O multiplexing 和 goroutine 构建了一个简洁而高性能的原生网络模型(基于 Go ...
Java LockSupport以及park、unpark方法源码深度解析
介绍了JUC中的LockSupport阻塞工具以及park.unpark方法的底层原理,从Java层面深入至JVM层面. 文章目录 1 LockSupport的概述 2 LockSupport的特征和 ...
Spring源码深度解析(郝佳)-学习-源码解析-基于注解bean定义（一）
我们在之前的博客 Spring源码深度解析(郝佳)-学习-ASM 类字节码解析简单的对字节码结构进行了分析,今天我们站在前面的基础上对Spring中类注解的读取,并创建BeanDefinition做 ...
《Spring源码深度解析郝佳第2版》AOP
往期博客 <Spring源码深度解析郝佳第2版>容器的基本实现与XML文件的加载 <Spring源码深度解析郝佳第2版>XML标签的解析 <Spring源码深度解 ...
《Spring源码深度解析郝佳第2版》ApplicationContext
往期博客: <Spring源码深度解析郝佳第2版>容器的基本实现与XML文件的加载 <Spring源码深度解析郝佳第2版>XML标签的解析 <Spring源码深度 ...
《Spring源码深度解析郝佳第2版》SpringBoot体系分析、Starter的原理
往期博客 <Spring源码深度解析郝佳第2版>容器的基本实现与XML文件的加载 <Spring源码深度解析郝佳第2版>XML标签的解析 <Spring源码深度解 ...
Spring源码深度解析(郝佳)-学习-源码解析-创建AOP静态代理实现（八）
继上一篇博客,我们继续来分析下面示例的 Spring 静态代理源码实现. 静态 AOP使用示例加载时织入(Load -Time WEaving,LTW) 指的是在虚拟机载入字节码时动态织入 Aspe ...

ForkJoinPool源码深度解析

目录