多執行緒高併發程式設計(8) -- Fork/Join原始碼分析
阿新 • • 發佈：2020-05-11
一.概念
Fork/Join就是將一個大任務分解(fork)成許多個獨立的小任務，然後多執行緒並行去處理這些小任務，每個小任務處理完得到結果再進行合併(join)得到最終的結果。
流程：任務繼承RecursiveTask，重寫compute方法，使用ForkJoinPool的submit提交任務，任務在某個執行緒中執行，工作任務中的compute方法的程式碼開始對任務進行分析，如果符合條件就進行任務拆分，拆分成多個子任務，每個子任務進行資料的計算或操作，得到結果返回給上一層任務開啟執行緒進行合併，最終通過get獲取整體處理結果。【只能將任務1個切分為兩個，不能切分為3個或其他數量】
- ForkJoinTask：代表fork/join裡面的任務型別，一般用它的兩個子類RecursiveTask（任務有返回值）和RecursiveAction（任務沒有返回值），任務的處理邏輯包括任務的切分都是在重寫compute方法裡面進行處理。只有ForkJoinTask任務可以被拆分執行和合並執行。【可檢視上篇Future原始碼分析的類圖結構】【ForkJoinTask使用了模板模式進行設計，將ForkJoinTask的執行相關程式碼進行隱藏，通過提供抽象類（即子類RecursiveTask、RecursiveAction）暴露使用者的實際業務處理。】
- RecursiveTask：在進行exec之後會使用一個result的變數進行接受返回的結果；
public abstract class RecursiveTask<V> extends ForkJoinTask<V> { V result; protected abstract V compute(); public final V getRawResult() { return result; } protected final void setRawResult(V value) { result = value; } protected final boolean exec() { result = compute(); return true; } } - RecursiveAction：在進行exec之後沒有返回結果；
public abstract class RecursiveAction extends ForkJoinTask<Void> { protected abstract void compute(); public final Void getRawResult() { return null; } protected final void setRawResult(Void mustBeNull) { } protected final boolean exec() { compute(); return true; } }
- RecursiveTask：在進行exec之後會使用一個result的變數進行接受返回的結果；
- ForkJoinPool：fork/join框架的管理者，最原始的任務都要交給它來處理。它負責控制整個fork/join有多少個工作執行緒，工作執行緒的建立、機會都是由它來控制。它還負責workQueue佇列的建立和分配，每當建立一個工作執行緒，它負責分配對應的workQueue，然後它把接到的活都交給工作執行緒去處理。是整個fork/join的容器。
- ForkJoinPool.WorkQueue：雙端佇列，負責儲存接收的任務；
- ForkJoinWorkerThread：fork/join裡面真正幹活的”工人“，它繼承了Thread，所以本質是一個執行緒。它有一個ForkJoinPool.WorkQueue的佇列存放著它要乾的活，接活之前它要向ForkJoinPool註冊（registerWorker），拿到相應的workQueue，然後就從workQueue裡面拿任務出來處理。它是依附於ForkJoinPool而存活，如果ForkJoinPool銷燬了，它也會跟著結束。【每一個ForkJoinWorkerThread執行緒都具有一個獨立的任務等待佇列workQueue。】
- 當使用ForkJoinPool進行submit任務提交時，建立1個workQueue將任務放進去，然後進行fork任務切分，如果切分後的任務放的進去之前的workQueue就放進去，不行就隨機選取workQueue放進去，如果還放不了就建立一個新的workQueue放進去；
public class ForkJoinWorkerThread extends Thread {
final ForkJoinPool pool;
final ForkJoinPool.WorkQueue workQueue;
protected ForkJoinWorkerThread(ForkJoinPool pool) {
super("aForkJoinWorkerThread");
this.pool = pool;
this.workQueue = pool.registerWorker(this);
}
}
二.用法
以前1+2+3+...+100這樣的處理可以用for迴圈處理，現在使用fork/join來處理：從下面結果可以看到，大任務被不斷的拆分成小任務，然後新增到工作執行緒的佇列中，每個小任務都會被工作執行緒從佇列中取出進行執行，然後每個小任務的結果的合併也由工作執行緒執行，然後不斷的彙總成最終結果。【task通過ForkJoinPool來執行，分割的子任務新增到當前工作執行緒的佇列中，進入佇列的頭部，當一個工作執行緒中沒有任務時，會從其他工作執行緒的佇列尾部獲取一個任務。（工作竊取：當前工作執行緒對應的佇列中沒有任務了，從其他工作執行緒對應的佇列中取出任務進行操作，然後將操作結果返還給對應佇列的執行緒。）】
public class MyFrokJoinTask extends RecursiveTask<Integer> {
private int begin;
private int end;
public MyFrokJoinTask(int begin, int end) {
this.begin = begin;
this.end = end;
}
public static void main(String[] args) throws Exception {
ForkJoinPool pool = new ForkJoinPool();
ForkJoinTask<Integer> result = pool.submit(new MyFrokJoinTask(1, 100));//提交任務
System.out.println("計算的值："+result.get());//得到最終的結果
}
@Override
protected Integer compute() {
int sum = 0;
if (end - begin <= 2) {
for (int i = begin; i <= end; i++) {
sum += i;
System.out.println("i:"+i);
}
} else {
MyFrokJoinTask d1 = new MyFrokJoinTask(begin, (begin + end) / 2);
MyFrokJoinTask d2 = new MyFrokJoinTask((begin + end) / 2+1, end);
d1.fork();//任務拆分
d2.fork();//任務拆分
Integer a = d1.join();//每個任務的結果
Integer b = d2.join();//每個任務的結果
sum = a + b;//彙總任務結果
System.out.println("sum:" + sum + ",a:" + a + ",b:" + b);
}
System.out.println("name:"+Thread.currentThread().getName());
return sum;
}
}
//=========結果============
i:1
i:2
name:ForkJoinPool-1-worker-1
i:3
i:4
name:ForkJoinPool-1-worker-1
sum:10,a:3,b:7
name:ForkJoinPool-1-worker-1
i:5
i:6
i:7
name:ForkJoinPool-1-worker-1
sum:28,a:10,b:18
name:ForkJoinPool-1-worker-1
...............
...............
sum:91,a:28,b:63
sum:99,a:45,b:54
name:ForkJoinPool-1-worker-3
name:ForkJoinPool-1-worker-1
i:23
i:24
i:25
name:ForkJoinPool-1-worker-2
sum:135,a:63,b:72
name:ForkJoinPool-1-worker-2
sum:234,a:99,b:135
name:ForkJoinPool-1-worker-3
sum:325,a:91,b:234
name:ForkJoinPool-1-worker-1
sum:1275,a:325,b:950
name:ForkJoinPool-1-worker-1
sum:5050,a:1275,b:3775
name:ForkJoinPool-1-worker-1
計算的值：5050
三.分析
ForkJoinPool
ForkJoinPool forkJoinPool = new ForkJoinPool();
//Runtime.getRuntime().availableProcessors()當前作業系統可以使用的CPU核心數量
public ForkJoinPool() {
this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
defaultForkJoinWorkerThreadFactory, null, false);
}
//this呼叫到下面這段程式碼
public ForkJoinPool(int parallelism,
ForkJoinWorkerThreadFactory factory,
UncaughtExceptionHandler handler,
boolean asyncMode) {
this(checkParallelism(parallelism), //並行度
checkFactory(factory), //工作執行緒建立工廠
handler, //異常處理handler
asyncMode ? FIFO_QUEUE : LIFO_QUEUE, //任務隊列出隊模式 非同步：先進先出，同步：後進先出
"ForkJoinPool-" + nextPoolId() + "-worker-");
checkPermission();
}
//上面的this最終呼叫到下面這段程式碼
private ForkJoinPool(int parallelism,
ForkJoinWorkerThreadFactory factory,
UncaughtExceptionHandler handler,
int mode,
String workerNamePrefix) {
this.workerNamePrefix = workerNamePrefix;
this.factory = factory;
this.ueh = handler;
this.config = (parallelism & SMASK) | mode;
long np = (long)(-parallelism); // offset ctl counts
this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
}
- parallelism：可並行數量，fork/join框架將依據這個並行數量的設定，決定框架內並行執行的執行緒數量。並行的每一個任務都會有一個執行緒進行處理；
- factory：當fork/join建立一個新的執行緒時，同樣會用到執行緒建立工廠。它實現了ForkJoinWorkerThreadFactory介面，使用預設的的介面實現類DefaultForkJoinWorkerThreadFactory來實現newThread方法建立一個新的工作執行緒；
public static interface ForkJoinWorkerThreadFactory { /** * Returns a new worker thread operating in the given pool. */ public ForkJoinWorkerThread newThread(ForkJoinPool pool); } static final class DefaultForkJoinWorkerThreadFactory implements ForkJoinWorkerThreadFactory { public final ForkJoinWorkerThread newThread(ForkJoinPool pool) { return new ForkJoinWorkerThread(pool); } } - handler：異常捕獲處理器。當執行的任務出現異常，並從任務中被丟擲時，就會被handler捕獲；
- asyncMode：fork/join為每一個獨立的工作執行緒準備了對應的待執行任務佇列，這個任務佇列是使用陣列進行組合的雙向佇列。即可以使用先進先出的工作模式，也可以使用後進先出的工作模式；
Fork()和Join()
fork/join框架中提供的fork()和join()是最重要的兩個方法，它們和parallelism(”可並行任務數量“)配合工作，可以導致拆分的子任務T1.1、T1.2甚至TX在fork/join中不同的執行效果(上面1+2....+100的每次執行的子任務都是不同的)。即TX子任務或等待其他已存在的執行緒執行關聯的子任務(sum操作)，或在執行TX的執行緒中”遞迴“執行其他任務(將1-50進行拆分後的子任務遞迴執行），或啟動一個新的執行緒執行子任務(執行1-50另一邊拆分的任務，即50-100的子任務）。
fork()用於將新建立的子任務放入當前執行緒的workQueue佇列中，fork/join框架將根據當前正在併發執行ForkJoinTask任務的ForkJoinWorkerThread執行緒狀態，決定是讓這個任務在佇列中等待，還是建立一個新的ForkJoinWorkedThread執行緒執行它，又或者是喚起其他正在等待任務的ForkJoinWorkerThread執行緒執行它。
join()用於讓當前執行緒阻塞，直到對應的子任務完成執行並返回執行結果。或者，如果這個子任務存在於當前執行緒的任務等待佇列workQueue中，則取出這個子任務進行”遞迴“執行，其目的是儘快得到當前子任務的執行結果，然後繼續執行。
提交任務：
-
sumbit的第一次提交：ForkJoinPool.submit(ForkJoinTask<T> task) -> externalPush(task) -> externalSubmit(task)
-
submit：
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; } - externalPush：將任務新增到隨機選取的佇列中或新建立的佇列中；
final void externalPush(ForkJoinTask<?> task) { WorkQueue[] ws; WorkQueue q; int m; int r = ThreadLocalRandom.getProbe();//當前執行緒的一個隨機數 int rs = runState;//當前容器的狀態 //如果隨機選取的佇列還有空位置可以存放、佇列加鎖鎖定成功，任務就放入佇列中 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 && (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);//挪動下次任務存放的槽的位置 U.putIntVolatile(q, QLOCK, 0);//佇列解鎖 if (n <= 1)//當前陣列元素少時，進行喚醒當前執行緒；或者當沒有活動執行緒或執行緒數較少時，新增新的執行緒 signalWork(ws, q); return; } U.compareAndSwapInt(q, QLOCK, 1, 0);//佇列解鎖 } externalSubmit(task);//升級版的externalPush } volatile int runState; // lockable status鎖定狀態 // runState: SHUTDOWN為負數，其他的為2的次冪 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;//關閉 - externalSubmit：佇列新增任務失敗，進行升級版操作，即建立佇列陣列和建立佇列後，將任務放入新建立的佇列中；
private void externalSubmit(ForkJoinTask<?> task) { 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; boolean move = false; /** *ForkJoinPool執行器停止工作了，丟擲異常 *ForkJoinPool extends AbstractExecutorService *abstract class AbstractExecutorService implements ExecutorService *interface ExecutorService extends Executor *interface Executor執行提交的物件Runnable任務 */ if ((rs = runState) < 0) { tryTerminate(false, false); // help terminate throw 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) { 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，config是CPU核數 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); } } //第三次遍歷，把任務放入佇列中 else if ((q = ws[k = r & m & SQMASK]) != null) { 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 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); } }
-
-
fork任務切分的提交：ForkJoinTask.fork() -> ForkJoinWorkerThread.workQueue.push(task)/ForkJoinPool.common.externalPush(task) -> ForkJoinPool.push(task)/externalPush(task)
- fork：
public final ForkJoinTask<V> fork() { Thread t; if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)//當前執行緒是workerThread，任務直接放入workerThread當前的workQueue ((ForkJoinWorkerThread)t).workQueue.push(this); else ForkJoinPool.common.externalPush(this);//將任務新增到隨機選取的佇列中或新建立的佇列中 return this; } -
push：
public class ForkJoinPool extends AbstractExecutorService { static final class WorkQueue { 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)//陣列所有元素都滿了進行2倍擴容 growArray(); } } final ForkJoinTask<?>[] growArray() { ForkJoinTask<?>[] oldA = array; int size = oldA != null ? oldA.length << 1 : INITIAL_QUEUE_CAPACITY;//2倍擴容或初始化 if (size > MAXIMUM_QUEUE_CAPACITY) throw new RejectedExecutionException("Queue capacity exceeded"); int oldMask, t, b; ForkJoinTask<?>[] a = array = new ForkJoinTask<?>[size]; if (oldA != null && (oldMask = oldA.length - 1) >= 0 && (t = top) - (b = base) > 0) { int mask = size - 1; do { // emulate poll from old array, push to new array遍歷從舊陣列中取出放到新陣列中 ForkJoinTask<?> x; int oldj = ((b & oldMask) << ASHIFT) + ABASE; int j = ((b & mask) << ASHIFT) + ABASE; x = (ForkJoinTask<?>)U.getObjectVolatile(oldA, oldj);//從舊陣列中取出 if (x != null && U.compareAndSwapObject(oldA, oldj, x, null))//將舊陣列取出的位置的物件置為null U.putObjectVolatile(a, j, x);//放入新陣列 } while (++b != t); } return a; } } }
- fork：
任務的消費
任務的消費的執行鏈路是ForkJoinTask.doExec() -> RecursiveTask.exec()/RecursiveAction.exec() -> 覆蓋重寫的compute()
-
doExec：任務的執行入口
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);//任務執行完設定狀態為NORMAL，並喚醒其他等待任務 } return s; } protected abstract boolean exec(); private int setCompletion(int completion) { for (int s;;) { if ((s = status) < 0) return s; if (U.compareAndSwapInt(this, STATUS, s, s | completion)) {//任務狀態修改為NORMAL if ((s >>> 16) != 0)//狀態不是SMASK synchronized (this) { notifyAll(); }//喚醒其他等待任務 return completion; } } } /** The run status of this task 任務的執行狀態*/ volatile int status; // accessed directly by pool and workers由ForkJoinPool池或ForkJoinWorkerThread控制 static final int DONE_MASK = 0xf0000000; // mask out non-completion bits static final int NORMAL = 0xf0000000; // must be negative static final int CANCELLED = 0xc0000000; // must be < NORMAL static final int EXCEPTIONAL = 0x80000000; // must be < CANCELLED static final int SIGNAL = 0x00010000; // must be >= 1 << 16 static final int SMASK = 0x0000ffff; // short bits for tags
任務真正執行處理邏輯
任務提交到ForkJoinPool，最終真正的是由繼承Thread的ForkJoinWorkerThread的run方法來執行消費任務的，ForkJoinWorkerThread處理哪個任務是由join來出隊的；
-
ForkJoinTask.join()
public final V join() { int s; if ((s = doJoin() & DONE_MASK) != NORMAL) reportException(s); return getRawResult();//得到返回結果 } private int doJoin() { int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w; /** * (s = status) < 0 判斷任務是否已經完成，完成直接返回s * 任務未完成： * 1）執行緒是ForkJoinWorkerThread，tryUnpush任務出隊然後消費任務doExec * 1.1）出隊或消費失敗，執行awaitJoin進行自旋，如果任務狀態是完成就退出，否則繼續嘗試出隊，直到任務完成或超時為止； * 2）如果執行緒不是ForkJoinWorkerThread，執行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(); } private void reportException(int s) { if (s == CANCELLED)//取消 throw new CancellationException(); if (s == EXCEPTIONAL)//異常 rethrow(getThrowableException()); } - awaitJoin：
public class ForkJoinPool{ 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; } } - externalAwaitDone：
private int externalAwaitDone() { /** * 當前任務是CountedCompleter * 1）是則執行ForkJoinPool.common.externalHelpComplete() * 2）否則執行ForkJoinPool.common.tryExternalUnpush(this)進行任務出隊 * 2.1）出隊成功，進行doExec()消費，否則進行阻塞等待 */ 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)) {//任務狀態標記為SIGNAL synchronized (this) { if (status >= 0) { try { wait(0L);//阻塞等待 } catch (InterruptedException ie) {//有中斷異常 interrupted = true;//設定中斷標識為true } } else notifyAll();//任務完成喚醒其他任務 } } } while ((s = status) >= 0); if (interrupted) Thread.currentThread().interrupt();//當前執行緒進行中斷 } return s; } final int externalHelpComplete(CountedCompleter<?> task, int maxTasks) { WorkQueue[] ws; int n; int r = ThreadLocalRandom.getProbe(); //沒有任務直接結束，有任務則執行helpComplete //helpComplete：執行隨機選取的佇列的任務，若選取的佇列中沒有任務了，則竊取其他佇列的任務並執行 return ((ws = workQueues) == null || (n = ws.length) == 0) ? 0 : helpComplete(ws[(n - 1) & r & SQMASK], task, maxTasks); } -
run和工作竊取
任務是由workThread來竊取的，workThread是一個執行緒。執行緒的所有邏輯都是由run()方法執行：
public class ForkJoinWorkerThread extends Thread {
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);
}
}
}
}
}
public class ForkJoinPool{
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)//從佇列中竊取任務成功，scan()進行任務竊取
w.runTask(t);//執行任務，內部方法呼叫了doExec()進行任務的消費
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; if ((q = ws[k]) != null) { //隨機選中了非空佇列 q if ((n = (b = q.base) - q.top) < 0 && (a = q.array) != null) { // non-empty long i = (((a.length - 1) & b) << ASHIFT) + ABASE; //從尾部出隊,b是尾部下標 if ((t = ((ForkJoinTask<?>) U.getObjectVolatile(a, i))) != null && q.base == b) { if (ss >= 0) { if (U.compareAndSwapObject(a, i, t, null)) { //利用cas出隊 q.base = b + 1; if (n < -1) // signal others signalWork(ws, q); return t; //出隊成功，成功竊取一個任務！ } } 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; }<br data-filtered="filtered"> //k = k + 1表示取下一個佇列 如果（k + 1） & m == origin表示 已經遍歷完所有隊列了 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; } - ForkJoinPool.runTask：
final void runTask(ForkJoinTask<?> task) { if (task != null) { 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(); } }
- scan：
四.總結
對於fork/join來說，在使用時還是存在下面的一些問題的：
- 在使用JVM的時候我們要考慮OOM的問題，如果我們的任務處理時間非常耗時，並且處理的資料非常大的時候，會造成OOM；
- ForkJoin是通過多執行緒的方式進行處理任務，那麼我們不得不考慮是否應該使用ForkJoin。因為當資料量不是特別大的時候，我們沒有必要使用ForkJoin。因為多執行緒會涉及到上下文的切換，所以資料量不大的時候使用序列比使用多執行緒快；
- 專案中進行本地測試發現，業務層Service進行excel表資料(資料量幾百)的複雜處理，進行單執行緒for迴圈統計消耗時間，然後與使用fork/join進行處理統計消耗時間，發現fork/join的消耗時間是單執行緒for的2倍；