parallelStream 底层 ForkJoinPool 实现

柔情只为你懂 2023-10-15 22:07 19阅读 0赞

ForkJoinPool源码解析  
前言  
Java 8中Stream是对集合（Collection）对象功能的增强，其特性之一提供了流的并行处理 -> parallelStream。本篇来分析下项目中经常使用的parallelStream底层实现机制。

正文  
以main函数为切入点分析, 采用parallelStream来处理集合数据。

public static void main(String\[\] args) \{  
  
List<Integer> list = new ArrayList<>();  
list.addAll(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16));  
//1.1 .parallelStream()  
//1.2 .forEach(e ->\{\})  
list.parallelStream().forEach(e -> \{  
try \{  
Thread.currentThread().getThreadGroup().list();  
System.out.println("执行线程名："+Thread.currentThread() + " id " + e);  
\} catch (InterruptedException h) \{  
h.printStackTrace();  
\}  
\});  
\}  
1.1 list调用 parallelStream() 返回一个并行流。

default Stream<E> parallelStream() \{  
return StreamSupport.stream(spliterator(), true);  
\}  
spliterator()方法返回一个迭代器对象。Spliterator是java 8引入的接口，用于并行遍历和划分源元素的迭代器。

//ArrayList类：  
@Override  
public Spliterator<E> spliterator() \{  
return new ArrayListSpliterator<>(this, 0, -1, 0);  
\}  
1.2 forEach()用于遍历集合数据，该方法用于接收一个Consumer接口函数。Consumer是java 8提供的函数式接口之一。

Stream接口：  
  
void forEach(Consumer<? super T> action);  
Java 8函数式接口  
函数式接口定义：有且仅有一个抽象方法。

常用函数接口：

Consumer 消费接口，接收T（泛型）类型参数，没有返回值。

Function 该接口的抽象函数式有返回值。

Predicate 返回值固定的boolean。

从forEach出发  
//流的源阶段  
static class Head<E\_IN, E\_OUT> extends ReferencePipeline<E\_IN, E\_OUT> \{  
  
@Override  
public void forEach(Consumer<? super E\_OUT> action) \{  
if (!isParallel()) \{  
sourceStageSpliterator().forEachRemaining(action);  
\}  
else \{  
//并行实现  
super.forEach(action);  
\}  
\}  
\}  
//流的源阶段与中间阶段  
abstract class ReferencePipeline<P\_IN, P\_OUT>  
extends AbstractPipeline<P\_IN, P\_OUT, Stream<P\_OUT>>  
implements Stream<P\_OUT> \{  
@Override  
public void forEach(Consumer<? super P\_OUT> action) \{  
//makeRef方法构造一个TerminalOp对象，该操作对流的每个元素执行操作。  
evaluate(ForEachOps.makeRef(action, false));  
\}  
\}  
//使用终端操作评估管道以生成结果。  
final <R> R evaluate(TerminalOp<E\_OUT, R> terminalOp) \{  
assert getOutputShape() == terminalOp.inputShape();  
if (linkedOrConsumed)  
throw new IllegalStateException(MSG\_STREAM\_LINKED);  
//标记管道已使用  
linkedOrConsumed = true;  
  
return isParallel()  
? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))  
: terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));  
\}  
//静态抽象类：ForEachOp  
@Override  
public <S> Void evaluateParallel(PipelineHelper<T> helper,  
Spliterator<S> spliterator) \{  
if (ordered)  
new ForEachOrderedTask<>(helper, spliterator, this).invoke();  
else  
//1.3 用于执行并行操作的任务  
new ForEachTask<>(helper, spliterator, helper.wrapSink(this)).invoke();  
return null;  
\}  
forEach阶段简单的介绍了下，重点在下文。

来张ForkJoinPool运行流程图：

、

正式分析ForkJoinPool源码之前，先看相关类的属性定义

ForkJoinPool线程池状态定义：

