點擊上方?好好學java?，選擇?星標?公眾號

重磅資訊、干貨，第一時間送達 今日推薦：SQL 語法速成手冊文末點擊閱讀原文，去B站看視頻，別忘記關注哦 個人原創100W+訪問量博客：點擊前往，查看更多

來源：jianshu.com/p/704a6c5d337c

---

一、概述

線程池，顧名思義就是存放線程的池子，池子里存放了很多可以復用的線程。

如果不用類似線程池的容器，每當我們需要執行用戶任務的時候都去創建新的線程，任務執行完之后線程就被回收了，這樣頻繁地創建和銷毀線程會浪費大量的系統資源。

因此，線程池通過線程復用機制，并對線程進行統一管理，具有以下優點：

• 降低系統資源消耗。通過復用已存在的線程，降低線程創建和銷毀造成的消耗；

• 提高響應速度。當有任務到達時，無需等待新線程的創建便能立即執行；

• 提高線程的可管理性。線程是稀缺資源，如果無限制的創建，不僅會消耗大量系統資源，還會降低系統的穩定性，使用線程池可以進行對線程進行統一的分配、調優和監控。

ThreadPoolExecutor是線程池框架的一個核心類，本文通過對ThreadPoolExecutor源碼的分析（基于JDK 1.8），來深入分析線程池的實現原理。

二、ThreadPoolExecutor類的屬性

先從ThreadPoolExecutor類中的字段開始：

附上我歷時三個月總結的?Java 面試 + Java 后端技術學習指南，筆者這幾年及春招的總結，github 1.4k star，拿去不謝！

下載方式

1.?首先掃描下方二維碼

2.?后臺回復「Java面試」即可獲取

// 線程池的控制狀態,用高3位來表示線程池的運行狀態,低29位來表示線程池中工作線程的數量private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));//值為29,用來表示偏移量private static final int COUNT_BITS = Integer.SIZE - 3;//線程池的最大容量,其值的二進制為:00011111111111111111111111111111（29個1）private static final int CAPACITY = (1 << COUNT_BITS) - 1;// 線程池的運行狀態，總共有5個狀態，用高3位來表示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;//任務緩存隊列，用來存放等待執行的任務private final BlockingQueue<Runnable> workQueue;//全局鎖，對線程池狀態等屬性修改時需要使用這個鎖private final ReentrantLock mainLock = new ReentrantLock();//線程池中工作線程的集合，訪問和修改需要持有全局鎖private final HashSet<Worker> workers = new HashSet<Worker>();// 終止條件private final Condition termination = mainLock.newCondition();//線程池中曾經出現過的最大線程數private int largestPoolSize;//已完成任務的數量private long completedTaskCount;//線程工廠private volatile ThreadFactory threadFactory;//任務拒絕策略private volatile RejectedExecutionHandler handler;//線程存活時間private volatile long keepAliveTime;//是否允許核心線程超時private volatile boolean allowCoreThreadTimeOut;//核心池大小，若allowCoreThreadTimeOut被設置，核心線程全部空閑超時被回收的情況下會為0private volatile int corePoolSize;//最大池大小，不得超過CAPACITYprivate volatile int maximumPoolSize;//默認的任務拒絕策略private static final RejectedExecutionHandler defaultHandler =new AbortPolicy();private static final RuntimePermission shutdownPerm =new RuntimePermission("modifyThread");private final AccessControlContext acc;

在ThreadPoolExecutor類的這些屬性中，線程池狀態是控制線程池生命周期至關重要的屬性，這里就以線程池狀態為出發點進行研究。

通過上面的源碼可知，線程池的運行狀態總共有5種，其值和含義分別如下：

• RUNNING: ?高3位為111，接受新任務并處理阻塞隊列中的任務

• SHUTDOWN: 高3位為000，不接受新任務但會處理阻塞隊列中的任務

• STOP:高3位為001，不會接受新任務，也不會處理阻塞隊列中的任務，并且中斷正在運行的任務

