前言

　　之前研究了一下如何使用ScheduledThreadPoolExecutor動態(tài)創(chuàng)建定時任務(wù)(Springboot定時任務(wù)原理及如何動態(tài)創(chuàng)建定時任務(wù))，簡單了解了ScheduledThreadPoolExecutor相關(guān)源碼。今天看了同學(xué)寫的ThreadPoolExecutor 的源碼解讀，甚是NB，必須轉(zhuǎn)發(fā)一下。

讀了一下 ThreadPoolExecutor 的源碼(JDK 11), 簡單的做個筆記.

Executor 框架

Executor

Executor?接口只有一個方法:

public interface Executor {void execute(Runnable command); }

Executor?接口提供了一種將任務(wù)提交和任務(wù)執(zhí)行機制解耦的方法.?Executor?的實現(xiàn)并不須要是異步的.

ExecutorService

ExecutorService?在?Executor?的基礎(chǔ)上, 提供了一些管理終止的方法和可以生成?Future?來跟蹤一個或多個異步任務(wù)的進(jìn)度的方法:

• shutdown()?方法會啟動比較柔和的關(guān)閉過程, 并且不會阻塞.?ExecutorService?將會繼續(xù)執(zhí)行已經(jīng)提交的任務(wù), 但不會再接受新的任務(wù). 如果?ExecutorService?已經(jīng)被關(guān)閉, 則不會有附加的操作.
• shutdownNow()?方法會嘗試停止正在執(zhí)行的任務(wù), 不再執(zhí)行等待執(zhí)行的任務(wù), 并且返回等待執(zhí)行的任務(wù)列表, 不會阻塞. 這個方法只能嘗試停止任務(wù), 典型的取消實現(xiàn)是通過中斷來取消任務(wù), 因此不能響應(yīng)中斷的任務(wù)可能永遠(yuǎn)不會終止.
• invokeAll()?方法執(zhí)行給定集合中的所有任務(wù), 當(dāng)所有任務(wù)完成時返回?Future?的列表, 支持中斷. 如果在此操作正在進(jìn)行時修改了給定的集合，則此方法的結(jié)果未定義.
• invokeAny()?方法會執(zhí)行給定集合中的任務(wù), 當(dāng)有一個任務(wù)完成時, 返回這個任務(wù)的結(jié)果, 并取消其他未完成的任務(wù), 支持中斷. 如果在此操作正在進(jìn)行時修改了給定的集合，則此方法的結(jié)果未定義.

AbstractExecutorService

AbstractExecutorService?提供了一些?ExecutorService?的執(zhí)行方法的默認(rèn)實現(xiàn). 這個方法使用了?newTaskFor()?方法返回的?RunnableFuture?(默認(rèn)是?FutureTask?) 來實現(xiàn)?submit()?、invokeAll()、?invokeAny()?方法.

RunnableFuture?繼承了?Runnable?和?Future?, 在?run()?方法成功執(zhí)行后, 將會設(shè)置完成狀態(tài), 并允許獲取執(zhí)行的結(jié)果:

public interface RunnableFuture<V> extends Runnable, Future<V> {/*** Sets this Future to the result of its computation* unless it has been cancelled.*/void run(); }

FutureTask

FutureTask?實現(xiàn)了?RunnableFuture?接口, 表示一個可取消的計算任務(wù), 只能在任務(wù)完成之后獲取結(jié)果, 并且在任務(wù)完成后, 就不再能取消或重啟, 除非使用?runAndReset()?方法.

FutureTask?有 7 個狀態(tài):

• NEW
• COMPLETING
• NORMAL
• EXCEPTIONAL
• CANCELLED
• INTERRUPTING
• INTERRUPTED

可能的狀態(tài)轉(zhuǎn)換:

• NEW -> COMPLETING -> NORMAL
• NEW -> COMPLETING -> EXCEPTIONAL
• NEW -> CANCELLED
• NEW -> INTERRUPTING -> INTERRUPTED