public class ForkJoinPool extends AbstractExecutorService \{  
private static final int RSLOCK = 1; //锁定态  
private static final int RSIGNAL = 1 << 1; //唤醒态  
private static final int STARTED = 1 << 2; //开始态  
private static final int STOP = 1 << 29; //停止态  
private static final int TERMINATED = 1 << 30; //终止态  
private static final int SHUTDOWN = 1 << 31; //关闭态  
//线程池状态  
volatile int runState;  
  
//线程池的控制变量字段  
/\*\*  
\* long类型ctl字段为64位，包含4个16位子字段  
\* AC: 活动线程数 49-64位  
\* TC: 总线程数 33-48位  
\* SS: 顶部等待线程的版本计数和状态(第32位为1表示inative, 其余15位表示版本计数) 17-32位  
\* ID: WorkQueue在WorkQueue\[\]中索引位置 1-16位  
\* SP: 低32位。  
\*  
\* AC为负：没有足够活跃的线程  
\* TC为负：总线程数不足  
\* SP非0：有等待的线程。  
\*\*/  
volatile long ctl;  
  
\}

ctl图示

WorkQueue属性定义：

static final class WorkQueue \{  
......  
//扫描状态，<0：不活跃；奇数：扫描任务；偶数：执行任务  
volatile int scanState;  
//线程探针值，可用于计算workqueue队列在数组中槽位  
int hint;  
//高16位记录队列模式（FIFO、LIFO），低16位记录工作队列数组槽位（workqueue在数组中索引下标）  
int config;  
//队列状态，1：锁定；0：未锁定；<0：终止  
volatile int qlock;  
//下一个poll操作的索引(栈底/队列头)，线程从base端窃取任务  
volatile int base; // index of next slot for poll  
//下一个push操作的索引(栈顶/队列尾)  
int top; // index of next slot for push  
//任务数组  
ForkJoinTask<?>\[\] array;  
//队列所属线程池（可能为空）  
final ForkJoinPool pool;  
//当前WorkQueue所属工作线程，如果是外部提交的任务，则owner为空  
final ForkJoinWorkerThread owner; // owning thread or null if shared  
//当前正在进行join等待的其它任务  
volatile ForkJoinTask<?> currentJoin; // task being joined in awaitJoin  
//当前正在偷取的任务  
volatile ForkJoinTask<?> currentSteal; // mainly used by helpStealer  
......  
\}  
config字段高16位队列模式图：

ForkJoinTask任务状态定义：

public abstract class ForkJoinTask<V> implements Future<V>, Serializable \{  
//任务运行状态，初始为0  
volatile int status;  
static final int DONE\_MASK = 0xf0000000; //任务状态掩码  
static final int NORMAL = 0xf0000000; //完成状态  
static final int CANCELLED = 0xc0000000; //取消状态  
static final int EXCEPTIONAL = 0x80000000; //异常状态  
static final int SIGNAL = 0x00010000; //信号状态，用于唤醒阻塞线程  
static final int SMASK = 0x0000ffff; //低位掩码  
\}  
ForkJoinTask初探  
1.3 跟踪 invoke方法

//ForkJoinTask抽象类  
public final V invoke() \{  
int s;  
//获取任务执行状态  
if ((s = doInvoke() & DONE\_MASK) != NORMAL)  
reportException(s);  
//计算结果  
return getRawResult();  
\}  
  
