java并发编程(4)：ForkJoinPool框架源码详解

作者: 桥头放牛娃 | 来源:发表于2019-10-30 23:11 被阅读0次

1、ForkJoinPool简介

ForkJoinPool运用fork-join的原理，使用分而治之的思想，将大任务进行拆分，直到拆分成无法可再拆分的最小单元，并将拆分后的任务分配给多线程执行，最终再将执行结果进行join。同时利用工作窃取算法，使得任务能及时被空闲线程处理。故ForkJoinPool适于可将大任务分割成类似的小任务的场景。

1.1、ForkJoinPool类继承结构

ForkJoinPool也继承于AbstractExecutorService抽象类，实现ExecutorService相关接口。

ForkJoinPool类继承结构.png

1.2、ForkJoinPool体系主要类简介

ForkJoinTask：提交到ForkJoinPool中的任务，在使用时主要有三个实现。RecursiveTask：可以递归执行的ForkJoinTask；RecursiveAction：无返回值的RecursiveTask；CountedCompleter：执行完成后，触发自定义钩子函数。

ForkJoinWorkerThread：运行 ForkJoinTask 任务的工作线程。每个ForkJoinWorkerThread都关联其所属的ForkJoinPool及其工作队列WorkQueue。

WorkQueue：任务队列，支持LIFO的栈式操作和FIFO的队列操作。

2、ForkJoinTask源码解析

ForkJoinTask将任务fork成足够小的任务，并发解决这些小任务，然后将这些小任务结果join。这种思想充分利用了CPU的多核系统，使得CPU的利用率得到大幅度提升，减少了任务执行时间。通常我们会利用ForkJoinTask的fork方法来分割任务，利用join方法来合并任务。

2.1、ForkJoinTask任务状态

statue为ForkJoinTask的状态，其初始状态为0，标识正则处理任务状态；NORMAL：标识任务正常结束；CANCELLED：标识任务被取消；EXCEPTIONAL：标识任务执行异常；SIGNAL：表示有依赖当前任务结果的任务，需要执行完成后进行通知。

//任务状态，初始值为0
volatile int status; // accessed directly by pool and workers
// 任务状态的掩码
static final int DONE_MASK   = 0xf0000000;  
// 正常状态，负数，标识任务已经完成
static final int NORMAL      = 0xf0000000;  
// 任务取消，非负，<NORMAL
static final int CANCELLED   = 0xc0000000;  
// 任务异常，非负，<CANCELLED
static final int EXCEPTIONAL = 0x80000000;  
// 通知状态，>= 1<<16，有其他任务依赖当前任务，任务结束前，通知其他任务join当前任务的结果。
static final int SIGNAL      = 0x00010000;  
// 低位掩码
static final int SMASK       = 0x0000ffff;

2.2、主要方法实现

ForkJoinTask的主要方法有异步执行方法fork()，获取结果方法join()，执行任务方法invoke系列，其他获取状态即结果等方法。

fork()源码解析：

public final ForkJoinTask<V> fork() {
    Thread t;
    //如果线程类型为ForkJoinWorkerThread，则将任务推入workQueue进行处理,
    //否则，交由ForkJoinPool的common线程池进行处理
    if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
        ((ForkJoinWorkerThread)t).workQueue.push(this);
    else
        ForkJoinPool.common.externalPush(this);
    return this;
}

join()源码解析：

public final V join() {
    int s;
    //调用doJoin()进行任务的执行，若任务结果为非正常完成，则根据状态抛出不同的异常，
    //如若状态为CANCELLED，则抛出CancellationException()，异常；
    //若状态为EXCEPTIONAL，则抛出包装后的异常
    if ((s = doJoin() & DONE_MASK) != NORMAL)
        reportException(s);
    return getRawResult();
}

private int doJoin() {
    int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
    //1、若任务状态为正常完成（status < 0），则返回任务的正常完成状态；
    //2、若执行任务的当前线程类型为ForkJoinWorkerThread，且将任务从线程的工作队列中移除成功，
    //则调用doExec()执行任务，若任务执行状态为正常结束，则返回状态，否则awaitJoin()等待任务结束。
    //3、否则调用externalAwaitDone()等待任务执行完成。
    return (s = status) < 0 ? s :
        ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
        (w = (wt = (ForkJoinWorkerThread)t).workQueue).
        tryUnpush(this) && (s = doExec()) < 0 ? s :
        wt.pool.awaitJoin(w, this, 0L) :
        externalAwaitDone();
}

