Java線程池執(zhí)行原理是什么

本篇內(nèi)容介紹了“Java線程池執(zhí)行原理是什么”的有關(guān)知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！

線程池狀態(tài)

首先認(rèn)識兩個貫穿線程池代碼的參數(shù)：

runState：線程池運行狀態(tài)  workerCount：工作線程的數(shù)量

線程池用一個32位的int來同時保存runState和workerCount，其中高3位是runState，其余29位是workerCount。代碼中會反復(fù)使用runStateOf和workerCountOf來獲取runState和workerCount。

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));private static final int COUNT_BITS = Integer.SIZE - 3;private static final int CAPACITY = (1 << COUNT_BITS) - 1;// 線程池狀態(tài)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;// ctl操作private static int runStateOf(int c) { return c & ~CAPACITY; }private static int workerCountOf(int c) { return c & CAPACITY; }private static int ctlOf(int rs, int wc) { return rs | wc; }

RUNNING：可接收新任務(wù)，可執(zhí)行等待隊列里的任務(wù)  SHUTDOWN：不可接收新任務(wù)，可執(zhí)行等待隊列里的任務(wù)  STOP：不可接收新任務(wù)，不可執(zhí)行等待隊列里的任務(wù)，并且嘗試終止所有在運行任務(wù)  TIDYING：所有任務(wù)已經(jīng)終止，執(zhí)行terminated()  TERMINATED：terminated()執(zhí)行完成

線程池狀態(tài)默認(rèn)從RUNNING開始流轉(zhuǎn)，到狀態(tài)TERMINATED結(jié)束，中間不需要經(jīng)過每一種狀態(tài)，但不能讓狀態(tài)回退。下面是狀態(tài)變化可能的路徑和變化條件：

Worker的創(chuàng)建

線程池是由Worker類負(fù)責(zé)執(zhí)行任務(wù)，Worker繼承了AbstractQueuedSynchronizer，引出了Java并發(fā)框架的核心AQS。

AbstractQueuedSynchronizer，簡稱AQS，是Java并發(fā)包里一系列同步工具的基礎(chǔ)實現(xiàn)，原理是根據(jù)狀態(tài)位來控制線程的入隊阻塞、出隊喚醒來處理同步。

AQS不會在這里展開討論，只需要知道Worker包裝了Thread，由它去執(zhí)行任務(wù)。

調(diào)用execute將會根據(jù)線程池的情況創(chuàng)建Worker，可以歸納出下圖四種情況：

public void execute(Runnable command) {if (command == null)throw new NullPointerException();int c = ctl.get();//1if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}//2if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();if (! isRunning(recheck) && remove(command))//3reject(command);else if (workerCountOf(recheck) == 0)//4addWorker(null, false);}//5else if (!addWorker(command, false))//6reject(command);}

標(biāo)記1對應(yīng)第一種情況，要留意addWorker傳入了core，core=true為corePoolSize，core=false為maximumPoolSize，

新增時需要檢查workerCount是否超過允許的最大值。

標(biāo)記2對應(yīng)第二種情況，檢查線程池是否在運行，并且將任務(wù)加入等待隊列。標(biāo)記3再檢查一次線程池狀態(tài)，如果線程池忽然處于非運行狀態(tài)，那就將等待隊列剛加的任務(wù)刪掉，再交給RejectedExecutionHandler處理。標(biāo)記4發(fā)現(xiàn)沒有worker，就先補充一個空任務(wù)的worker。

標(biāo)記5對應(yīng)第三種情況，等待隊列不能再添加任務(wù)了，調(diào)用addWorker添加一個去處理。

標(biāo)記6對應(yīng)第四種情況，addWorker的core傳入false，返回調(diào)用失敗，代表workerCount已經(jīng)超出maximumPoolSize，那就交給RejectedExecutionHandler處理。

private boolean addWorker(Runnable firstTask, boolean core) {//1retry:for (;;) {int c = ctl.get();int rs = runStateOf(c);// Check if queue empty only if necessary.if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {int wc = workerCountOf(c);if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get(); // Re-read ctlif (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop}}//2boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {w = new Worker(firstTask);final Thread t = w.thread;if (t != null) {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 rs = runStateOf(ctl.get());if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();workers.add(w);int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}if (workerAdded) {t.start();workerStarted = true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted;}

標(biāo)記1的第一段代碼，目的很簡單，是為workerCount加一。至于為什么代碼寫了這么長，是因為線程池的狀態(tài)在不斷

變化，并發(fā)環(huán)境下需要保證變量的同步性。外循環(huán)判斷線程池狀態(tài)、任務(wù)非空和隊列非空，內(nèi)循環(huán)使用CAS機制保證workerCount正確地遞增。不了解CAS可以看認(rèn)識非阻塞的同步機制CAS，后續(xù)增減workerCount都會使用CAS。

標(biāo)記2的第二段代碼，就比較簡單。創(chuàng)建一個新Worker對象，將Worker添加進workers里（Set集合）。成功添加后，啟動worker里的線程。在finally里判斷線程是否啟動成功，不成功直接調(diào)用addWorkerFailed。

private void addWorkerFailed(Worker w) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {if (w != null)workers.remove(w);decrementWorkerCount();tryTerminate();} finally {mainLock.unlock();}}

addWorkerFailed將減少已經(jīng)遞增的workerCount，并且調(diào)用tryTerminate結(jié)束線程池。

Worker的執(zhí)行

Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;this.thread = getThreadFactory().newThread(this);}public void run() {runWorker(this);}