private int doInvoke() \{  
int s; Thread t; ForkJoinWorkerThread wt;  
return (s = doExec()) < 0 ? s :  
((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?  
(wt = (ForkJoinWorkerThread)t).pool.  
//1.4 帮助执行或阻塞线程  
awaitJoin(wt.workQueue, this, 0L) :  
//阻塞一个非工作线程直到完成  
externalAwaitDone();  
\}  
externalAwaitDone()方法这里就不贴代码了，简单说下逻辑。后面扫描任务的scan方法有详解，流程差不多。

外部线程，也就是前面例子中的main线程执行完doExec()完成任务拆分逻辑后，此时任务状态还不是完成态，则由main线程帮助执行子任务。

main线程先执行自己初始化创建队列中的任务，后扫描其它工作队列Workqueue中的任务。

上面流程执行完后，判断任务状态是否为完成态，如果不是完成态，说明有工作线程在执行任务，cas操作设置状态为唤醒态，然后main线程调用wait方法阻塞自己，等其它线程唤醒。工作线程任务执行完后会调用notifyAll唤醒主线程。最终任务全部执行完毕返回。

1.4 java.util.concurrent.ForkJoinPool\#awaitJoin：

//帮助或阻塞，直到给定的任务完成或超时  
final int awaitJoin(WorkQueue w, ForkJoinTask<?> task, long deadline) \{  
int s = 0;  
if (task != null && w != null) \{  
//记录前一个等待的任务  
ForkJoinTask<?> prevJoin = w.currentJoin;  
//设置当前任务  
U.putOrderedObject(w, QCURRENTJOIN, task);  
CountedCompleter<?> cc = (task instanceof CountedCompleter) ?  
(CountedCompleter<?>)task : null;  
for (;;) \{  
//任务执行完  
if ((s = task.status) < 0)  
break;  
if (cc != null)  
//帮助完成任务，先执行自己队列任务，再扫描其它队列任务  
helpComplete(w, cc, 0);  
//如果队列为空，说明任务被偷  
else if (w.base == w.top || w.tryRemoveAndExec(task))  
//遍历队列找到当前偷取此任务的队列，执行偷取者队列的任务  
helpStealer(w, task);  
if ((s = task.status) < 0)  
break;  
long ms, ns;  
if (deadline == 0L)  
ms = 0L;  
else if ((ns = deadline - System.nanoTime()) <= 0L)  
break;  
else if ((ms = TimeUnit.NANOSECONDS.toMillis(ns)) <= 0L)  
ms = 1L;  
//补偿措施，唤醒线程或创建线程执行任务  
if (tryCompensate(w)) \{  
task.internalWait(ms);  
U.getAndAddLong(this, CTL, AC\_UNIT);  
\}  
\}  
U.putOrderedObject(w, QCURRENTJOIN, prevJoin);  
\}  
return s;  
\}  
跟踪doExec方法，执行路径：ForkJoinTask\#doExec -> CountedCompleter\#exec -> ForEachTask\#compute。

来看看compute()方法做了哪些事情。

//ForEachTask静态类  
public void compute() \{  
//ArrayList.ArrayListSpliterator迭代器  
Spliterator<S> rightSplit = spliterator, leftSplit;  
//预估值，初始化时为集合总数，后面为任务分割后子任务内元素和  
long sizeEstimate = rightSplit.estimateSize(), sizeThreshold;  
if ((sizeThreshold = targetSize) == 0L)  
//目标值 = 预估值 / (总线程数 \* 4)  
targetSize = sizeThreshold = AbstractTask.suggestTargetSize(sizeEstimate);  
boolean isShortCircuit = StreamOpFlag.SHORT\_CIRCUIT.isKnown(helper.getStreamAndOpFlags());  
boolean forkRight = false;  
Sink<S> taskSink = sink;  
ForEachTask<S, T> task = this;  
//任务切分逻辑  
while (!isShortCircuit || !taskSink.cancellationRequested()) \{  
// 切分直至子任务的大小小于阈值  
if (sizeEstimate <= sizeThreshold ||  
//trySplit()将rightSplit平均分，返回平分后左边的任务  
//待切分任务小于等于1，停止切分  
(leftSplit = rightSplit.trySplit()) == null) \{  
//执行函数式接口  
task.helper.copyInto(taskSink, rightSplit);  
break;  
\}  
ForEachTask<S, T> leftTask = new ForEachTask<>(task, leftSplit);  
//原子更新挂起任务数量  
task.addToPendingCount(1);  
ForEachTask<S, T> taskToFork;  
if (forkRight) \{  
forkRight = false;  
rightSplit = leftSplit;  
taskToFork = task;  
task = leftTask;  
\}  
else \{  
forkRight = true;  
taskToFork = leftTask;  
\}  
//将子任务提交到线程池ForkJoinPool  
taskToFork.fork();  
sizeEstimate = rightSplit.estimateSize();  
\}  
task.spliterator = null;  
//如果挂起计数不为0，则递减。最终任务处理完后设置状态为完成态，并唤醒阻塞的线程  
task.propagateCompletion();  
\}  
compute方法主要逻辑就是达到拆分条件就平均拆分任务，待切分任务小于1时，停止切分，去执行任务。对应例子中的consumer函数式接口。

拆分后的任务会提交到队列中，由线程获取任务后继续调用 compute() 来判断拆分还是执行。

继续看fork逻辑：

//ForkJoinTask抽象类  
public final ForkJoinTask<V> fork() \{  
Thread t;  
//如果当前线程为ForkJoin工作线程，则任务放入自身工作线程队列  
if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)  
((ForkJoinWorkerThread)t).workQueue.push(this);  
else  
//如果是外部线程池调用，则任务push到公共的线程池执行，由于本文是main函数开头，所以会执行此处  
ForkJoinPool.common.externalPush(this);  
return this;  
\}  
ForkJoinPool类解析  
来看下公共线程池的创建：  
ForkJoinPool.common = makeCommonPool()  
ForkJoinPool类初始化时会执行静态代码块，静态代码块中会执行makeCommonPool方法，返回ForkJoinPool对象实例

//ForkJoinPool类  
  
//创建并返回一个公共线程池  
private static ForkJoinPool makeCommonPool() \{  
int parallelism = -1;  
ForkJoinWorkerThreadFactory factory = null;  
......  
if (fp != null)  
factory = ((ForkJoinWorkerThreadFactory)ClassLoader.  
getSystemClassLoader().loadClass(fp).newInstance());  
  
if (parallelism < 0 && // default 1 less than \#cores  
//线程池数量，默认cpu核数-1， 用户也可在环境变量中指定数量  
(parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)  
parallelism = 1;  
if (parallelism > MAX\_CAP)  
parallelism = MAX\_CAP;  
return new ForkJoinPool(parallelism, factory, handler, LIFO\_QUEUE,  
"ForkJoinPool.commonPool-worker-");  
\}  
回到externalPush方法：

//外部任务提交  
final void externalPush(ForkJoinTask<?> task) \{  
WorkQueue\[\] ws; WorkQueue q; int m;  
//探针值，用于计算WorkQueue槽位索引。  
//ThreadLocalRandom原理：获取随机数时，每个线程获取自己初始化种子，避免多线程使用同一个原子种子变量，从而导致原子变量的竞争。  
int r = ThreadLocalRandom.getProbe();  
int rs = runState;  
//外部线程第一次进入，此时workQueues还未初始化，if代码不会执行  
if ((ws = workQueues) != null && (m = (ws.length - 1)) >= 0 &&  
//这里会获取偶数槽位，外部提交的任务会放在WorkQueue数组的偶数槽位。后面会配图说明  
(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 &&  
(am = a.length - 1) > (n = (s = q.top) - q.base)) \{  
//计算任务在内存中偏移量  
int j = ((am & s) << ASHIFT) + ABASE;  
//向队列放入任务  
U.putOrderedObject(a, j, task);  
//top值 +1  
U.putOrderedInt(q, QTOP, s + 1);  
//解锁  
U.putIntVolatile(q, QLOCK, 0);  
//任务数小于等于1，尝试创建或激活工作线程  
if (n <= 1)  
signalWork(ws, q);  
return;  
\}  
U.compareAndSwapInt(q, QLOCK, 1, 0);  
\}  
externalSubmit(task);  
\}  
  
//完整版externalPush，用于将任务提交到池中，如果是第一次提交，则初始化工作队列  
private void externalSubmit(ForkJoinTask<?> task) \{  
int r; //线程探针值  
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();  
\}  
//线程池状态不是开始状态，则状态加锁，初始化WorkQueue 数组  
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  
//获取并行度config的值，前面说过config值为cpuh核数-1，比如我的系统是8核，那这里p=8-1=7。  
int p = config & SMASK; // ensure at least 2 slots  
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;  
//数组大小这里可以找到规律。找到大于2倍p的值最近的2^m。由于最后 (n+1)左移了一位，所以n最小值为4。  
//举例：p = 7, 那么 2p = 14, 大于2p的2^m = 16。所以这里m=4, n=16。  
//假如p=3, 2p=6，2^m=8 > 6。 则m=3, n=8。  
workQueues = new WorkQueue\[n\]; //由于我这里p=7, 所以初始化队列数组长度为16。  
ns = STARTED;  
\}  
\} finally \{  
//解锁，状态修改为开始态  
unlockRunState(rs, (rs & ~RSLOCK) | ns);  
\}  
\}  
//判断偶数槽位是否为空  
else if ((q = ws\[k = r & m & SQMASK\]) != null) \{  
//qlock属性定义，初始值0，1：锁定，< 0：终止  
//队列加锁  
if (q.qlock == 0 && U.compareAndSwapInt(q, QLOCK, 0, 1)) \{  
//任务数组  
ForkJoinTask<?>\[\] a = q.array;  
int s = q.top;  
boolean submitted = false;  
try \{  
if ((a != null && a.length > s + 1 - q.base) ||  
//growArray()逻辑：任务数组为空，则创建数组，初始大小为8192，不为空，则2倍扩容。任务移到新数组。  
(a = q.growArray()) != null) \{  
//计算top在内存中偏移量  
int j = (((a.length - 1) & s) << ASHIFT) + ABASE;  
//任务存储到任务数组  
U.putOrderedObject(a, j, task);  
//更新top值。top定义：push操作的索引(栈顶)  
U.putOrderedInt(q, QTOP, s + 1);  
submitted = true;  
\}  
\} finally \{  
//队列解锁  
U.compareAndSwapInt(q, QLOCK, 1, 0);  
\}  
if (submitted) \{  
//1.5 创建或激活工作线程  
signalWork(ws, q);  
return;  
\}  
\}  
move = true; // move on failure  
\}  
//队列数组槽位为空，则创建一个WorkQueue放入数组  
else if (((rs = runState) & RSLOCK) == 0) \{  
//外部线程所属队列是共享队列，参数owner传入null  
q = new WorkQueue(this, null);  
//记录这个探针值  
q.hint = r;  
//记录队列数组槽位和队列模式  
q.config = k | SHARED\_QUEUE;  
//共享队列初始化为inactive状态  
q.scanState = INACTIVE;  
rs = lockRunState(); // publish index  
if (rs > 0 && (ws = workQueues) != null &&  
k < ws.length && ws\[k\] == null)  
ws\[k\] = q;  
unlockRunState(rs, rs & ~RSLOCK);  
\}  
else  
move = true;  
if (move)  
r = ThreadLocalRandom.advanceProbe(r);  
\}  
\}  
  