final int doExec() {
    int s; boolean completed;
    //任务未完成？
    if ((s = status) >= 0) {
        try {
            //执行任务
            completed = exec();
        } catch (Throwable rex) {
            return setExceptionalCompletion(rex);
        }
        //设置任务状态为正常完成
        if (completed)
            s = setCompletion(NORMAL);
    }
    return s;
}

private int externalAwaitDone() {
    //任务处理
    int s = ((this instanceof CountedCompleter) ? // try helping
             ForkJoinPool.common.externalHelpComplete(
                 (CountedCompleter<?>)this, 0) :
             ForkJoinPool.common.tryExternalUnpush(this) ? doExec() : 0);
    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;
}

invoke系列方法源码解析：

public final V invoke() {
    int s;
    //执行任务并返回状态，处理同doJoin()类似
    if ((s = doInvoke() & DONE_MASK) != NORMAL)
        reportException(s);
    return getRawResult();
}


private int doInvoke() {
    int s; Thread t; ForkJoinWorkerThread wt;
    //执行任务并获取任务状态，状态<0表示正常完成，否则等待任务完成
    return (s = doExec()) < 0 ? s :
        ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
        (wt = (ForkJoinWorkerThread)t).pool.
        awaitJoin(wt.workQueue, this, 0L) :
        externalAwaitDone();
}

//处理两个任务
public static void invokeAll(ForkJoinTask<?> t1, ForkJoinTask<?> t2) {
    int s1, s2;
    //提交任务t2，交由线程池执行
    t2.fork();
    //执行任务t1并获取直接结果的任务状态
    if ((s1 = t1.doInvoke() & DONE_MASK) != NORMAL)
        t1.reportException(s1);
    //获取任务t2的直接结果状态
    if ((s2 = t2.doJoin() & DONE_MASK) != NORMAL)
        t2.reportException(s2);
}

//处理多个任务
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];
        //任务为空则跑NPE异常
        if (t == null) {
            if (ex == null)
                ex = new NullPointerException();
        }
        //非最后一个任务，则推入线程池执行
        else if (i != 0)
            t.fork();
        //最后一个任务直接调用doInvoke()执行    
        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);
}

获取任务状态及执行结果等方法：

方法名	说明
cancel	取消任务
isDone	判断任务是否正常完成
isCancelled	判断任务是否已取消
isCompletedAbnormally	判断任务是否非正常完成，如被取消或任务执行异常等
isCompletedNormally	判断任务是否执行正常完成，及任务状态是否为NORMAL
getException	获取任务执行的异常结果
completeExceptionally	将任务状态设置为异常，并设置异常结果
complete	将任务设置正常结束，并设置任务执行结果
get	获取任务执行结果，若任务取消或异常，则抛出异常；否则返回任务执行结果

3、WorkQueue源码详解

WorkQueue为ForkJoinPool的工作队列，其封装提交的任务ForkJoinTask、线程池ForkJoinPool、执行线程ForkJoinWorkerThread、及其他任务相关数据等。

3.1、主要属性说明

//
volatile int scanState;    // versioned, <0: inactive; odd:scanning
int stackPred;             // pool stack (ctl) predecessor
int nsteals;               // number of steals
int hint;                  // randomization and stealer index hint
int config;                // pool index and mode
volatile int qlock;        // 1: locked, < 0: terminate; else 0
volatile int base;         // index of next slot for poll
int top;                   // index of next slot for push
ForkJoinTask<?>[] array;   // the elements (initially unallocated)
final ForkJoinPool pool;   // the containing pool (may be null)
final ForkJoinWorkerThread owner; // owning thread or null if shared
volatile Thread parker;    // == owner during call to park; else null
volatile ForkJoinTask<?> currentJoin;  // task being joined in awaitJoin
volatile ForkJoinTask<?> currentSteal; // mainly used by helpStealer