• TIDYING: ?高3位為010，所有任務都已終止，工作線程數量為0，線程池將轉化到TIDYING狀態，即將要執行terminated()鉤子方法

• TERMINATED: 高3位為011，terminated()方法已經執行結束

然而，線程池中并沒有使用單獨的變量來表示線程池的運行狀態，而是使用一個AtomicInteger類型的變量ctl來表示線程池的控制狀態，其將線程池運行狀態與工作線程的數量打包在一個整型中，用高3位來表示線程池的運行狀態,低29位來表示線程池中工作線程的數量，對ctl的操作主要參考以下幾個函數：

// 通過與的方式，獲取ctl的高3位，也就是線程池的運行狀態private static int runStateOf(int c) { return c & ~CAPACITY; }//通過與的方式，獲取ctl的低29位，也就是線程池中工作線程的數量private static int workerCountOf(int c) { return c & CAPACITY; }//通過或的方式，將線程池狀態和線程池中工作線程的數量打包成ctlprivate static int ctlOf(int rs, int wc) { return rs | wc; }//SHUTDOWN狀態的值是0，比它大的均是線程池停止或清理狀態，比它小的是運行狀態private static boolean isRunning(int c) {return c < SHUTDOWN;}

接下來，我們看一下線程池狀態的所有轉換情況，如下：

• RUNNING -> SHUTDOWN：調用shutdown()，可能在finalize()中隱式調用

• (RUNNING or SHUTDOWN) -> STOP：調用shutdownNow()

• SHUTDOWN -> TIDYING：當緩存隊列和線程池都為空時

• STOP -> TIDYING：當線程池為空時

• TIDYING -> TERMINATED：當terminated()方法執行結束時

通常情況下，線程池有如下兩種狀態轉換流程：

RUNNING -> SHUTDOWN -> TIDYING -> TERMINATED

RUNNING -> STOP -> TIDYING -> TERMINATED

三、ThreadPoolExecutor類的構造方法

通常情況下，我們使用線程池的方式就是new一個ThreadPoolExecutor對象來生成一個線程池。接下來，先看ThreadPoolExecutor類的構造函數：

//間接調用最后一個構造函數，采用默認的任務拒絕策略AbortPolicy和默認的線程工廠public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue);//間接調用最后一個構造函數，采用默認的任務拒絕策略AbortPolicypublic ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);//間接調用最后一個構造函數，采用默認的默認的線程工廠public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);//前面三個分別調用了最后一個，主要的構造函數public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);

接下來，看下最后一個構造函數的具體實現：

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.acc = System.getSecurityManager() == null ?null :AccessController.getContext();//初始化對應的屬性this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;}

下面解釋下一下構造器中各個參數的含義：

1. corePoolSize

線程池中的核心線程數。當提交一個任務時，線程池創建一個新線程執行任務，直到當前線程數等于corePoolSize；如果當前線程數為corePoolSize，繼續提交的任務被保存到阻塞隊列中，等待被執行。

2. maximumPoolSize

線程池中允許的最大線程數。如果當前阻塞隊列滿了，且繼續提交任務，則創建新的線程執行任務，前提是當前線程數小于maximumPoolSize。

3. keepAliveTime

線程空閑時的存活時間。默認情況下，只有當線程池中的線程數大于corePoolSize時，keepAliveTime才會起作用，如果一個線程空閑的時間達到keepAliveTime，則會終止，直到線程池中的線程數不超過corePoolSize。但是如果調用了allowCoreThreadTimeOut(boolean)方法，keepAliveTime參數也會起作用，直到線程池中的線程數為0。

4. unit

keepAliveTime參數的時間單位。

5. workQueue

任務緩存隊列，用來存放等待執行的任務。如果當前線程數為corePoolSize，繼續提交的任務就會被保存到任務緩存隊列中，等待被執行。

一般來說，這里的BlockingQueue有以下三種選擇：

• SynchronousQueue：

  一個不存儲元素的阻塞隊列，每個插入操作必須等到另一個線程調用移除操作，否則插入操作一直處于阻塞狀態。

  因此，如果線程池中始終沒有空閑線程（任務提交的平均速度快于被處理的速度），可能出現無限制的線程增長。