externalSubmit方法三部曲：

初始化工作队列数组 WorkQueue\[\]，

创建工作队列 WorkQueue，

任务放入任务数组 ForkJoinTask\[\], 更新top值, 创建线程执行任务。

画个图帮助理解任务提交，看到这的点个赞把☺(\*￣︶￣)

m & r & SQMASK必为偶数，所以通过externalSubmit方法提交的任务都添加到了偶数索引的任务队列中（没有绑定的工作线程）。

1.5 java.util.concurrent.ForkJoinPool\#signalWork：

工作线程数不足：创建一个工作线程；

工作线程数足够：唤醒一个空闲（阻塞）的工作线程。

//创建或激活工作线程  
final void signalWork(WorkQueue\[\] ws, WorkQueue q) \{  
long c; int sp, i; WorkQueue v; Thread p;  
//ctl小于0时，活跃的线程数不够  
while ((c = ctl) < 0L) \{  
//取ctl的低32位，如果为0，说明没有等待的线程  
if ((sp = (int)c) == 0) \{  
//取tc的高位，不为0说明总线程数不够，创建线程  
if ((c & ADD\_WORKER) != 0L)  
tryAddWorker(c);  
break;  
\}  
if (ws == null)  
break;  
if (ws.length <= (i = sp & SMASK))  
break;  
if ((v = ws\[i\]) == null)  
break;  
int vs = (sp + SS\_SEQ) & ~INACTIVE;  
int d = sp - v.scanState;  
long nc = (UC\_MASK & (c + AC\_UNIT)) | (SP\_MASK & v.stackPred);  
if (d == 0 && U.compareAndSwapLong(this, CTL, c, nc)) \{  
v.scanState = vs;  
if ((p = v.parker) != null)  
//唤醒线程  
U.unpark(p);  
break;  
\}  
if (q != null && q.base == q.top)  
break;  
\}  
\}  
  