scanState：如果WorkQueue没有属于自己的owner(下标为偶数的都没有),该值为 inactive 也就是一个负数；如果有自己的owner，该值的初始值为其在WorkQueue[]数组中的下标，也肯定是个奇数；如果这个值，变成了偶数，说明该队列所属的Thread正在执行Task。
stackPred：前任池（WorkQueue[]）索引，由此构成一个栈；
config：index | mode。如果下标为偶数的WorkQueue,则其mode是共享类型。如果有自己的owner 默认是 LIFO；
**qlock: **锁标识,在多线程往队列中添加数据，会有竞争，使用此标识抢占锁。1: locked, < 0: terminate; else 0
base：worker steal的偏移量,因为其他的线程都可以偷该队列的任务,所有base使用volatile标识。
top：owner执行任务的偏移量。
parker：如果 owner 挂起，则使用该变量做记录挂起owner的线程。
**currentJoin: **当前正在join等待结果的任务。
currentSteal：当前执行的任务是steal过来的任务，该变量做记录。

3.2、主要方法说明

入队方法：

//工作线程将任务提交到其对应的工作队列中
final void push(ForkJoinTask<?> task) {
    ForkJoinTask<?>[] a; ForkJoinPool p;
    int b = base, s = top, n;
    if ((a = array) != null) {    // ignore if queue removed
        int m = a.length - 1;     // fenced write for task visibility
        //计算放置任务的位置，并将任务保存到队列中
        U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task);
        //修改顶部数据
        U.putOrderedInt(this, QTOP, s + 1);
        //
        if ((n = s - b) <= 1) {
            if ((p = pool) != null)
                p.signalWork(p.workQueues, this);
        }
        //队列已满，则进行扩容
        else if (n >= m)
            growArray();
    }
}

出队方法：

final ForkJoinTask<?> pop() {
   ForkJoinTask<?>[] a; ForkJoinTask<?> t; int m;
   if ((a = array) != null && (m = a.length - 1) >= 0) {
       for (int s; (s = top - 1) - base >= 0;) {
           long j = ((m & s) << ASHIFT) + ABASE;
           //数组下班j处有任务，则cas获取获取任务，并修改top值
           if ((t = (ForkJoinTask<?>)U.getObject(a, j)) == null)
               break;
           if (U.compareAndSwapObject(a, j, t, null)) {
               U.putOrderedInt(this, QTOP, s);
               return t;
           }
       }
   }
   return null;
}

//从base到top获取任务
final ForkJoinTask<?> poll() {
   ForkJoinTask<?>[] a; int b; ForkJoinTask<?> t;
   while ((b = base) - top < 0 && (a = array) != null) {
       //获取base数据偏移量
       int j = (((a.length - 1) & b) << ASHIFT) + ABASE;
       //获取base处任务
       t = (ForkJoinTask<?>)U.getObjectVolatile(a, j);
       //base未被改动？
       if (base == b) {
           if (t != null) {
               //cas更改base处的数据，同时base+1
               if (U.compareAndSwapObject(a, j, t, null)) {
                   base = b + 1;
                   return t;
               }
           }
           //任务全部取完？
           else if (b + 1 == top) // now empty
               break;
       }
   }
   return null;
}

final ForkJoinTask<?> peek() {
   ForkJoinTask<?>[] a = array; int m;
   if (a == null || (m = a.length - 1) < 0)
       return null;
   //判断任务队列是FIFO或LILF？    
   int i = (config & FIFO_QUEUE) == 0 ? top - 1 : base;
   int j = ((i & m) << ASHIFT) + ABASE;
   //获取指定顶部或底部的任务
   return (ForkJoinTask<?>)U.getObjectVolatile(a, j);
}

执行任务：

final void execLocalTasks() {
    int b = base, m, s;
    ForkJoinTask<?>[] a = array;
    if (b - (s = top - 1) <= 0 && a != null &&
        (m = a.length - 1) >= 0) {
        //队列类型为FIFO？    
        if ((config & FIFO_QUEUE) == 0) {
            //遍历任务并执行
            for (ForkJoinTask<?> t;;) {
                if ((t = (ForkJoinTask<?>)U.getAndSetObject
                     (a, ((m & s) << ASHIFT) + ABASE, null)) == null)
                    break;
                U.putOrderedInt(this, QTOP, s);
                t.doExec();
                if (base - (s = top - 1) > 0)
                    break;
            }
        }
        else
            pollAndExecAll();
    }
}