• LinkedBlockingQueue：

  基于鏈表結構的阻塞隊列，如果不設置初始化容量，其容量為Integer.MAX_VALUE，即為無界隊列。

  因此，如果線程池中線程數達到了corePoolSize，且始終沒有空閑線程（任務提交的平均速度快于被處理的速度），任務緩存隊列可能出現無限制的增長。

• ArrayBlockingQueue：基于數組結構的有界阻塞隊列，按FIFO排序任務。

6. threadFactory

線程工廠，創建新線程時使用的線程工廠。

7. handler

任務拒絕策略，當阻塞隊列滿了，且線程池中的線程數達到maximumPoolSize，如果繼續提交任務，就會采取任務拒絕策略處理該任務，線程池提供了4種任務拒絕策略：

• AbortPolicy：丟棄任務并拋出RejectedExecutionException異常，默認策略；

• CallerRunsPolicy：由調用execute方法的線程執行該任務；

• DiscardPolicy：丟棄任務，但是不拋出異常;

• DiscardOldestPolicy：丟棄阻塞隊列最前面的任務，然后重新嘗試執行任務（重復此過程）。

  當然也可以根據應用場景實現RejectedExecutionHandler接口，自定義飽和策略，如記錄日志或持久化存儲不能處理的任務。

四、線程池的實現原理

1. 提交任務

線程池框架提供了兩種方式提交任務，submit()和execute()，通過submit()方法提交的任務可以返回任務執行的結果，通過execute()方法提交的任務不能獲取任務執行的結果。

submit()方法的實現有以下三種：

public Future<?> submit(Runnable task);public <T> Future<T> submit(Runnable task, T result);public <T> Future<T> submit(Callable<T> task);

下面以第一個方法為例簡單看一下submit()方法的實現：

public Future<?> submit(Runnable task) {if (task == null) throw new NullPointerException();RunnableFuture<Void> ftask = newTaskFor(task, null);execute(ftask);return ftask;}

submit()方法是在ThreadPoolExecutor的父類AbstractExecutorService類實現的，最終還是調用的ThreadPoolExecutor類的execute()方法，下面著重看一下execute()方法的實現。