private void tryAddWorker(long c) \{  
boolean add = false;  
do \{  
//活跃线程和总线程数都 +1  
long nc = ((AC\_MASK & (c + AC\_UNIT)) |  
(TC\_MASK & (c + TC\_UNIT)));  
if (ctl == c) \{  
int rs, stop;  
//加锁，并判断线程池状态是否为停止态  
if ((stop = (rs = lockRunState()) & STOP) == 0)  
//cas更新ctl的值  
add = U.compareAndSwapLong(this, CTL, c, nc);  
//解锁  
unlockRunState(rs, rs & ~RSLOCK);  
if (stop != 0)  
break;  
if (add) \{  
//1.6 创建新的工作线程  
createWorker();  
break;  
\}  
\}  
//ADD\_WORKER的第48位是1，其余位都为0，与ctl位与判断TC总线程是否已满，如果线程满了, 那么ctl的第48位为0，0 & ADD\_WORKER = 0  
//(int)c == 0 表示截取ctl的低32位（int是4字节，1字节是8位）  
\} while (((c = ctl) & ADD\_WORKER) != 0L && (int)c == 0);  
\}  
通过ctl的高32位来判断是否需要创建线程或激活线程。

上文有讲，公共线程池数量为cpu核数-1，我电脑是8核，所以总线程数为7。也就是并行度为7。高32位 +7最终等于0。