FutureTask?在更新 state 、 runner、 waiters 時, 都使用了?VarHandle.compareAndSet()?:

// VarHandle mechanics private static final VarHandle STATE; private static final VarHandle RUNNER; private static final VarHandle WAITERS; static {try {MethodHandles.Lookup l = MethodHandles.lookup();STATE = l.findVarHandle(FutureTask.class, "state", int.class);RUNNER = l.findVarHandle(FutureTask.class, "runner", Thread.class);WAITERS = l.findVarHandle(FutureTask.class, "waiters", WaitNode.class);} catch (ReflectiveOperationException e) {throw new ExceptionInInitializerError(e);}// Reduce the risk of rare disastrous classloading in first call to// LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773Class<?> ensureLoaded = LockSupport.class; }protected void set(V v) {if (STATE.compareAndSet(this, NEW, COMPLETING)) {outcome = v;STATE.setRelease(this, NORMAL); // final state finishCompletion();} }

來看一下?get()?方法:

public V get(long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException {if (unit == null)throw new NullPointerException();int s = state;if (s <= COMPLETING &&(s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)throw new TimeoutException();return report(s); }private int awaitDone(boolean timed, long nanos)throws InterruptedException {long startTime = 0L; WaitNode q = null;boolean queued = false;for (;;) {int s = state;if (s > COMPLETING) {// 已經(jīng)在終結(jié)狀態(tài), 返回狀態(tài)if (q != null)q.thread = null;return s;}else if (s == COMPLETING)// 已經(jīng)完成了, 但是狀態(tài)還是 COMPLETING Thread.yield();else if (Thread.interrupted()) {// 檢查中斷 removeWaiter(q);throw new InterruptedException();}else if (q == null) {// 沒有創(chuàng)建 WaitNode 節(jié)點, 如果 timed 并且 nanos 大于 0, 創(chuàng)建一個 WaitNodeif (timed && nanos <= 0L)return s;q = new WaitNode();}else if (!queued)// 將新的 WaitNode 放到鏈表頭部, 并嘗試 cas 到 waitersqueued = WAITERS.weakCompareAndSet(this, q.next = waiters, q);else if (timed) {final long parkNanos;if (startTime == 0L) { // first timestartTime = System.nanoTime();if (startTime == 0L)startTime = 1L;parkNanos = nanos;} else {long elapsed = System.nanoTime() - startTime;if (elapsed >= nanos) {// 超時了 removeWaiter(q);return state;}// park 的時間parkNanos = nanos - elapsed;}// nanos 比較慢, 再次檢查, 然后阻塞if (state < COMPLETING)LockSupport.parkNanos(this, parkNanos);}else// 不需要超時的阻塞LockSupport.park(this);} }

再來看下?run()?方法:

public void run() {if (state != NEW ||!RUNNER.compareAndSet(this, null, Thread.currentThread()))// 不在 NEW 狀態(tài), 或者 runner 不為 nullreturn;try {// callable 是在構(gòu)造器中指定的或用 Executors.callable(runnable, result) 創(chuàng)建的Callable<V> c = callable;if (c != null && state == NEW) {V result;boolean ran;try {result = c.call();ran = true;} catch (Throwable ex) {result = null;ran = false;// 設(shè)置異常狀態(tài)和異常結(jié)果 setException(ex);}if (ran)// 正常完成, 設(shè)置完成狀態(tài)和結(jié)果 set(result);}} finally {// runner must be non-null until state is settled to// prevent concurrent calls to run()runner = null;// state must be re-read after nulling runner to prevent// leaked interruptsint s = state;if (s >= INTERRUPTING)handlePossibleCancellationInterrupt(s);} }protected void set(V v) {if (STATE.compareAndSet(this, NEW, COMPLETING)) {outcome = v;STATE.setRelease(this, NORMAL); // final state finishCompletion();} }private void finishCompletion() {// assert state > COMPLETING;for (WaitNode q; (q = waiters) != null;) {if (WAITERS.weakCompareAndSet(this, q, null)) {// cas 移除 waiters, 對鏈表中的每個 Node 的線程 unparkfor (;;) {Thread t = q.thread;if (t != null) {q.thread = null;LockSupport.unpark(t);}WaitNode next = q.next;if (next == null)break;q.next = null; // unlink to help gcq = next;}break;}}// 默認(rèn)實現(xiàn)什么都沒做 done();callable = null; // to reduce footprint }

AbstractExecutorService 的執(zhí)行方法

來看下?AbstractExecutorService?實現(xiàn)的幾個執(zhí)行方法, 這里就只放上以?Callable?為參數(shù)的方法:

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {return new FutureTask<T>(callable); }public <T> Future<T> submit(Callable<T> task) {if (task == null) throw new NullPointerException();RunnableFuture<T> ftask = newTaskFor(task);execute(ftask);return ftask; }public <T> T invokeAny(Collection<? extends Callable<T>> tasks)throws InterruptedException, ExecutionException {try {return doInvokeAny(tasks, false, 0);} catch (TimeoutException cannotHappen) {assert false;return null;} }private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,boolean timed, long nanos)throws InterruptedException, ExecutionException, TimeoutException {if (tasks == null)throw new NullPointerException();int ntasks = tasks.size();if (ntasks == 0)throw new IllegalArgumentException();ArrayList<Future<T>> futures = new ArrayList<>(ntasks);ExecutorCompletionService<T> ecs =new ExecutorCompletionService<T>(this);try {ExecutionException ee = null;final long deadline = timed ? System.nanoTime() + nanos : 0L;Iterator<? extends Callable<T>> it = tasks.iterator();// 提交一個任務(wù)到 ecs futures.add(ecs.submit(it.next()));--ntasks;int active = 1;for (;;) {// 嘗試獲取第一個完成的任務(wù)的 FutureFuture<T> f = ecs.poll();if (f == null) {// 沒有完成的任務(wù)if (ntasks > 0) {// 還有沒提交的任務(wù), 再提交一個到 ecs--ntasks;futures.add(ecs.submit(it.next()));++active;}else if (active == 0)// 沒有還沒提交的任務(wù)和正在執(zhí)行的任務(wù)了break;else if (timed) {f = ecs.poll(nanos, NANOSECONDS);if (f == null)throw new TimeoutException();nanos = deadline - System.nanoTime();}elsef = ecs.take();}if (f != null) {// 存在已經(jīng)完成的任務(wù)--active;try {// 獲取結(jié)果并返回return f.get();} catch (ExecutionException eex) {ee = eex;} catch (RuntimeException rex) {ee = new ExecutionException(rex);}}}// 出錯, 拋出if (ee == null)ee = new ExecutionException();throw ee;} finally {// 取消所有已經(jīng)提交的任務(wù) cancelAll(futures);} }public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)throws InterruptedException {if (tasks == null)throw new NullPointerException();ArrayList<Future<T>> futures = new ArrayList<>(tasks.size());try {for (Callable<T> t : tasks) {// 提交任務(wù)RunnableFuture<T> f = newTaskFor(t);futures.add(f);execute(f);}for (int i = 0, size = futures.size(); i < size; i++) {Future<T> f = futures.get(i);if (!f.isDone()) {// 任務(wù)沒有完成, get() 等待任務(wù)完成try { f.get(); }catch (CancellationException | ExecutionException ignore) {}}}return futures;} catch (Throwable t) {cancelAll(futures);throw t;} }

構(gòu)造器

ThreadPoolExecutor?一共有4個構(gòu)造器, 這里就只放上兩個構(gòu)造器:

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), defaultHandler); }public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)throw new IllegalArgumentException();if (workQueue == null || threadFactory == null || handler == null)throw new NullPointerException();this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler; }

參數(shù)說明:

• corePoolSize: 在線程池中保持的線程的數(shù)量, 即使這些線程是空閑的, 除非?allowCoreThreadTimeOut?被設(shè)置為?true;
• maximumPoolSize: 線程池中最大線程數(shù)量;
• keepAliveTime: 多余空閑線程在終止之前等待新任務(wù)的最長時間;
• unit:?keepAliveTime?的時間單位;
• workQueue: 任務(wù)的等待隊列, 用于存放等待執(zhí)行的任務(wù). 僅包含?execute()?方法提交的?Runnable;
• threadFactory: executor 用來創(chuàng)建線程的工廠, 默認(rèn)使用?Executors.defaultThreadFactory()?來創(chuàng)建一個新的工廠;
• handler: 任務(wù)因為達(dá)到了線程邊界和隊列容量而被阻止時的處理程序, 默認(rèn)使用?AbortPolicy.

狀態(tài)

ThreadPoolExecutor?有5個狀態(tài):

• RUNNING: 接受新任務(wù), 并且處理隊列中的任務(wù);
• SHUTDOWN: 不接受新任務(wù), 但是處理隊列中的任務(wù), 此時仍然可能創(chuàng)建新的線程;
• STOP: 不接受新任務(wù), 處理隊列中的任務(wù), 中斷正在運行的任務(wù);
• TIDYING: 所有的任務(wù)都終結(jié)了, workCount 的值是0, 將狀態(tài)轉(zhuǎn)換為 TIDYING 的線程會執(zhí)行?terminated()?方法;
• TERMINATED:?terminated()?方法執(zhí)行完畢.

狀態(tài)轉(zhuǎn)換:

• RUNNING -> SHUTDOWN , On invocation of shutdown()
• (RUNNING or SHUTDOWN) -> STOP , On invocation of shutdownNow()
• SHUTDOWN -> TIDYING , When both queue and pool are empty
• STOP -> TIDYING , When pool is empty
• TIDYING -> TERMINATED , When the terminated() hook method has completed

workCount 和 state 被打包在一個?AtomicInteger?中, 其中的高三位用于表示線程池狀態(tài)( state ), 低 29 位用于表示 workCount:

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); private static final int COUNT_BITS = Integer.SIZE - 3; private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;// runState is stored in the high-order bits private static final int RUNNING = -1 << COUNT_BITS; private static final int SHUTDOWN = 0 << COUNT_BITS; private static final int STOP = 1 << COUNT_BITS; private static final int TIDYING = 2 << COUNT_BITS; private static final int TERMINATED = 3 << COUNT_BITS;// Packing and unpacking ctl private static int runStateOf(int c) { return c & ~COUNT_MASK; } private static int workerCountOf(int c) { return c & COUNT_MASK; } private static int ctlOf(int rs, int wc) { return rs | wc; }

workCount 表示有效的線程數(shù)量, 是允許啟動且不允許停止的 worker 的數(shù)量, 與實際的線程數(shù)量瞬時不同. 用戶可見的線程池大小是 Worker 集合的大小.

Worker 與任務(wù)調(diào)度

工作線程被封裝在?Worker?中 , 并且存放在一個?HashSet?(workers) 中由 mainLock 保護(hù):

/*** Set containing all worker threads in pool. Accessed only when* holding mainLock.*/ private final HashSet<Worker> workers = new HashSet<>();private final class Workerextends AbstractQueuedSynchronizerimplements Runnable{/*** This class will never be serialized, but we provide a* serialVersionUID to suppress a javac warning.*/private static final long serialVersionUID = 6138294804551838833L;final Thread thread;Runnable firstTask;volatile long completedTasks;Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;this.thread = getThreadFactory().newThread(this);}/** Delegates main run loop to outer runWorker. */public void run() {runWorker(this);}... }