Worker在構(gòu)造函數(shù)里采用ThreadFactory創(chuàng)建Thread，在run方法里調(diào)用了runWorker，看來是真正執(zhí)行任務(wù)的地方。

final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;try {//1while (task != null || (task = getTask()) != null) {w.lock();//2if ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())wt.interrupt();try {//3beforeExecute(wt, task);Throwable thrown = null;try {task.run();} catch (RuntimeException x) {thrown = x; throw x;} catch (Error x) {thrown = x; throw x;} catch (Throwable x) {thrown = x; throw new Error(x);} finally {afterExecute(task, thrown);}} finally {task = null;//4w.completedTasks++;w.unlock();}}completedAbruptly = false; //5} finally {//6processWorkerExit(w, completedAbruptly);}}

標(biāo)記1進入循環(huán)，從getTask獲取要執(zhí)行的任務(wù)，直到返回null。這里達(dá)到了線程復(fù)用的效果，讓線程處理多個任務(wù)。

標(biāo)記2是一個比較復(fù)雜的判斷，保證了線程池在STOP狀態(tài)下線程是中斷的，非STOP狀態(tài)下線程沒有被中斷。如果你不了解Java的中斷機制，看如何正確結(jié)束Java線程這篇。

標(biāo)記3調(diào)用了run方法，真正執(zhí)行了任務(wù)。執(zhí)行前后提供了beforeExecute和afterExecute兩個方法，由子類實現(xiàn)。

標(biāo)記4里的completedTasks統(tǒng)計worker執(zhí)行了多少任務(wù)，最后累加進completedTaskCount變量，可以調(diào)用相應(yīng)方法返回一些統(tǒng)計信息。

標(biāo)記5的變量completedAbruptly表示worker是否異常終止，執(zhí)行到這里代表執(zhí)行正常，后續(xù)的方法需要這個變量。

標(biāo)記6調(diào)用processWorkerExit結(jié)束，后面會分析。

接著來看worker從等待隊列獲取任務(wù)的getTask方法：

private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out?for (;;) {int c = ctl.get();int rs = runStateOf(c);//1// Check if queue empty only if necessary.if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {decrementWorkerCount();return null;}int wc = workerCountOf(c);//2// Are workers subject to culling?boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))return null;continue;}//3try {Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;timedOut = true;} catch (InterruptedException retry) {timedOut = false;}}}

標(biāo)記1檢查線程池的狀態(tài)，這里就體現(xiàn)出SHUTDOWN和STOP的區(qū)別。如果線程池是SHUTDOWN狀態(tài)，還會先處理完等待隊列的任務(wù)；如果是STOP狀態(tài)，就不再處理等待隊列里的任務(wù)了。

標(biāo)記2先看allowCoreThreadTimeOut這個變量，false時worker空閑，也不會結(jié)束；true時，如果worker空閑超過keepAliveTime，就會結(jié)束。接著是一個很復(fù)雜的判斷，好難轉(zhuǎn)成文字描述，自己看吧。注意一下wc>maximumPoolSize，出現(xiàn)這種可能是在運行中調(diào)用setMaximumPoolSize，還有wc>1，在等待隊列非空時，至少保留一個worker。

標(biāo)記3是從等待隊列取任務(wù)的邏輯，根據(jù)timed分為等待keepAliveTime或者阻塞直到有任務(wù)。

最后來看結(jié)束worker需要執(zhí)行的操作：

private void processWorkerExit(Worker w, boolean completedAbruptly) {//1if (completedAbruptly) // If abrupt, then workerCount wasn't adjusteddecrementWorkerCount();//2final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {completedTaskCount += w.completedTasks;workers.remove(w);} finally {mainLock.unlock();}//3tryTerminate();int c = ctl.get();//4if (runStateLessThan(c, STOP)) {if (!completedAbruptly) {int min = allowCoreThreadTimeOut ? 0 : corePoolSize;if (min == 0 && ! workQueue.isEmpty())min = 1;if (workerCountOf(c) >= min)return; // replacement not needed}addWorker(null, false);}}

正常情況下，在getTask里就會將workerCount減一。標(biāo)記1處用變量completedAbruptly判斷worker是否異常退出，如果是，需要補充對workerCount的減一。

標(biāo)記2將worker處理任務(wù)的數(shù)量累加到總數(shù)，并且在集合workers中去除。

標(biāo)記3嘗試終止線程池，后續(xù)會研究。

標(biāo)記4處理線程池還是RUNNING或SHUTDOWN狀態(tài)時，如果worker是異常結(jié)束，那么會直接addWorker。如果allowCoreThreadTimeOut=true，并且等待隊列有任務(wù)，至少保留一個worker；如果allowCoreThreadTimeOut=false，workerCount不少于corePoolSize。

總結(jié)一下worker：線程池啟動后，worker在池內(nèi)創(chuàng)建，包裝了提交的Runnable任務(wù)并執(zhí)行，執(zhí)行完就等待下一個任務(wù)，不再需要時就結(jié)束。

線程池的關(guān)閉

線程池的關(guān)閉不是一關(guān)了事，worker在池里處于不同狀態(tài)，必須安排好worker的”后事”，才能真正釋放線程池。ThreadPoolExecutor提供兩種方法關(guān)閉線程池：

shutdown：不能再提交任務(wù)，已經(jīng)提交的任務(wù)可繼續(xù)運行；  shutdownNow：不能再提交任務(wù)，已經(jīng)提交但未執(zhí)行的任務(wù)不能運行，在運行的任務(wù)可繼續(xù)運行，但會被中斷，返回已經(jīng)提交但未執(zhí)行的任務(wù)。

public void shutdown() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess(); //1 安全策略機制advanceRunState(SHUTDOWN); //2interruptIdleWorkers(); //3onShutdown(); //4 空方法，子類實現(xiàn)} finally {mainLock.unlock();}tryTerminate(); //5}

shutdown將線程池切換到SHUTDOWN狀態(tài)，并調(diào)用interruptIdleWorkers請求中斷所有空閑的worker，最后調(diào)用tryTerminate嘗試結(jié)束線程池。

public List<Runnable> shutdownNow() {List<Runnable> tasks;final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();advanceRunState(STOP);interruptWorkers();tasks = drainQueue(); //1} finally {mainLock.unlock();}tryTerminate();return tasks;}

shutdownNow和shutdown類似，將線程池切換為STOP狀態(tài)，中斷目標(biāo)是所有worker。drainQueue會將等待隊列里未執(zhí)行的任務(wù)返回。

interruptIdleWorkers和interruptWorkers實現(xiàn)原理都是遍歷workers集合，中斷條件符合的worker。

上面的代碼多次出現(xiàn)調(diào)用tryTerminate，這是一個嘗試將線程池切換到TERMINATED狀態(tài)的方法。

final void tryTerminate() {for (;;) {int c = ctl.get();//1if (isRunning(c) ||runStateAtLeast(c, TIDYING) ||(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))return;//2if (workerCountOf(c) != 0) { // Eligible to terminateinterruptIdleWorkers(ONLY_ONE);return;}//3final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {try {terminated();} finally {ctl.set(ctlOf(TERMINATED, 0));termination.signalAll();}return;}} finally {mainLock.unlock();}// else retry on failed CAS}}

標(biāo)記1檢查線程池狀態(tài)，下面幾種情況，后續(xù)操作都沒有必要，直接return。

RUNNING（還在運行，不能停）  TIDYING或TERMINATED（已經(jīng)沒有在運行的worker）  SHUTDOWN并且等待隊列非空（執(zhí)行完才能停）

標(biāo)記2在worker非空的情況下又調(diào)用了interruptIdleWorkers，你可能疑惑在shutdown時已經(jīng)調(diào)用過了，為什么又調(diào)用，而且每次只中斷一個空閑worker？

你需要知道，shutdown時worker可能在執(zhí)行中，執(zhí)行完阻塞在隊列的take，不知道要結(jié)束，所有要補充調(diào)用interruptIdleWorkers。每次只中斷一個是因為processWorkerExit時，還會執(zhí)行tryTerminate，自動中斷下一個空閑的worker。

標(biāo)記3是最終的狀態(tài)切換。線程池會先進入TIDYING狀態(tài)，再進入TERMINATED狀態(tài)，中間提供了terminated這個空方法供子類實現(xiàn)。

調(diào)用關(guān)閉線程池方法后，需要等待線程池切換到TERMINATED狀態(tài)。awaitTermination檢查限定時間內(nèi)線程池是否進入TERMINATED狀態(tài)，代碼如下：

public boolean awaitTermination(long timeout, TimeUnit unit)throws InterruptedException {long nanos = unit.toNanos(timeout);final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {for (;;) {if (runStateAtLeast(ctl.get(), TERMINATED))return true;if (nanos <= 0)return false;nanos = termination.awaitNanos(nanos);}} finally {mainLock.unlock();}}

“Java線程池執(zhí)行原理是什么”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識可以關(guān)注億速云網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實用文章！