java线程池(五)：ForkJoinPool源码分析之一(外部提交及worker执行过程)

//界限定义
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的计算字段和掩码 
// 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

// 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;

// 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

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

public ForkJoinPool() {
    this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
         defaultForkJoinWorkerThreadFactory, null, false);
}

public <T> T invoke(ForkJoinTask<T> task) {
    if (task == null)
        throw new NullPointerException();
    externalPush(task);
    return task.join();
}

public void execute(ForkJoinTask<?> task) {
    if (task == null)
        throw new NullPointerException();
    externalPush(task);
}

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);
}

public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
    if (task == null)
        throw new NullPointerException();
    externalPush(task);
    return task;
}

public <T> ForkJoinTask<T> submit(Callable<T> task) {
    ForkJoinTask<T> job = new ForkJoinTask.AdaptedCallable<T>(task);
    externalPush(job);
    return job;
}

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;
}

/**
 * 尝试将任务添加到提交者当前的队列中，此方法只处理大多数情况，实际上是根据随机数指定一个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的完整版本，处理哪些不常用的逻辑。如第一次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);
    }
}

/**
 * 此处将激活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;
    }
}

/**
 * 尝试新增一个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);
}

// 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;
}

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);
}

/**
 * 创建线程，并建立线程与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;
}

/**
 *此方法的主要目的是在创建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);
}

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);
            }
        }
    }
}

/**
 * 通过调用线程的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方法进行任务窃取，扫描从一个随机位置开始，如果出现竞争则通过魔数继续随机移动，反之则线性移动，直到所有队列上的相同校验连续两次出现为空，则说明没有任何任务可以窃取，因此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;
}

/**
 * 如果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;
}

/**
 *如果不能窃取到任务，那么就将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;
}