//执行任务
final void runTask(ForkJoinTask<?> task) {
    if (task != null) {
        //设置WorkQueue状态为执行任务状态
        scanState &= ~SCANNING; // mark as busy
        //执行窃取的任务
        (currentSteal = task).doExec();
        U.putOrderedObject(this, QCURRENTSTEAL, null); // release for GC
        //执行所有本地
        execLocalTasks();
        ForkJoinWorkerThread thread = owner;
        if (++nsteals < 0)      // collect on overflow
            transferStealCount(pool);
        scanState |= SCANNING;
        if (thread != null)
            thread.afterTopLevelExec();
    }
}

4、ForkJoinWorkerThread源码解析

ForkJoinWorkerThread为ForkJoinPool的运行线程实现类，其关联了对应的ForkJoinPool及WorkQueue。

构造函数：

//构造函数
protected ForkJoinWorkerThread(ForkJoinPool pool) {
    // Use a placeholder until a useful name can be set in registerWorker
    super("aForkJoinWorkerThread");
    this.pool = pool;
    this.workQueue = pool.registerWorker(this);
}

执行任务的钩子函数：

//线程启动时的钩子函数
protected void onStart() {
}
//线程结束的钩子函数
protected void onTermination(Throwable exception) {
}

执行任务：

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

//ForkJoinPool#runWorker（）
final void runWorker(WorkQueue w) {
    w.growArray();                   // allocate queue
    int seed = w.hint;               // initially holds randomization hint
    int r = (seed == 0) ? 1 : seed;  // avoid 0 for xorShift
    for (ForkJoinTask<?> t;;) {
        //扫描任务，并执行任务
        if ((t = scan(w, r)) != null)
            w.runTask(t);
        //等待窃取任务  
        else if (!awaitWork(w, r))
            break;
        r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift
    }
}

5、ForkJoinPool源码详解

5.1、主要属性

// Bounds
//低16位掩码，索引的最大位数，
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
//偶数槽位数，最多64个偶数槽位（0x007e = 0111 1110,有效的是中间6个1的位置，111111 = 63，再加上000000（0槽位），总共64个）
static final int SQMASK       = 0x007e;        // max 64 (even) slots

// Masks and units for WorkQueue.scanState and ctl sp subfield
//WorkQueue的状态：正在扫描任务
static final int SCANNING     = 1;             // false when running tasks
//WorkQueue的状态：非活动状态
static final int INACTIVE     = 1 << 31;       // must be negative
//版本号（防止CAS的ABA问题）
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
//任务队列模式为LIFO
static final int LIFO_QUEUE   = 0;
//任务队列模式为FIFO
static final int FIFO_QUEUE   = 1 << 16;
//任务队列模式为共享模式
static final int SHARED_QUEUE = 1 << 31;       // must be negative


//线程工厂类
public static final ForkJoinWorkerThreadFactory
    defaultForkJoinWorkerThreadFactory;

//默认的公共线程池
static final ForkJoinPool common;

//并行度
static final int commonParallelism;

//最大备用线程数
private static int commonMaxSpares;

//线程变化序列号
private static int poolNumberSequence;


//ctl的低32位掩码
private static final long SP_MASK    = 0xffffffffL;
//ctl的高32位掩码
private static final long UC_MASK    = ~SP_MASK;

// Active counts
//活跃线程的计算shift
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
//工作线程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); // 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

5.2、状态说明

ctl参数说明：

ctl参数说明.png

字段ctl是ForkJoinPool的核心状态，它是一个64位的long类型数值，包含4个16位子字段:

AC: 活动的工作线程数量减去目标并行度（目标并行度：最大的工作线程数量，所以AC一般是负值，等于0时，说明活动线程已经达到饱和了）
TC: 总的工作线程数量总数减去目标并行度（TC一般也是负值，等于0时，说明总的工作线程已经饱和，并且，AC一般小于等于TC）
SS: 栈顶工作线程状态和版本数（每一个线程在挂起时都会持有前一个等待线程所在工作队列的索引，由此构成一个等待的工作线程栈，栈顶是最新等待的线程，第一位表示状态1.不活动 0.活动，后15表示版本号，标识ID的版本-最后16位）。
ID: 栈顶工作线程所在工作队列的池索引。