public void execute(Runnable command) {if (command == null)throw new NullPointerException();//獲取線程池控制狀態int c = ctl.get();// (1)//worker數量小于corePoolSizeif (workerCountOf(c) < corePoolSize) {//創建worker,addWorker方法boolean參數用來判斷是否創建核心線程if (addWorker(command, true))//成功則返回return;//失敗則再次獲取線程池控制狀態c = ctl.get();}//(2)//線程池處于RUNNING狀態，將任務加入workQueue任務緩存隊列if (isRunning(c) && workQueue.offer(command)) {// 再次檢查，獲取線程池控制狀態，防止在任務入隊的過程中線程池關閉了或者線程池中沒有線程了int recheck = ctl.get();//線程池不處于RUNNING狀態，且將任務從workQueue移除成功if (! isRunning(recheck) && remove(command))//采取任務拒絕策略reject(command);//worker數量等于0else if (workerCountOf(recheck) == 0)//創建workeraddWorker(null, false);}//(3)else if (!addWorker(command, false)) //創建workerreject(command); //如果創建worker失敗，采取任務拒絕策略}

execute()方法的執行流程可以總結如下：

• 若線程池工作線程數量小于corePoolSize,則創建新線程來執行任務

• 若工作線程數量大于或等于corePoolSize,則將任務加入BlockingQueue

• 若無法將任務加入BlockingQueue(BlockingQueue已滿),且工作線程數量小于maximumPoolSize,則創建新的線程來執行任務

• 若工作線程數量達到maximumPoolSize,則創建線程失敗,采取任務拒絕策略

可以結合下面的兩張圖來理解線程池提交任務的執行流程。

2. 創建線程

從execute()方法的實現可以看出，addWorker()方法主要負責創建新的線程并執行任務，代碼實現如下：? ?

//addWorker有兩個參數:Runnable類型的firstTask,用于指定新增的線程執行的第一個任務;boolean類型的core,表示是否創建核心線程 //該方法的返回值代表是否成功新增一個線程private boolean addWorker(Runnable firstTask, boolean core) {retry:for (;;) {int c = ctl.get();int rs = runStateOf(c);// (1)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;//CAS操作遞增workCount//如果成功，那么創建線程前的所有條件校驗都滿足了，準備創建線程執行任務，退出retry循環//如果失敗，說明有其他線程也在嘗試往線程池中創建線程(往線程池提交任務可以是并發的)，則繼續往下執行if (compareAndIncrementWorkerCount(c))break retry;//重新獲取線程池控制狀態c = ctl.get();// 如果線程池的狀態發生了變更,如有其他線程關閉了這個線程池,那么需要回到外層的for循環if (runStateOf(c) != rs)continue retry;//如果只是CAS操作失敗的話，進入內層的for循環就可以了}}//到這里，創建線程前的所有條件校驗都滿足了，可以開始創建線程來執行任務//worker是否已經啟動boolean workerStarted = false;//是否已將這個worker添加到workers這個HashSet中boolean workerAdded = false;Worker w = null;try {//創建一個worker，從這里可以看出對線程的包裝w = new Worker(firstTask);//取出worker中的線程對象,Worker的構造方法會調用ThreadFactory來創建一個新的線程final Thread t = w.thread;if (t != null) {//獲取全局鎖, 并發的訪問線程池workers對象必須加鎖,持有鎖的期間線程池也不會被關閉final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {//重新獲取線程池的運行狀態int rs = runStateOf(ctl.get());//小于SHUTTDOWN即RUNNING//等于SHUTDOWN并且firstTask為null,不接受新的任務,但是會繼續執行等待隊列中的任務if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {//worker里面的thread不能是已啟動的if (t.isAlive())throw new IllegalThreadStateException();//將新創建的線程加入到線程池中workers.add(w);int s = workers.size();// 更新largestPoolSizeif (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}//線程添加線程池成功，則啟動新創建的線程if (workerAdded) {t.start();workerStarted = true;}}} finally {//若線程啟動失敗,做一些清理工作,例如從workers中移除新添加的worker并遞減wokerCountif (! workerStarted)addWorkerFailed(w);}//返回線程是否啟動成功return workerStarted;}

因為代碼(1)處的邏輯不利于理解，我們通過(1)的等價實現來理解:? ?

if (rs>=SHUTDOWN && !(rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) //等價實現 rs>=SHUTDOWN && (rs != SHUTDOWN || firstTask != null || workQueue.isEmpty())

其含義為,滿足下列條件之一則直接返回false，線程創建失敗:

• rs > SHUTDOWN,也就是STOP,TIDYING或TERMINATED，此時不再接受新的任務，且中斷正在執行的任務

• rs = SHUTDOWN且firstTask != null，此時不再接受任務，但是仍會處理任務緩存隊列中的任務

• rs = SHUTDOWN，隊列為空

多說一句，若線程池處于 SHUTDOWN， firstTask 為 null，且 workQueue 非空，那么還得創建線程繼續處理任務緩存隊列中的任務。

總結一下，addWorker()方法完成了如下幾件任務：

1. 原子性的增加workerCount

2. 將用戶給定的任務封裝成為一個worker，并將此worker添加進workers集合中

3. 啟動worker對應的線程

4. 若線程啟動失敗，回滾worker的創建動作，即從workers中移除新添加的worker，并原子性的減少workerCount

3. 工作線程的實現

從addWorker()方法的實現可以看出，工作線程的創建和啟動都跟ThreadPoolExecutor中的內部類Worker有關。下面我們分析Worker類來看一下工作線程的實現。

Worker類繼承自AQS類，具有鎖的功能；實現了Runable接口，可以將自身作為一個任務在線程中執行。

private final class Workerextends AbstractQueuedSynchronizerimplements Runnable

Worker的主要字段就下面三個，代碼也比較簡單。

//用來封裝worker的線程，線程池中真正運行的線程,通過線程工廠創建而來final Thread thread;//worker所對應的第一個任務，可能為空Runnable firstTask;//記錄當前線程完成的任務數volatile long completedTasks;

Worker的構造函數如下。

Worker(Runnable firstTask) {//設置AQS的state為-1，在執行runWorker()方法之前阻止線程中斷setState(-1);//初始化第一個任務this.firstTask = firstTask;//利用指定的線程工廠創建一個線程，注意，參數是Worker實例本身this//也就是當執行start方法啟動線程thread時，真正執行的是Worker類的run方法this.thread = getThreadFactory().newThread(this);}

Worker類繼承了AQS類，重寫了其相應的方法，實現了一個自定義的同步器，實現了不可重入鎖。

//是否持有獨占鎖protected boolean isHeldExclusively() {return getState() != 0;}//嘗試獲取鎖protected boolean tryAcquire(int unused) {if (compareAndSetState(0, 1)) {//設置獨占線程setExclusiveOwnerThread(Thread.currentThread());return true;}return false;}//嘗試釋放鎖protected boolean tryRelease(int unused) {//設置獨占線程為nullsetExclusiveOwnerThread(null);setState(0);return true;}//獲取鎖public void lock() { acquire(1); }//嘗試獲取鎖public boolean tryLock() { return tryAcquire(1); }//釋放鎖public void unlock() { release(1); }//是否持有鎖public boolean isLocked() { return isHeldExclusively(); }

Worker類還提供了一個中斷線程thread的方法。

void interruptIfStarted() {Thread t;//AQS狀態大于等于0，worker對應的線程不為null，且該線程沒有被中斷if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {try {t.interrupt();} catch (SecurityException ignore) {}}}

再來看一下Worker類的run()方法的實現，會發現run()方法最終調用了ThreadPoolExecutor類的runWorker()方法。

public void run() {runWorker(this);}

4. 線程復用機制

通過上文可以知道，worker中的線程start 后，執行的是worker的run()方法，而run()方法最終會調用ThreadPoolExecutor類的runWorker()方法，runWorker()方法實現了線程池中的線程復用機制。下面我們來看一下runWorker()方法的實現。

final void runWorker(Worker w) {//獲取當前線程Thread wt = Thread.currentThread();//獲取w的firstTaskRunnable task = w.firstTask;//設置w的firstTask為nullw.firstTask = null;// 釋放鎖，設置AQS的state為0，允許中斷w.unlock();//用于標識線程是否異常終止，finally中processWorkerExit()方法會有不同邏輯boolean completedAbruptly = true;try {//循環調用getTask()獲取任務,不斷從任務緩存隊列獲取任務并執行while (task != null || (task = getTask()) != null) {//進入循環內部，代表已經獲取到可執行的任務，則對worker對象加鎖，保證線程在執行任務過程中不會被中斷w.lock();if ((runStateAtLeast(ctl.get(), STOP) || //若線程池狀態大于等于STOP，那么意味著該線程要中斷(Thread.interrupted() && //線程被中斷runStateAtLeast(ctl.get(), STOP))) && //且是因為線程池內部狀態變化而被中斷!wt.isInterrupted()) //確保該線程未被中斷//發出中斷請求wt.interrupt();try {//開始執行任務前的Hook方法beforeExecute(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 {//執行任務后的Hook方法afterExecute(task, thrown);}} finally {//置空task，準備通過getTask()獲取下一個任務task = null;//completedTasks遞增w.completedTasks++;//釋放掉worker持有的獨占鎖w.unlock();}}completedAbruptly = false;} finally {//到這里，線程執行結束，需要執行結束線程的一些清理工作//線程執行結束可能有兩種情況：//1.getTask()返回null，也就是說，這個worker的使命結束了，線程執行結束//2.任務執行過程中發生了異常//第一種情況，getTask()返回null，那么getTask()中會將workerCount遞減//第二種情況，workerCount沒有進行處理，這個遞減操作會在processWorkerExit()中處理processWorkerExit(w, completedAbruptly);}}

runWorker()方法是線程池的核心，實現了線程池中的線程復用機制，來看一下

runWorker()方法都做了哪些工作：

1. 運行第一個任務firstTask之后，循環調用getTask()方法獲取任務,不斷從任務緩存隊列獲取任務并執行；

2. 獲取到任務之后就對worker對象加鎖，保證線程在執行任務的過程中不會被中斷，任務執行完會釋放鎖；

3. 在執行任務的前后，可以根據業務場景重寫beforeExecute()和afterExecute()等Hook方法；

4. 執行通過getTask()方法獲取到的任務

5. 線程執行結束后，調用processWorkerExit()方法執行結束線程的一些清理工作

從runWorker()方法的實現可以看出，runWorker()方法中主要調用了getTask()方法和processWorkerExit()方法，下面分別看一下這兩個方法的實現。

getTask()的實現

getTask()方法用來不斷地從任務緩存隊列獲取任務并交給線程執行，下面分析一下其實現。?

private Runnable getTask() {//標識當前線程是否超時未能獲取到task對象boolean timedOut = false;for (;;) {//獲取線程池的控制狀態int c = ctl.get();//獲取線程池的運行狀態int rs = runStateOf(c);//如果線程池狀態大于等于STOP，或者處于SHUTDOWN狀態，并且阻塞隊列為空，線程池工作線程數量遞減，方法返回null，回收線程if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {decrementWorkerCount();return null;}//獲取worker數量int wc = workerCountOf(c);//標識當前線程在空閑時，是否應該超時回收// 如果allowCoreThreadTimeOut為ture，或當前線程數大于核心池大小，則需要超時回收boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;//如果worker數量大于maximumPoolSize(有可能調用了 setMaximumPoolSize(),導致worker數量大于maximumPoolSize)if ((wc > maximumPoolSize || (timed && timedOut)) //或者獲取任務超時&& (wc > 1 || workQueue.isEmpty())) { //workerCount大于1或者阻塞隊列為空（在阻塞隊列不為空時，需要保證至少有一個工作線程）if (compareAndDecrementWorkerCount(c))//線程池工作線程數量遞減，方法返回null，回收線程return null;//線程池工作線程數量遞減失敗，跳過剩余部分，繼續循環continue;}try {//如果允許超時回收，則調用阻塞隊列的poll()，只在keepAliveTime時間內等待獲取任務，一旦超過則返回null//否則調用take()，如果隊列為空，線程進入阻塞狀態，無限時等待任務，直到隊列中有可取任務或者響應中斷信號退出Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();//若task不為null，則返回成功獲取的task對象if (r != null)return r;// 若返回task為null，表示線程空閑時間超時，則設置timeOut為truetimedOut = true;} catch (InterruptedException retry) {//如果此worker發生了中斷，采取的方案是重試，沒有超時//在哪些情況下會發生中斷？調用setMaximumPoolSize()，shutDown()，shutDownNow()timedOut = false;}}}

接下來總結一下getTask()方法會在哪些情況下返回：

1. 線程池處于RUNNING狀態，阻塞隊列不為空，返回成功獲取的task對象

2. 線程池處于SHUTDOWN狀態，阻塞隊列不為空，返回成功獲取的task對象

3. 線程池狀態大于等于STOP，返回null，回收線程

4. 線程池處于SHUTDOWN狀態，并且阻塞隊列為空，返回null，回收線程

5. worker數量大于maximumPoolSize，返回null，回收線程

6. 線程空閑時間超時，返回null，回收線程

processWorkerExit()的實現

processWorkerExit()方法負責執行結束線程的一些清理工作，下面分析一下其實現。

private void processWorkerExit(Worker w, boolean completedAbruptly) {//如果用戶任務執行過程中發生了異常，則需要遞減workerCountif (completedAbruptly)decrementWorkerCount();final ReentrantLock mainLock = this.mainLock;//獲取全局鎖mainLock.lock();try {//將worker完成任務的數量累加到總的完成任務數中completedTaskCount += w.completedTasks;//從workers集合中移除該workerworkers.remove(w);} finally {//釋放鎖mainLock.unlock();}//嘗試終止線程池tryTerminate();//獲取線程池控制狀態int c = ctl.get();if (runStateLessThan(c, STOP)) { //線程池運行狀態小于STOPif (!completedAbruptly) { //如果用戶任務執行過程中發生了異常，則直接調用addWorker()方法創建線程//是否允許核心線程超時int min = allowCoreThreadTimeOut ? 0 : corePoolSize;//允許核心超時并且workQueue阻塞隊列不為空，那線程池中至少有一個工作線程if (min == 0 && ! workQueue.isEmpty())min = 1;//如果工作線程數量workerCount大于等于核心池大小corePoolSize，//或者允許核心超時并且workQueue阻塞隊列不為空時，線程池中至少有一個工作線程，直接返回if (workerCountOf(c) >= min)return;//若不滿足上述條件，則調用addWorker()方法創建線程}//創建新的線程取代當前線程addWorker(null, false);}}

processWorkerExit()方法中主要調用了tryTerminate()方法，下面看一下tryTerminate()方法的實現。

final void tryTerminate() {for (;;) {//獲取線程池控制狀態int c = ctl.get();if (isRunning(c) || //線程池的運行狀態為RUNNINGrunStateAtLeast(c, TIDYING) || //線程池的運行狀態大于等于TIDYING(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) //線程池的運行狀態為SHUTDOWN且阻塞隊列不為空//不能終止，直接返回return;//只有當線程池的運行狀態為STOP，或線程池運行狀態為SHUTDOWN且阻塞隊列為空時，可以執行到這里//如果線程池工作線程的數量不為0if (workerCountOf(c) != 0) {//僅僅中斷一個空閑的workerinterruptIdleWorkers(ONLY_ONE);return;}//只有當線程池工作線程的數量為0時可以執行到這里final ReentrantLock mainLock = this.mainLock;//獲取全局鎖mainLock.lock();try {if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { //CAS操作設置線程池運行狀態為TIDYING，工作線程數量為0try {//執行terminated()鉤子方法terminated();} finally {//設置線程池運行狀態為TERMINATED，工作線程數量為0ctl.set(ctlOf(TERMINATED, 0));//喚醒在termination條件上等待的所有線程termination.signalAll();}return;}} finally {//釋放鎖mainLock.unlock();}//若CAS操作失敗則重試}}

tryTerminate()方法的作用是嘗試終止線程池，它會在所有可能終止線程池的地方被調用，滿足終止線程池的條件有兩個：首先，線程池狀態為STOP,或者為SHUTDOWN且任務緩存隊列為空；其次，工作線程數量為0。

滿足了上述兩個條件之后，tryTerminate()方法獲取全局鎖，設置線程池運行狀態為TIDYING，之后執行terminated()鉤子方法，最后設置線程池狀態為TERMINATED。

至此，線程池運行狀態變為TERMINATED，工作線程數量為0，workers已清空，且workQueue也已清空，所有線程都執行結束，線程池的生命周期到此結束。

5. 關閉線程池

關閉線程池有兩個方法，shutdown()和shutdownNow()，下面分別看一下這兩個方法的實現。

shutdown()的實現

shutdown()方法將線程池運行狀態設置為SHUTDOWN，此時線程池不會接受新的任務，但會處理阻塞隊列中的任務。

public void shutdown() {final ReentrantLock mainLock = this.mainLock;//獲取全局鎖mainLock.lock();try {//檢查shutdown權限checkShutdownAccess();//設置線程池運行狀態為SHUTDOWNadvanceRunState(SHUTDOWN);//中斷所有空閑workerinterruptIdleWorkers();//用onShutdown()鉤子方法onShutdown();} finally {//釋放鎖mainLock.unlock();}//嘗試終止線程池tryTerminate();}

shutdown()方法首先會檢查是否具有shutdown的權限，然后設置線程池的運行狀態為SHUTDOWN，之后中斷所有空閑的worker，再調用onShutdown()鉤子方法，最后嘗試終止線程池。

shutdown()方法調用了interruptIdleWorkers()方法中斷所有空閑的worker，其實現如下。

private void interruptIdleWorkers() {interruptIdleWorkers(false);}//onlyOne標識是否只中斷一個線程private void interruptIdleWorkers(boolean onlyOne) {final ReentrantLock mainLock = this.mainLock;//獲取全局鎖mainLock.lock();try {//遍歷workers集合for (Worker w : workers) {//worker對應的線程Thread t = w.thread;//線程未被中斷且成功獲得鎖if (!t.isInterrupted() && w.tryLock()) {try {//發出中斷請求t.interrupt();} catch (SecurityException ignore) {} finally {//釋放鎖w.unlock();}}//若只中斷一個線程，則跳出循環if (onlyOne)break;}} finally {//釋放鎖mainLock.unlock();}}

shutdownNow()的實現

shutdownNow()方法將線程池運行狀態設置為STOP，此時線程池不會接受新任務，也不會處理阻塞隊列中的任務，并且中斷正在運行的任務。

public List<Runnable> shutdownNow() {List<Runnable> tasks;final ReentrantLock mainLock = this.mainLock;//獲取全局鎖mainLock.lock();try {//檢查shutdown權限checkShutdownAccess();//設置線程池運行狀態為STOPadvanceRunState(STOP);//中斷所有workerinterruptWorkers();//將任務緩存隊列中等待執行的任務取出并放到list中tasks = drainQueue();} finally {//釋放鎖mainLock.unlock();}//嘗試終止線程池tryTerminate();//返回任務緩存隊列中等待執行的任務列表return tasks;}

shutdownNow()方法與shutdown()方法相似，不同之處在于，前者設置線程池的運行狀態為STOP，之后中斷所有的worker(并非只是空閑的worker)，嘗試終止線程池之后，返回任務緩存隊列中等待執行的任務列表。

shutdownNow()方法調用了interruptWorkers()方法中斷所有的worker(并非只是空閑的worker)，其實現如下。

private void interruptWorkers() {final ReentrantLock mainLock = this.mainLock;//獲取全局鎖mainLock.lock();try {//遍歷workers集合for (Worker w : workers)//調用Worker類的interruptIfStarted()方法中斷線程w.interruptIfStarted();} finally {//釋放鎖mainLock.unlock();}}

五、總結

至此，我們已經閱讀了線程池框架的核心類ThreadPoolExecutor類的大部分源碼，由衷地贊嘆這個類很多地方設計的巧妙之處：

• 將線程池的運行狀態和工作線程數量打包在一起，并使用了大量的位運算

• 使用CAS操作更新線程控制狀態ctl，確保對ctl的更新是原子操作

• 內部類Worker類繼承了AQS，實現了一個自定義的同步器，實現了不可重入鎖

• 使用while循環自旋地從任務緩存隊列中獲取任務并執行，實現了線程復用機制

• 調用interrupt()方法中斷線程，但注意該方法并不能直接中斷線程的運行，只是發出了中斷信號，配合BlockingQueue的take()，poll()方法的使用，打斷線程的阻塞狀態

其實，線程池的本質就是生產者消費者模式，線程池的調用者不斷向線程池提交任務，線程池里面的工作線程不斷獲取這些任務并執行(從任務緩存隊列獲取任務或者直接執行任務)。

讀完本文，相信大家對線程池的實現原理有了深刻的認識，比如向線程池提交一個任務之后線程池的執行流程，一個任務從被提交到被執行會經歷哪些過程，一個工作線程從被創建到正常執行到執行結束的執行過程，等等。

最后，再附上我歷時三個月總結的?Java 面試 + Java 后端技術學習指南，筆者這幾年及春招的總結，github 1.4k star，拿去不謝！

下載方式

1.?首先掃描下方二維碼

2.?后臺回復「Java面試」即可獲取

總結

以上是生活随笔為你收集整理的深入分析线程池的实现原理的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。