詳解Java并發包中線程池ThreadPoolExecutor
線程池的使用主要是解決兩個問題：①當執行大量異步任務的時候線程池能夠提供更好的性能，在不使用線程池時候，每當需要執行異步任務的時候直接new一個線程來運行的話，線程的創建和銷毀都是需要開銷的。而線程池中的線程是可復用的，不需要每次執行異步任務的時候重新創建和銷毀線程；②線程池提供一種資源限制和管理的手段，比如可以限制線程的個數，動態的新增線程等等。
在下面的分析中，我們可以看到，線程池使用一個Integer的原子類型變量來記錄線程池狀態和線程池中的線程數量，通過線程池狀態來控制任務的執行，每個工作線程Worker線程可以處理多個任務。
二、ThreadPoolExecutor類2.1、ThreadPoolExecutor成員變量以含義ThreadPoolExecutor繼承了AbstractExecutorService，其中的成員變量ctl是一個Integer類型的原子變量，用來記錄線程池的狀態和線程池中的線程的個數。這里（Integer看做32位）ctl高三位表示線程池的狀態，后面的29位表示線程池中的線程個數。如下所示是ThreadPoolExecutor源碼中的成員變量
//（高3位）表示線程池狀態，（低29位）表示線程池中線程的個數；// 默認狀態是RUNNING，線程池中線程個數為0private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));//表示具體平臺下Integer的二進制位數-3后的剩余位數表示的數才是線程的個數;//其中Integer.SIZE=32,-3之后的低29位表示的就是線程的個數了private static final int COUNT_BITS = Integer.SIZE - 3;//線程最大個數（低29位）00011111111111111111111111111111（1<<29-1）private static final int CAPACITY = (1 << COUNT_BITS) - 1;//線程池狀態（高3位表示線程池狀態）//111 00000000000000000000000000000private static final int RUNNING = -1 << COUNT_BITS;//000 00000000000000000000000000000private static final int SHUTDOWN = 0 << COUNT_BITS;//001 00000000000000000000000000000private static final int STOP = 1 << COUNT_BITS;//010 00000000000000000000000000000private static final int TIDYING = 2 << COUNT_BITS;//011 00000000000000000000000000000private static final int TERMINATED = 3 << COUNT_BITS;//獲取高3位（運行狀態）==> c & 11100000000000000000000000000000private static int runStateOf(int c) { return c & ~CAPACITY; }//獲取低29位（線程個數）==> c & 00011111111111111111111111111111private static int workerCountOf(int c) { return c & CAPACITY; }//計算原子變量ctl新值（運行狀態和線程個數）private static int ctlOf(int rs, int wc) { return rs | wc; }
下面我們簡單解釋一下上面的線程狀態的含義：
①RUNNING：接受新任務并處理阻塞隊列中的任務
②SHUTDOWN：拒絕新任務但是處理阻塞隊列中的任務
③STOP：拒絕新任務并拋棄阻塞隊列中的任務，同時會中斷當前正在執行的任務
④TIDYING：所有任務執行完之后（包含阻塞隊列中的任務）當前線程池中活躍的線程數量為0，將要調用terminated方法
⑥TERMINATED：終止狀態。terminated方法調用之后的狀態
2.2、ThreadPoolExecutor的參數以及實現原理①corePoolSize：線程池核心現車個數
②workQueue：用于保存等待任務執行的任務的阻塞隊列（比如基于數組的有界阻塞隊列ArrayBlockingQueue、基于鏈表的無界阻塞隊列LinkedBlockingQueue等等）
③maximumPoolSize：線程池最大線程數量
④ThreadFactory：創建線程的工廠
⑤RejectedExecutionHandler：拒絕策略，表示當隊列已滿并且線程數量達到線程池最大線程數量的時候對新提交的任務所采取的策略，主要有四種策略：AbortPolicy（拋出異常）、CallerRunsPolicy（只用調用者所在線程來運行該任務）、DiscardOldestPolicy（丟掉阻塞隊列中最近的一個任務來處理當前提交的任務）、DiscardPolicy（不做處理，直接丟棄掉）
⑥keepAliveTime：存活時間，如果當前線程池中的數量比核心線程數量多，并且當前線程是閑置狀態，該變量就是這些線程的最大生存時間
⑦TimeUnit：存活時間的時間單位。
根據上面的參數介紹，簡單了解一下線程池的實現原理，以提交一個新任務為開始點，分析線程池的主要處理流程
①newFixedThreadPool：創建一個核心線程個數和最大線程個數均為nThreads的線程池，并且阻塞隊列長度為Integer.MAX_VALUE,keepAliveTime=0說明只要線程個數比核心線程個數多并且當前空閑即回收。
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());}
②newSingleThreadExecutor：創建一個核心線程個數和最大線程個數都為1 的線程池，并且阻塞隊列長度為Integer.MAX_VALUE，keepAliveTime=0說明只要線程個數比核心線程個數多并且當前線程空閑即回收該線程。
public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));}
③newCachedThreadPoolExecutor：創建一個按需創建線程的線程池，初始線程個數為0，最多線程個數為Integer.MAX_VALUE，并且阻塞隊列為同步隊列（最多只有一個元素），keepAliveTime=60說明只要當前線程在60s內空閑則回收。這個類型的線程池的特點就是：加入同步隊列的任務會被馬上執行，同步隊列中最多只有一個任務
public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());}2.4、ThreadPoolExecutor中的其他成員
//獨占鎖，用來控制新增工作線程Worker操作的原子性private final ReentrantLock mainLock = new ReentrantLock();//工作線程集合，Worker繼承了AQS接口和Runnable接口，是具體處理任務的線程對象//Worker實現AQS，并自己實現了簡單不可重入獨占鎖，其中state=0表示當前鎖未被獲取狀態，state=1表示鎖被獲取，//state=-1表示Work創建時候的默認狀態，創建時候設置state=-1是為了防止runWorker方法運行前被中斷private final HashSet<Worker> workers = new HashSet<Worker>();//termination是該鎖對應的條件隊列，在線程調用awaitTermination時候用來存放阻塞的線程private final Condition termination = mainLock.newCondition();三、execute(Runnable command)方法實現
executor方法的作用是提交任務command到線程池執行，可以簡單的按照下面的圖進行理解，ThreadPoolExecutor的實現類似于一個生產者消費者模型，當用戶添加任務到線程池中相當于生產者生產元素，workers工作線程則直接執行任務或者從任務隊列中獲取任務，相當于消費之消費元素。
public void execute(Runnable command) { //（1）首先檢查任務是否為null，為null拋出異常，否則進行下面的步驟 if (command == null)throw new NullPointerException(); //（2）ctl值中包含了當前線程池的狀態和線程池中的線程數量 int c = ctl.get(); //（3）workerCountOf方法是獲取低29位，即獲取當前線程池中的線程個數，如果小于corePoolSize，就開啟新的線程運行 if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true)) return;c = ctl.get(); } //（4）如果線程池處理RUNNING狀態，就添加任務到阻塞隊列中 if (isRunning(c) && workQueue.offer(command)) {//（4-1）二次檢查，獲取ctl值int recheck = ctl.get();//（4-2）如果當前線程池不是出于RUNNING狀態，就從隊列中刪除任務，并執行拒絕策略if (! isRunning(recheck) && remove(command)) reject(command);//（4-3）否則，如果線程池為空，就添加一個線程else if (workerCountOf(recheck) == 0) addWorker(null, false); } //（5）如果隊列滿，則新增線程，如果新增線程失敗，就執行拒絕策略 else if (!addWorker(command, false))reject(command);}
我們在看一下上面代碼的執行流程，按照標記的數字進行分析：
①步驟（3）判斷當前線程池中的線程個數是否小于corePoolSize，如果小于核心線程數，會向workers里面新增一個核心線程執行任務。
②如果當前線程池中的線程數量大于核心線程數，就執行（4）。（4）首先判斷當前線程池是否處于RUNNING狀態，如果處于該狀態，就添加任務到任務隊列中，這里需要判斷線程池的狀態是因為線程池可能已經處于非RUNNING狀態，而在非RUNNING狀態下是需要拋棄新任務的。
③如果想任務隊列中添加任務成功，需要進行二次校驗，因為在添加任務到任務隊列后，可能線程池的狀態發生了變化，所以這里需要進行二次校驗，如果當前線程池已經不是RUNNING狀態了，需要將任務從任務隊列中移除，然后執行拒絕策略；如果二次校驗通過，則執行4-3代碼重新判斷當前線程池是否為空，如果線程池為空沒有線程，那么就需要新創建一個線程。
④如果上面的步驟（4）創建添加任務失敗，說明隊列已滿，那么（5）會嘗試再開啟新的線程執行任務（類比上圖中的thread3和thread4，即不是核心線程的那些線程），如果當前線程池中的線程個數已經大于最大線程數maximumPoolSize，表示不能開啟新的線程。這就屬于線程池滿并且任務隊列滿，就需要執行拒絕策略了。
下面我們在看看addWorker方法的實現
private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) {int c = ctl.get();int rs = runStateOf(c);//(6)檢查隊列是否只在必要時候為空if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false;//(7)使用CAS增加線程個數for (;;) { //根據ctl值獲得當前線程池中的線程數量 int wc = workerCountOf(c); //(7-1)如果線程數量超出限制，返回false if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false; //(7-2)CAS增加線程數量，同時只有一個線程可以成功 if (compareAndIncrementWorkerCount(c))break retry; c = ctl.get(); // 重新讀取ctl值 //(7-3)CAS失敗了，需要查看當前線程池狀態是否發生變化，如果發生變化需要跳轉到外層循環嘗試重新獲取線程池狀態，否則內層循環重新進行CAS增加線程數量 if (runStateOf(c) != rs)continue retry;} } //(8)執行到這里說明CAS增加新線程個數成功了，我們需要開始創建新的工作線程Worker boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try {//(8-1)創建新的workerw = new Worker(firstTask);final Thread t = w.thread;if (t != null) { final ReentrantLock mainLock = this.mainLock; //(8-2)加獨占鎖，保證workers的同步，可能線程池中的多個線程調用了線程池的execute方法 mainLock.lock(); try {// (8-3)重新檢查線程池狀態，以免在獲取鎖之前調用shutdown方法改變線程池狀態int rs = runStateOf(ctl.get());if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException(); //(8-4)添加新任務 workers.add(w); int s = workers.size(); if (s > largestPoolSize)largestPoolSize = s; workerAdded = true;} } finally {mainLock.unlock(); } //(8-6)添加新任務成功之后，啟動任務 if (workerAdded) {t.start();workerStarted = true; }} } finally {if (! workerStarted) addWorkerFailed(w); } return workerStarted;}
簡單再分析說明一下上面的代碼，addWorker方法主要分為兩部分，第一部分是使用CAS線程安全的添加線程數量，第二部分則是創建新的線程并且將任務并發安全的添加到新的workers之中，然后啟動線程執行。
①代碼（6）中檢查隊列是否只在必要時候為空，只有線程池狀態符合條件才能夠進行下面的步驟，從（6）中的判斷條件來看，下面的集中情況addWorker會直接返回false
( I )當前線程池狀態為STOP，TIDYING或者TERMINATED ； (I I）當前線程池狀態為SHUTDOWN并且已經有了第一個任務; (I I I)當前線程池狀態為SHUTDOWN并且任務隊列為空
②外層循環中判斷條件通過之后，在內層循環中使用CAS增加線程數，當CAS成功就退出雙重循環進行（8）步驟代碼的執行，如果失敗需要查看當前線程池的狀態是否發生變化，如果發生變化需要進行外層循環重新判斷線程池狀態然后在進入內層循環重新進行CAS增加線程數，如果線程池狀態沒有發生變化但是上一次CAS失敗就繼續進行CAS嘗試。
③執行到（8）代碼處，表明當前已經成功增加 了線程數，但是還沒有線程執行任務。ThreadPoolExecutor中使用全局獨占鎖mainLock來控制將新增的工作線程Worker線程安全的添加到工作者線程集合workers中。
④（8-2）獲取了獨占鎖，但是在獲取到鎖之后，還需要進行重新檢查線程池的狀態，這是為了避免在獲取全局獨占鎖之前其他線程調用了shutDown方法關閉了線程池。如果線程池已經關閉需要釋放鎖。否則將新增的線程添加到工作集合中，釋放鎖啟動線程執行任務。
上面的addWorker方法最后幾行中，會判斷添加工作線程是否成功，如果失敗，會執行addWorkerFailed方法，將任務從workers中移除，并且workerCount做-1操作。
private void addWorkerFailed(Worker w) { final ReentrantLock mainLock = this.mainLock; //獲取鎖 mainLock.lock(); try { //如果worker不為null if (w != null) //workers移除worker workers.remove(w); //通過CAS操作，workerCount-1 decrementWorkerCount(); tryTerminate(); } finally { //釋放鎖 mainLock.unlock(); }}四、工作線程Worker的執行4.1、工作線程Worker類源碼分析
上面查看addWorker方法在CAS更新線程數成功之后，下面就是創建新的Worker線程執行任務，所以我們這里先查看Worker類，下面是Worker類的源碼，我們可以看出，Worker類繼承了AQS并實現了Runnable接口，所以他既是一個自定義的同步組件，也是一個執行任務的線程類。下面我們分析Worker類的執行
private final class Worker extends AbstractQueuedSynchronizer implements Runnable{ /** 使用線程工廠創建的線程，執行任務 */ final Thread thread; /** 初始化執行任務 */ Runnable firstTask; /** 計數 */ volatile long completedTasks; /** * 給出初始firstTask，線程創建工廠創建新的線程 */ Worker(Runnable firstTask) {setState(-1); // 防止在調用runWorker之前被中斷this.firstTask = firstTask;this.thread = getThreadFactory().newThread(this); //使用threadFactory創建線程 } /** run方法實際上執行的是runWorker方法 */ public void run() {runWorker(this); } // 關于同步狀態（鎖） // // 同步狀態state=0表示鎖未被獲取 // 同步狀態state=1表示鎖被獲取 protected boolean isHeldExclusively() {return getState() != 0; } //下面都是重寫AQS的方法，Worker為自定義的同步組件 protected boolean tryAcquire(int unused) {if (compareAndSetState(0, 1)) { setExclusiveOwnerThread(Thread.currentThread()); return true;}return false; } protected boolean tryRelease(int unused) {setExclusiveOwnerThread(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(); } void interruptIfStarted() {Thread t;if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) { try {t.interrupt(); } catch (SecurityException ignore) { }} }}
在構造函數中我們可以看出，首先將同步狀態state置為-1，而Worker這個同步組件的state有三個值，其中state=-1表示Work創建時候的默認狀態，創建時候設置state=-1是為了防止runWorker方法運行前被中斷前面說到過這個結論，這里置為-1是為了避免當前Worker在調用runWorker方法之前被中斷（當其他線程調用線程池的shutDownNow時候，如果Worker的state>=0則會中斷線程），設置為-1就不會被中斷了。而Worker實現Runnable接口，那么需要重寫run方法，在run方法中，我們可以看到，實際上執行的是runWorker方法，在runWorker方法中，會首先調用unlock方法，該方法會將state置為0，所以這個時候調用shutDownNow方法就會中斷當前線程，而這個時候已經進入了runWork方法了，就不會在還沒有執行runWorker方法的時候就中斷線程。
4.2、runWorker方法的源碼分析final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // 這個時候調用unlock方法，將state置為0，就可以被中斷了 boolean completedAbruptly = true; try {//(10)如果當前任務為null，或者從任務隊列中獲取到的任務為null，就跳轉到(11)處執行清理工作while (task != null || (task = getTask()) != null) { //task不為null，就需要線程執行任務，這個時候，需要獲取工作線程內部持有的獨占鎖 w.lock(); /**如果線程池已被停止(STOP)(至少大于STOP狀態)，要確保線程都被中斷 * 如果狀態不對，檢查當前線程是否中斷并清除中斷狀態，并且再次檢查線程池狀態是否大于STOP * 如果上述滿足，檢查該對象是否處于中斷狀態，不清除中斷標記 */ if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())//中斷該對象wt.interrupt(); try {//執行任務之前要做的事情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 { //執行任務之后的方法 afterExecute(task, thrown);} } finally {task = null;//更新當前已完成任務數量w.completedTasks++;//釋放鎖w.unlock(); }}completedAbruptly = false; } finally {//執行清理工作：處理并退出當前workerprocessWorkerExit(w, completedAbruptly); }}
我們梳理一下runWorker方法的執行流程
①首先先執行unlock方法，將Worker的state置為0，這樣工作線程就可以被中斷了（后續的操作如果線程池關閉就需要線程被中斷）
②首先判斷判斷當前的任務（當前工作線程中的task，或者從任務隊列中取出的task）是否為null，如果不為null就往下執行，為null就執行processWorkerExit方法。
③獲取工作線程內部持有的獨占鎖（避免在執行任務期間，其他線程調用shutdown后正在執行的任務被中斷，shutdown只會中斷當前被阻塞掛起的沒有執行任務的線程）
④然后執行beforeExecute()方法，該方法為擴展接口代碼，表示在具體執行任務之前所做的一些事情，然后執行task.run()方法執行具體任務，執行完之后會調用afterExecute()方法，用以處理任務執行完畢之后的工作，也是一個擴展接口代碼。
⑤更新當前線程池完成的任務數，并釋放鎖
4.3、執行清理工作的方法processWorkerExit下面是方法processWorkerExit的源碼，在下面的代碼中
①首先(1-1)處統計線程池完成的任務個數，并且在此之前獲取全局鎖，然后更新當前的全局計數器，然后從工作線程集合中移除當前工作線程，完成清理工作。
②代碼(1-2)調用了tryTerminate方法，在該方法中，判斷了當前線程池狀態是SHUTDOWN并且隊列不為空或者當前線程池狀態為STOP并且當前線程池中沒有活動線程，則置線程池狀態為TERMINATED。如果設置稱為了TERMINATED狀態，還需要調用全局條件變量termination的signalAll方法喚醒所有因為調用線程池的awaitTermination方法而被阻塞住的線程，使得線程池中的所有線程都停止，從而使得線程池為TERMINATED狀態。
③代碼(1-3)處判斷當前線程池中的線程個數是否小于核心線程數，如果是，需要新增一個線程保證有足夠的線程可以執行任務隊列中的任務或者提交的任務。
private void processWorkerExit(Worker w, boolean completedAbruptly) { /* *completedAbruptly:是由runWorker傳過來的參數，表示是否突然完成的意思 *當在就是在執行任務過程當中出現異常，就會突然完成，傳true * *如果是突然完成，需要通過CAS操作，更新workerCount(-1操作) *不是突然完成，則不需要-1，因為getTask方法當中已經-1(getTask方法中執行了decrementWorkerCount()方法) */ if (completedAbruptly)decrementWorkerCount(); //(1-1)在統計完成任務個數之前加上全局鎖，然后統計線程池中完成的任務個數并更新全局計數器，并從工作集中刪除當前worker final ReentrantLock mainLock = this.mainLock; mainLock.lock(); //獲得全局鎖 try {completedTaskCount += w.completedTasks; //更新已完成的任務數量workers.remove(w); //將完成該任務的線程worker從工作線程集合中移除 } finally {mainLock.unlock(); //釋放鎖 } /**(1-2) * 這一個方法調用完成了下面的事情： * 判斷如果當前線程池狀態是SHUTDOWN并且工作隊列為空， * 或者當前線程池狀態是STOP并且當前線程池里面沒有活動線程， * 則設置當前線程池狀態為TERMINATED，如果設置成了TERMINATED狀態， * 還需要調用條件變量termination的signAll方法激活所有因為調用線程池的awaitTermination方法而被阻塞的線程 */ tryTerminate(); //（1-3）如果當前線程池中線程數小于核心線程，則增加核心線程數 int c = ctl.get(); //判斷當前線程池的狀態是否小于STOP（RUNNING或者SHUTDOWN） if (runStateLessThan(c, STOP)) {//如果任務忽然完成，執行后續的代碼if (!completedAbruptly) { //allowCoreThreadTimeOut表示是否允許核心線程超時，默認為false //min這里當默認為allowCoreThreadTimeOut默認為false的時候，min置為coorPoolSize int min = allowCoreThreadTimeOut ? 0 : corePoolSize; //這里說明：如果允許核心線程超時，那么allowCoreThreadTimeOut可為true，那么min值為0，不需要維護核心線程了 //如果min為0并且任務隊列不為空 if (min == 0 && ! workQueue.isEmpty())min = 1; //這里表示如果min為0，且隊列不為空，那么至少需要一個核心線程存活來保證任務的執行 //如果工作線程數大于min,表示當前線程數滿足，直接返回 if (workerCountOf(c) >= min)return; // replacement not needed}addWorker(null, false); }}
在tryTerminate方法中，我們簡單說明了該方法的作用，下面是該方法的源碼，可以看出源碼實現上和上面所總結的功能是差不多的
final void tryTerminate() { for (;;) {//獲取線程池狀態int c = ctl.get();//如果線程池狀態為RUNNING//或者狀態大于TIDYING//或者狀態==SHUTDOWN并未任務隊列不為空//直接返回，不能調用terminated方法if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) return;//如果線程池中工作線程數不為0，需要中斷線程if (workerCountOf(c) != 0) { // Eligible to terminate interruptIdleWorkers(ONLY_ONE); return;}//獲得線程池的全局鎖final ReentrantLock mainLock = this.mainLock;mainLock.lock();try { //通過CAS操作，將線程池狀態設置為TIDYING if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { //private static int ctlOf(int rs, int wc) { return rs | wc; }try { //調用terminated方法 terminated();} finally { //最終將線程狀態設置為TERMINATED ctl.set(ctlOf(TERMINATED, 0)); //調用條件變量termination的signaAll方法喚醒所有因為 //調用線程池的awaitTermination方法而被阻塞的線程 //private final Condition termination = mainLock.newCondition(); termination.signalAll();}return; }} finally { mainLock.unlock();}// else retry on failed CAS }}五、補充（shutdown、shutdownNow、awaitTermination方法）5.1、shutdown操作
我們在使用線程池的時候知道，調用shutdown方法之后線程池就不會再接受新的任務了，但是任務隊列中的任務還是需要執行完的。調用該方法會立刻返回，并不是等到線程池的任務隊列中的所有任務執行完畢在返回的。
public void shutdown() { //獲得線程池的全局鎖 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try {//進行權限檢查checkShutdownAccess();//設置當前線程池的狀態的SHUTDOWN，如果線程池狀態已經是該狀態就會直接返回，下面我們會分析這個方法的源碼advanceRunState(SHUTDOWN);//設置中斷 標志interruptIdleWorkers();onShutdown(); // hook for ScheduledThreadPoolExecutor } finally {mainLock.unlock(); } //嘗試將狀態變為TERMINATED，上面已經分析過該方法的源碼 tryTerminate();}
該方法的源碼比較簡短，首先檢查了安全管理器，是查看當前調用shutdown命令的線程是否有關閉線程的權限，如果有權限還需要看調用線程是否有中斷工作線程的權限，如果沒有權限將會拋出SecurityException異常或者空指針異常。下面我們查看一下advanceRunState 方法的源碼。
private void advanceRunState(int targetState) { for (;;) {//下面的方法執行的就是：//首先獲取線程的ctl值，然后判斷當前線程池的狀態如果已經是SHUTDOWN，那么if條件第一個為真就直接返回//如果不是SHUTDOWN狀態，就需要CAS的設置當前狀態為SHUTDOWNint c = ctl.get();if (runStateAtLeast(c, targetState) || //private static int ctlOf(int rs, int wc) { return rs | wc; } ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c)))) break; }}
我們可以看出advanceRunState 方法實際上就是判斷當前線程池的狀態是否為SHUTDWON，如果是那么就返回，否則就需要設置當前狀態為SHUTDOWN。
我們再來看看shutdown方法中調用線程中斷的方法interruptIdleWorkers源碼
private void interruptIdleWorkers() { interruptIdleWorkers(false);}private void interruptIdleWorkers(boolean onlyOne) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try {for (Worker w : workers) { Thread t = w.thread; //如果工作線程沒有被中斷，并且沒有正在運行設置中斷標志 if (!t.isInterrupted() && w.tryLock()) {try { //需要中斷當前線程 t.interrupt();} catch (SecurityException ignore) {} finally { w.unlock();} } if (onlyOne)break;} } finally {mainLock.unlock(); }}
上面的代碼中，需要設置所有空閑線程的中斷標志。首先獲取線程池的全局鎖，同時只有一個線程可以調用shutdown方法設置中斷標志。然后嘗試獲取工作線程Worker自己的鎖，獲取成功則可以設置中斷標志（這是由于正在執行任務的線程需要獲取自己的鎖，并且不可重入，所以正在執行的任務沒有被中斷），這里要中斷的那些線程是阻塞到getTask()方法并嘗試從任務隊列中獲取任務的線程即空閑線程。
5.2、shutdownNow操作在使用線程池的時候，如果我們調用了shutdownNow方法，線程池不僅不會再接受新的任務，還會將任務隊列中的任務丟棄，正在執行的任務也會被中斷，然后立刻返回該方法，不會等待激活的任務完成，返回值為當前任務隊列中被丟棄的任務列表
public List<Runnable> shutdownNow() { List<Runnable> tasks; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try {checkShutdownAccess(); //還是進行權限檢查advanceRunState(STOP); //設置線程池狀態臺STOPinterruptWorkers(); //中斷所有線程tasks = drainQueue(); //將任務隊列中的任務移動到task中 } finally {mainLock.unlock(); } tryTerminate(); return tasks; //返回tasks}
從上面的代碼中，我們可以可以發現，shutdownNow方法也是首先需要檢查調用該方法的線程的權限，之后不同于shutdown方法之處在于需要即刻設置當前線程池狀態為STOP，然后中斷所有線程（空閑線程+正在執行任務的線程），移除任務隊列中的任務
private void interruptWorkers() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try {for (Worker w : workers) //不需要判斷當前線程是否在執行任務（即不需要調用w.tryLock方法），中斷所有線程 w.interruptIfStarted(); } finally {mainLock.unlock(); }}5.3、awaitTermination操作
當線程調用該方法之后，會阻塞調用者線程，直到線程池狀態為TERMINATED狀態才會返回，或者等到超時時間到之后會返回，下面是該方法的源碼。
//調用該方法之后，會阻塞調用者線程，直到線程池狀態為TERMINATED狀態才會返回，或者等到超時時間到之后會返回public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { long nanos = unit.toNanos(timeout); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try {//阻塞當前線程，（獲取了Worker自己的鎖），那么當前線程就不會再執行任務（因為獲取不到鎖）for (;;) { //當前線程池狀態為TERMINATED狀態，會返回true if (runStateAtLeast(ctl.get(), TERMINATED))return true; //超時時間到返回false if (nanos <= 0)return false; nanos = termination.awaitNanos(nanos);} } finally {mainLock.unlock(); }}
在上面的代碼中，調用者線程需要首先獲取線程Worker自己的獨占鎖，然后在循環判斷當前線程池是否已經是TERMINATED狀態，如果是則直接返回，否則說明當前線程池中還有線程正在執行任務，這時候需要查看當前設置的超時時間是否小于0，小于0說明不需要等待就直接返回，如果大于0就需要調用條件變量termination的awaitNanos方法等待設置的時間，并在這段時間之內等待線程池的狀態變為TERMINATED。
我們在前面說到清理線程池的方法processWorkerExit的時候，需要調用tryTerminated方法，在該方法中會查看當前線程池狀態是否為TERMINATED，如果是該狀態也會調用termination.signalAll()方法喚醒所有線程池中因調用awaitTermination而被阻塞住的線程。
如果是設置了超時時間，那么termination的awaitNanos方法也會返回，這時候需要重新檢查線程池狀態是否為TERMINATED，如果是則返回，不是就繼續阻塞自己。
以上就是Java并發包中線程池ThreadPoolExecutor原理探究的詳細內容，更多關于Java 并發包 線程池 ThreadPoolExecutor的資料請關注好吧啦網其它相關文章！
相關文章：
網公網安備