runState状态说明：

STARTED 1
STOP 1 << 1
TERMINATED 1<<2
SHUTDOWN 1<<29
RSLOCK 1<<30
RSIGNAL 1<<31

runState记录了线程池的运行状态，特别地，除了SHUTDOWN是负数外，其他值都是正数，RSLOCK和RSIGNAL是跟锁相关。

5.3、主要方法

5.3.1、构造方法

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

5.3.2、提交任务

invoke/execute/submit任务提交：

//提交任务并等待任务执行完成，然后返回执行结果
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);
}

//提交任务并返回任务，ForkJoinTask可获取任务的异步执行结果
public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
    if (task == null)
        throw new NullPointerException();
    externalPush(task);
    return task;
}

提交任务主要有三中方法，invoke()，execute()，submit()，它们最终都是调用externalPush()进行处理，都属于外部提交，置于偶数索引的工作队列。

externalPush()添加任务：

final void externalPush(ForkJoinTask<?> task) {
    WorkQueue[] ws; WorkQueue q; int m;
    //探针值，用于计算WorkQueue槽位索引
    int r = ThreadLocalRandom.getProbe();
    int rs = runState;
    if ((ws = workQueues) != null  //线程池不为空
        && (m = (ws.length - 1)) >= 0   //线程池长度大于0
        && (q = ws[m & r & SQMASK]) != null  //获取偶数槽位的WorkQueue
        && r != 0 && rs > 0  //探针值不为0
        &&U.compareAndSwapInt(q, QLOCK, 0, 1)) { //加锁
        ForkJoinTask<?>[] a; int am, n, s;
        if ((a = q.array) != null //线程的任务队列不为空
            &&(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为top+1
            U.putIntVolatile(q, QLOCK, 0); //解锁
            //若之前的任务数<=1，则此槽位的线程可能在等待，同时可能其他槽位的线程也在等，此时需要唤醒线程来执行任务
            if (n <= 1)
                signalWork(ws, q);
            return;
        }
        U.compareAndSwapInt(q, QLOCK, 1, 0); //添加任务失败，则解锁
    }
    //若if条件中有不满足的，或是添加任务失败，则通过externalSubmit()来添加任务
    externalSubmit(task);
}

signalWork()唤醒worker线程：

final void signalWork(WorkQueue[] ws, WorkQueue q) {
    long c; int sp, i; WorkQueue v; Thread p;
    while ((c = ctl) < 0L) {     //活跃线程数太少，则创建工作线程
        if ((sp = (int)c) == 0) {    //无空闲线程？
            // (c & ADD_WORKER) != 0L，说明TC的最高位为1，为负值，而TC = 总的线程数 - 并行度 < 0，
            // 表示总的线程数 < 并行度，说明工作线程的个数还很少
            if ((c & ADD_WORKER) != 0L) 
                tryAddWorker(c); //尝试添加线程
            break;
        }
        //未开始或已停止
        if (ws == null)                            // unstarted/terminated
            break;
        // 空闲线程栈顶端线程的所属工作队列索引（正常来讲，应该小于WorkQueue[]的长度的）   
        if (ws.length <= (i = sp & SMASK))         // terminated
            break;
        //正则终止？      
        if ((v = ws[i]) == null)                   // terminating
            break;
        // 作为下一个scanState待更新的值（增加了版本号，并且调整为激活状态）   
        int vs = (sp + SS_SEQ) & ~INACTIVE;        // next scanState
        // 如果d为0，则说明scanState还为更新过，然后才考虑CAS ctl
        int d = sp - v.scanState;                  // screen CAS
        // 下一个ctl的值，AC + 1 | 上一个等待线程的索引
        long nc = (UC_MASK & (c + AC_UNIT)) | (SP_MASK & v.stackPred);
        if (d == 0 && U.compareAndSwapLong(this, CTL, c, nc)) {
            v.scanState = vs;                      // activate v
            // 如果线程阻塞了，唤醒它
            if ((p = v.parker) != null)
                U.unpark(p);
            break;
        }
        // 没有任务，直接退出
        if (q != null && q.base == q.top)          // no more work
            break;
    }
}

externalSubmit()添加任务：

externalSubmit()包含了三方面的操作：

若线程未初始化，则初始化线程池，长度是2的幂次方；
若选中的槽位位空，则初始化一个共享模式的工作队列；
若选中槽位不为空，则获取任务队列，并将任务提交到任务队列，成功则唤醒沉睡的线程；若失败则专业槽位。

private void externalSubmit(ForkJoinTask<?> task) {
    int r;                                    // initialize caller's probe
    //初始化当前线程的探针值，用于计算WorkQueue的索引
    if ((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 terminate
            throw new RejectedExecutionException();
        }
        //1、线程池状态为还未初始化？；
        //2、线程池为空？
        //3、线程池中工作线程数为0？
        else if ((rs & STARTED) == 0 ||     // initialize
                 ((ws = workQueues) == null || 
                 (m = ws.length - 1) < 0)) {
            int ns = 0;
            //加锁
            rs = lockRunState();
            try {
                //加锁后再次判断线程池状态，避免重复初始化
                if ((rs & STARTED) == 0) {
                    U.compareAndSwapObject(this, STEALCOUNTER, null,
                                           new AtomicLong());
                    // create workQueues array with size a power of two
                    int p = config & SMASK; // ensure at least 2 slots
                    //保证n是2的幂次方
                    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];
                    ns = STARTED;
                }
            } finally {
                //解锁
                unlockRunState(rs, (rs & ~RSLOCK) | ns);
            }
        }
        //获取随机偶数槽位的WorkQueue
        else if ((q = ws[k = r & m & SQMASK]) != null) {
            //对WorkQueue加锁
            if (q.qlock == 0 && U.compareAndSwapInt(q, QLOCK, 0, 1)) {
                ForkJoinTask<?>[] a = q.array;
                int s = q.top;
                boolean submitted = false; // initial submission or resizing
                try {                      // locked version of push
                    //若WorkQueue的任务队列为空，则初始化任务队列（growArray）
                    if ((a != null && a.length > s + 1 - q.base) ||
                        (a = q.growArray()) != null) {
                        //计算任务索引的下标    
                        int j = (((a.length - 1) & s) << ASHIFT) + ABASE;
                        U.putOrderedObject(a, j, task);
                        U.putOrderedInt(q, QTOP, s + 1);
                        submitted = true;
                    }
                } finally {
                    //解锁
                    U.compareAndSwapInt(q, QLOCK, 1, 0);
                }
                //唤醒挂起的线程
                if (submitted) {
                    signalWork(ws, q);
                    return;
                }
            }
            move = true;                   // move on failure
        }
        //在未加锁的情况下，创建新线程
        else if (((rs = runState) & RSLOCK) == 0) { // create new queue
            q = new WorkQueue(this, null);
            q.hint = r;
            //共享模式
            q.config = k | SHARED_QUEUE;
            //未激活
            q.scanState = INACTIVE;
            //加锁
            rs = lockRunState();           // 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 if busy
        if (move)
            r = ThreadLocalRandom.advanceProbe(r);
    }
}

5.3.3、执行任务

runWorker()是在ForkJoinWorkerThread的run()方法中调用，即在启动worker线程调用的。其主要工作是获取任务并执行任务，若线程池关闭，则等待任务队列的任务执行完成并退出。

final void runWorker(WorkQueue w) {
    w.growArray();                   // allocate queue
    int seed = w.hint;               // initially holds randomization hint
    int r = (seed == 0) ? 1 : seed;  // avoid 0 for xorShift
    for (ForkJoinTask<?> t;;) {
        //扫描任务
        if ((t = scan(w, r)) != null)
            //工作线程执行任务
            w.runTask(t);
            //没有任务执行则等待
        else if (!awaitWork(w, r))
            break;
        //随机值更新    
        r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift
    }
}

scan()扫描任务:

private ForkJoinTask<?> scan(WorkQueue w, int r) {
    WorkQueue[] ws; int m;
    //线程池不为空？
    if ((ws = workQueues) != null && (m = ws.length - 1) > 0 && w != null) {
        int ss = w.scanState;                     // initially non-negative
        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) {
                //有任务
                if ((n = (b = q.base) - q.top) < 0 &&
                    (a = q.array) != null) {      // non-empty
                    //数组地址
                    long i = (((a.length - 1) & b) << ASHIFT) + ABASE;
                    if ((t = ((ForkJoinTask<?>)
                              U.getObjectVolatile(a, i))) != null &&
                        q.base == b) {
                        //获取任务并更新base等索引信息    
                        if (ss >= 0) {
                            if (U.compareAndSwapObject(a, i, t, null)) {
                                q.base = b + 1;
                                //通知其他线程
                                if (n < -1)       // signal others
                                    signalWork(ws, q);
                                return t;
                            }
                        }
                        //设置WorkQueue的状态
                        else if (oldSum == 0 &&   // try to activate
                                 w.scanState < 0)
                            tryRelease(c = ctl, ws[m & (int)c], AC_UNIT);
                    }
                    if (ss < 0)                   // refresh
                        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;
            }
            //未扫描到任务，准备inactive此工作队列
            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;
}

awaitWork()为等待任务。若工作线程未获取到任务，则会执行此方法。

private boolean awaitWork(WorkQueue w, int r) {
        if (w == null || w.qlock < 0) // w已经终止，返回false，不再扫描任务
            return false;
        for (int pred = w.stackPred, spins = SPINS, ss;;) { 
            if ((ss = w.scanState) >= 0) // 如果已经active，跳出，返回true，继续扫描任务
                break;
            else if (spins > 0) { // 如果spins > 0，自旋等待
                r ^= r << 6;
                r ^= r >>> 21;
                r ^= r << 7;
                if (r >= 0 && --spins == 0) { // 随机消耗自旋次数
                    WorkQueue v;
                    WorkQueue[] ws;
                    int s, j;
                    AtomicLong sc;
                    if (pred != 0 // 除了自己，还有等待的线程-工作队列
                            && (ws = workQueues) != null // 线程池还在
                            && (j = pred & SMASK) < ws.length // 前任索引还在池范围内
                            && (v = ws[j]) != null // 前任任务队列还在
                            && (v.parker == null || v.scanState >= 0)) // 前任线程已经唤醒，且工作队列已经激活
                        spins = SPINS; // 上面的一系列判断表明，很快就有任务了，先不park，继续自旋
                }
            } else if (w.qlock < 0) // 自旋之后，再次检查工作队列是否终止，若是，退出扫描
                return false;
            else if (!Thread.interrupted()) { // 如果线程中断了，清除中断标记，不考虑park，否则进入该分支
                long c, prevctl, parkTime, deadline;
                int ac = (int) ((c = ctl) >> AC_SHIFT) + (config & SMASK); // 计算活跃线程的个数
                if ((ac <= 0 && tryTerminate(false, false)) || (runState & STOP) != 0) // 线程池正在终止，退出扫描
                    return false;
                if (ac <= 0 && ss == (int) c) { // 自己是栈顶等待者
                    prevctl = (UC_MASK & (c + AC_UNIT)) | (SP_MASK & pred); // 设置为前一次的ctl
                    int t = (short) (c >>> TC_SHIFT); // 总的线程数
                    if (t > 2 && U.compareAndSwapLong(this, CTL, c, prevctl)) // 总线程数过多，直接退出扫描
                        return false;
                    parkTime = IDLE_TIMEOUT * ((t >= 0) ? 1 : 1 - t); // 计算等待时间
                    deadline = System.nanoTime() + parkTime - TIMEOUT_SLOP;
                } else
                    prevctl = parkTime = deadline = 0L;
                Thread wt = Thread.currentThread();
                U.putObject(wt, PARKBLOCKER, this); // 加锁
                w.parker = wt; // 设置parker，准备阻塞
                if (w.scanState < 0 && ctl == c) // 阻塞前再次检查状态
                    U.park(false, parkTime);
                U.putOrderedObject(w, QPARKER, null); // 唤醒后，置空parker
                U.putObject(wt, PARKBLOCKER, null); // 解锁
                if (w.scanState >= 0) // 已激活，跳出继续扫描
                    break;
                if (parkTime != 0L && ctl == c && deadline - System.nanoTime() <= 0L
                        && U.compareAndSwapLong(this, CTL, c, prevctl)) // 超时，未等到任务，跳出，不再执行扫描任务，削减工作线程
                    return false; // shrink pool
            }
        }
        return true;
    }