1.6 java.util.concurrent.ForkJoinPool\#createWorker：

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;  
\}  
线程工厂创建线程。

外部线程提交任务后创建线程，线程扫描到任务后会继续创建线程，直到达到总线程数。

本地测试中某一次线程创建流程图如下，可以看到main线程创建了worker-1和worker-3工作线程，worker-1又创建了worker-2和worker-4工作线程。都属于同一个main线程组。

ForkJoinWorkerThread  
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);  
\}  
\}  
\}  
\}  
java.util.concurrent.ForkJoinPool\#runWorker  
final void runWorker(WorkQueue w) \{  
//初始化或两倍扩容  
w.growArray();  
//获取创建工作队列的线程探针值  
int seed = w.hint;  
int r = (seed == 0) ? 1 : seed;  
for (ForkJoinTask<?> t;;) \{  
//扫描任务  
if ((t = scan(w, r)) != null)  
//1.7 执行获取到的任务  
w.runTask(t);  
else if (!awaitWork(w, r))  
break;  
r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift  
\}  
\}  
  
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;  
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) \{  
//获取base的偏移量  
long i = (((a.length - 1) & b) << ASHIFT) + ABASE;  
//FIFO模式，从base端获取任务  
if ((t = ((ForkJoinTask<?>)  
U.getObjectVolatile(a, i))) != null &&  
q.base == b) \{  
//判断线程是否活跃状态  
if (ss >= 0) \{  
//任务被偷了。更新任务数组a在内存中偏移量为base位置的值为空  
if (U.compareAndSwapObject(a, i, t, null)) \{  
//更新base值  
q.base = b + 1;  
if (n < -1)  
//数组还有任务，创建或激活线程执行任务  
signalWork(ws, q);  
//返回扫描到的任务  
return t;  
\}  
\}  
else if (oldSum == 0 &&  
w.scanState < 0)  
tryRelease(c = ctl, ws\[m & (int)c\], AC\_UNIT);  
\}  
if (ss < 0)  
ss = w.scanState;  
r ^= r << 1; r ^= r >>> 3; r ^= r << 10;  
origin = k = r & m;  
oldSum = checkSum = 0;  
continue;  
\}  
checkSum += b;  
\}  
//m二进制位都为1，(k+1)&m 会遍历工作队列数组所有槽位  
if ((k = (k + 1) & m) == origin) \{  
if ((ss >= 0 || (ss == (ss = w.scanState))) &&  
oldSum == (oldSum = checkSum)) \{  
if (ss < 0 || w.qlock < 0)  
break;  
int ns = ss | INACTIVE;  
//活跃线程-1  
long nc = ((SP\_MASK & ns) |  
(UC\_MASK & ((c = ctl) - AC\_UNIT)));  
w.stackPred = (int)c;  
U.putInt(w, QSCANSTATE, ns);  
if (U.compareAndSwapLong(this, CTL, c, nc))  
ss = ns;  
else  
w.scanState = ss; // back out  
\}  
checkSum = 0;  
\}  
\}  
\}  
return null;  
\}  
scan扫描任务，从其它队列的base端偷取。