Worker.run()方法很簡單, 直接調(diào)用了?runWorker()?方法, 來看一下這個方法的源碼:

final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;try {while (task != null || (task = getTask()) != null) {// task 不為 null 或 獲取到了需要執(zhí)行的任務(wù); getTask() 會阻塞, 并在線程需要退出時返回 null w.lock();// 檢查線程池狀態(tài)和線程的中斷狀態(tài), 如果被中斷, 代表線程池正在 STOPif ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())// 重新設(shè)置中斷狀態(tài) wt.interrupt();try {// 執(zhí)行前的鉤子 beforeExecute(wt, task);try {// 執(zhí)行任務(wù) task.run();// 執(zhí)行后的鉤子afterExecute(task, null);} catch (Throwable ex) {// 執(zhí)行后的鉤子 afterExecute(task, ex);throw ex;}} finally {// 更新狀態(tài), 準(zhǔn)備處理下一個任務(wù)task = null;w.completedTasks++;w.unlock();}}completedAbruptly = false;} finally {// 處理 Worker 的退出 processWorkerExit(w, completedAbruptly);} }

getTask()?方法會在以下4種情況返回 null :

• workCount 大于 maximumPoolSize;
• 線程池已經(jīng)處于 STOP 狀態(tài);
• 線程池已經(jīng)處于 SHUTDOWN 狀態(tài), 并且任務(wù)隊列為空;
• 等待任務(wù)時超時, 并且超時的 worker 需要被終止.

private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out?for (;;) {int c = ctl.get();if (runStateAtLeast(c, SHUTDOWN)&& (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {// 線程池已經(jīng)處于 SHUTDOWN 狀態(tài), 并且不在需要線程 (線程池已經(jīng)處于 STOP 狀態(tài) 或 workQueue 為空) decrementWorkerCount();return null;}int wc = workerCountOf(c);// 是否需要剔除超時的 workerboolean timed = allowCoreThreadTimeOut || wc > corePoolSize;if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {// 需要剔除當(dāng)前 worker, 嘗試調(diào)整 workerCountif (compareAndDecrementWorkerCount(c))// 成功 返回 nullreturn null;continue;}try {// 阻塞獲取任務(wù)Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;// 設(shè)置超時標(biāo)記, 下一次循環(huán)中檢查是否需要返回 nulltimedOut = true;} catch (InterruptedException retry) {// 被中斷, 設(shè)置超時標(biāo)記, 下一次循環(huán)中檢查是否需要返回 nulltimedOut = false;}} }

processWorkerExit()?方法負(fù)責(zé)垂死 worker 的清理和簿記, 只會被工作線程調(diào)用:

private void processWorkerExit(Worker w, boolean completedAbruptly) {if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted decrementWorkerCount();final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// 更新線程池完成的任務(wù)數(shù)量completedTaskCount += w.completedTasks;workers.remove(w);} finally {mainLock.unlock();}// 嘗試轉(zhuǎn)換線程池狀態(tài)到終止 tryTerminate();int c = ctl.get();if (runStateLessThan(c, STOP)) {if (!completedAbruptly) {// 不是由于用戶代碼異常而突然退出int min = allowCoreThreadTimeOut ? 0 : corePoolSize;if (min == 0 && ! workQueue.isEmpty())min = 1;if (workerCountOf(c) >= min)// 不需要在添加新 workerreturn;}// 嘗試添加新的 workeraddWorker(null, false);} }

提交任務(wù)

ThreadPoolExecutor?沒有重寫?submit()?方法, 我們只要看一下?execute()?就夠了:

public void execute(Runnable command) {if (command == null)throw new NullPointerException();int c = ctl.get();if (workerCountOf(c) < corePoolSize) {// 有效線程數(shù)量小于 corePoolSize 嘗試調(diào)用 addWorker 來增加一個線程(在 addWorker 方法中使用 corePoolSize 來檢查是否需要增加線程), 使用 corePoolSize 作為, 并把 command 作為新線程的第一個任務(wù)if (addWorker(command, true))return;// 調(diào)用失敗, 重新獲取狀態(tài)c = ctl.get();}if (isRunning(c) && workQueue.offer(command)) {// 線程池仍然在運行, 將 command 加入 workQueue 成功, 再次檢查狀態(tài), 因為此時線程池狀態(tài)可能已經(jīng)改變, 按照新的狀態(tài)拒絕 command 或嘗試添加新的線程int recheck = ctl.get();if (! isRunning(recheck) && remove(command))// 不再是運行中狀態(tài), 嘗試從隊列移除 command(還會嘗試將線程池狀態(tài)轉(zhuǎn)換為 TERMINATED), 拒絕command reject(command);else if (workerCountOf(recheck) == 0)// 有效線程數(shù)量為 0 , 創(chuàng)建新的線程, 在 addWorker 方法中使用 maximumPoolSize 來檢查是否需要增加線程addWorker(null, false);}else if (!addWorker(command, false))// 將任務(wù)放入隊列失敗或線程池不在運行狀態(tài), 并且嘗試添加線程失敗(此時線程池已經(jīng) shutdown 或飽和), 拒絕任務(wù) reject(command); }

addWorker()?方法有兩個參數(shù)?Runnable firstTask?和?boolean core?.?firstTask?是新建的工作線程的第一個任務(wù);?core?如果為 true , 表示用 corePoolSize 作為邊界條件, 否則表示用 maximumPoolSize. 這里的 core 用布爾值是為了確保檢查最新的狀態(tài).

addWorker()?主要做了這么兩件事情:

• 是否可以在當(dāng)前線程池狀態(tài)和給定的邊界條件(core or maximum)下創(chuàng)建一個新的工作線程;
• 如果可以, 調(diào)整 worker counter, 如果可能的話, 創(chuàng)建一個新的 worker 并啟動它, 把 firstTask 作為這個新 worker 的第一個任務(wù);

來看下?addWorker()?方法的源碼:

private boolean addWorker(Runnable firstTask, boolean core) {// 重試標(biāo)簽 retry:for (int c = ctl.get();;) {// 獲取最新的狀態(tài), 檢查狀態(tài)if (runStateAtLeast(c, SHUTDOWN)&& (runStateAtLeast(c, STOP)|| firstTask != null|| workQueue.isEmpty()))// 如果線程池狀態(tài)已經(jīng)進(jìn)入 SHUDOWN, 并且不再需要工作線程(已經(jīng)進(jìn)入 STOP 狀態(tài) 或 firstTask 不為 null 或 workQueue為空) 返回 falsereturn false;for (;;) {if (workerCountOf(c)>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))// 有效線程數(shù)量大于邊界條件, 返回 falsereturn false;if (compareAndIncrementWorkerCount(c))// 調(diào)整 workerCount, break retry, 退出外部循環(huán)break retry;c = ctl.get(); // Re-read ctlif (runStateAtLeast(c, SHUTDOWN))// 因為狀態(tài)變化導(dǎo)致 CAS 失敗, continue retry, 重試外部循環(huán)continue retry;// 由于 workerCount 改變導(dǎo)致 CAS 失敗, 重試內(nèi)嵌循環(huán) }}boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {// 新建 Workerw = new Worker(firstTask);final Thread t = w.thread;if (t != null) {// threadFactory 成功創(chuàng)建了線程final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.int c = ctl.get();// 重新檢查狀態(tài)if (isRunning(c) ||(runStateLessThan(c, STOP) && firstTask == null)) {// 線程池在 RUNNING 狀態(tài) 或 需要線程(線程池還不在 STOP 狀態(tài) 并且 firstTask 為 null)// 檢查線程是否可啟動if (t.isAlive()) throw new IllegalThreadStateException();// 將 worker 添加到 workers workers.add(w);// 更新 largestPoolSizeint s = workers.size();if (s > largestPoolSize)largestPoolSize = s;// 更新 worker 添加的標(biāo)記workerAdded = true;}} finally {mainLock.unlock();}if (workerAdded) {// 啟動線程, 更新啟動標(biāo)記 t.start();workerStarted = true;}}} finally {if (! workerStarted)// 失敗回滾 addWorkerFailed(w);}return workerStarted; }private void addWorkerFailed(Worker w) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// 從 workers 中移除 workerif (w != null)workers.remove(w);// 調(diào)整 workerCount() decrementWorkerCount();// 嘗試將線程池狀態(tài)改變?yōu)?TERMINATED tryTerminate();} finally {mainLock.unlock();} }

線程池關(guān)閉

來看一下線程池的關(guān)閉方法:

public void shutdown() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();// 如果線程池狀態(tài)還沒有達(dá)到SHUTDOWN, 將線程池狀態(tài)改為 SHUTDOWN advanceRunState(SHUTDOWN);// 中斷空閑的工作者線程 interruptIdleWorkers();// 鉤子onShutdown(); // hook for ScheduledThreadPoolExecutor} finally {mainLock.unlock();}// 嘗試轉(zhuǎn)換狀態(tài)到終止 tryTerminate(); }public List<Runnable> shutdownNow() {List<Runnable> tasks;final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();// 如果線程池狀態(tài)還沒有達(dá)到 STOP, 將線程池狀態(tài)改為 STOP advanceRunState(STOP);// 中斷所有 worker interruptWorkers();// 獲取任務(wù)隊列中的任務(wù), 并將這些任務(wù)從任務(wù)隊列中刪除tasks = drainQueue();} finally {mainLock.unlock();}// 嘗試轉(zhuǎn)換狀態(tài)到終止 tryTerminate();return tasks; }public boolean awaitTermination(long timeout, TimeUnit unit)throws InterruptedException {long nanos = unit.toNanos(timeout);final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// 等待線程池終止或超時while (runStateLessThan(ctl.get(), TERMINATED)) {if (nanos <= 0L)// 剩余時間小于 0 , 超時return false;nanos = termination.awaitNanos(nanos);}return true;} finally {mainLock.unlock();} }

tryTerminate()?方法中, 如果成功將線程池狀態(tài)轉(zhuǎn)換到了 TERMINATED, 將會termination.signalAll()?來喚醒等待線程池終結(jié)的線程:

final void tryTerminate() {for (;;) {int c = ctl.get();if (isRunning(c) ||runStateAtLeast(c, TIDYING) ||(runStateLessThan(c, STOP) && ! workQueue.isEmpty()))// 狀態(tài)不需要改變 (處于 RUNNING 狀態(tài) 或 已經(jīng)處于 TIDYING 狀態(tài) 或 (還沒到達(dá) STOP 狀態(tài), 并且 workQueue 不為空))return;if (workerCountOf(c) != 0) { // Eligible to terminate// 中斷一個空閑的 worker, 以傳播關(guān)閉狀態(tài)到工作線程 interruptIdleWorkers(ONLY_ONE);return;}final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {// 將狀態(tài)成功更新為 TIDYINGtry {// 默認(rèn)實現(xiàn)沒有做任何事情 terminated();} finally {// 將線程池狀態(tài)更新為 TERMINATEDctl.set(ctlOf(TERMINATED, 0));// 喚醒等待終結(jié)的線程 termination.signalAll();}return;}} finally {mainLock.unlock();}// else retry on failed CAS } }

　　原文出處：https://www.cnblogs.com/FJH1994/p/10362452.html

轉(zhuǎn)載于:https://www.cnblogs.com/hujunzheng/p/10364923.html

總結(jié)

以上是生活随笔為你收集整理的转载：ThreadPoolExecutor 源码阅读的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。