base属性加了volatile关键字，保证了共享变量的内存可见性。获取任务后通过cas乐观锁操作将被窃取的队列中任务置位空。

此窃取机制减少了top和base端的竞争(队列线程和窃取线程分别从top和base端操作)，用cas操作也提高了效率。

比如外部线程分割了3个任务，那么top+3=4099,三个任务放在内存偏移量为4096、4097、4098的位置。启动的工作线程从base端偷取任务。如下图：

1.7 java.util.concurrent.ForkJoinPool.WorkQueue\#runTask：

final void runTask(ForkJoinTask<?> task) \{  
if (task != null) \{  
//此时scanState变成了偶数，表示正在执行任务  
scanState &= ~SCANNING; // mark as busy  
//执行窃取到的任务，doExec上面有讲，会在compute方法中执行任务或分割任务，分割的任务放入自己队列的任务数组里。  
(currentSteal = task).doExec();  
//将窃取到的任务字段置空  
U.putOrderedObject(this, QCURRENTSTEAL, null); // release for GC  
//执行自个任务数组里任务。任务来源就是上面分割后的子任务。  
execLocalTasks();  
ForkJoinWorkerThread thread = owner;  
//窃取任务数+1  
if (++nsteals < 0) // collect on overflow  
//任务数累加到pool的stealCounter字段中  
transferStealCount(pool);  
//恢复扫描状态  
scanState |= SCANNING;  
if (thread != null)  
//钩子方法  
thread.afterTopLevelExec();  
\}  
\}  
  
final void execLocalTasks() \{  
int b = base, m, s;  
ForkJoinTask<?>\[\] a = array;  
//任务数组中有任务  
if (b - (s = top - 1) <= 0 && a != null &&  
(m = a.length - 1) >= 0) \{  
//队列模式，判断是否为后进先出模式 LIFO。  
if ((config & FIFO\_QUEUE) == 0) \{  
for (ForkJoinTask<?> t;;) \{  
//从top端取任务  
if ((t = (ForkJoinTask<?>)U.getAndSetObject  
(a, ((m & s) << ASHIFT) + ABASE, null)) == null)  
break;  
//修改内存中top值  
U.putOrderedInt(this, QTOP, s);  
//执行任务  
t.doExec();  
if (base - (s = top - 1) > 0)  
break;  
\}  
\}  
else  
//先进先出模式 FIFO, 轮询从base端取任务运行，直到为空  
pollAndExecAll();  
\}  
\}

runTask方法主要逻辑就是将扫描到的任务执行，执行过程中可能会继续分割。执行完窃取的任务后，就执行自个队列里的任务。

至此，ForkJoinPool就分析到这。

总结：

大概可以用一句话总结ForkJoinPool原理：一个可以并行执行任务的线程池，可以处理一个可递归划分的任务并获取结果。

参考博文：

\[1\] https://blog.csdn.net/qq\_27785239/article/details/103395079  
\[2\] https://blog.csdn.net/yusimiao/article/details/114009972  
————————————————  
版权声明：本文为CSDN博主「阿飞技术」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。  
原文链接：https://blog.csdn.net/u011385940/article/details/121845167