Java執行緒池ThreadPoolExecutor實現原理剖析
【編者的話】在Java中,使用執行緒池來非同步執行一些耗時任務是非常常見的操作。最初我們一般都是直接使用new Thread().start的方式,但我們知道,執行緒的建立和銷燬都會耗費大量的資源,關於執行緒可以參考之前的一篇部落格《 ofollow,noindex" target="_blank">Java執行緒那點事兒 》,因此我們需要重用執行緒資源。
當然也有其他待解決方案,比如說coroutine,目前Kotlin已經支援了,JDK也已經有了相關的提案: Project Loom ,目前的實現方式和Kotlin有點類似,都是基於ForkJoinPool,當然目前還有很多限制以及問題沒解決,比如synchronized還是鎖住當前執行緒等。
繼承結構
繼承結構看起來很清晰,最頂層的Executor只提供了一個最簡單的void execute(Runnable command)方法,然後是ExecutorService,ExecutorService提供了一些管理相關的方法,例如關閉、判斷當前執行緒池的狀態等,另外不同於Executor#execute,ExecutorService提供了一系列方法,可以將任務包裝成一個Future,從而使得任務提交方可以跟蹤任務的狀態。而父類AbstractExecutorService則提供了一些預設的實現。
構造器
ThreadPoolExecutor的構造器提供了非常多的引數,每一個引數都非常的重要,一不小心就容易踩坑,因此設定的時候,你必須要知道自己在幹什麼。
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; }
- corePoolSize、 maximumPoolSize。執行緒池會自動根據corePoolSize和maximumPoolSize去調整當前執行緒池的大小。當你通過submit或者execute方法提交任務的時候,如果當前執行緒池的執行緒數小於corePoolSize,那麼執行緒池就會建立一個新的執行緒處理任務, 即使其他的core執行緒是空閒的。如果當前執行緒數大於corePoolSize並且小於maximumPoolSize,那麼只有在佇列"滿"的時候才會建立新的執行緒。因此這裡會有很多的坑,比如你的core和max執行緒數設定的不一樣,希望請求積壓在佇列的時候能夠實時的擴容,但如果制定了一個無界佇列,那麼就不會擴容了,因為佇列不存在滿的概念。
- keepAliveTime。如果當前執行緒池中的執行緒數超過了corePoolSize,那麼如果在keepAliveTime時間內都沒有新的任務需要處理,那麼超過corePoolSize的這部分執行緒就會被銷燬。預設情況下是不會回收core執行緒的,可以通過設定allowCoreThreadTimeOut改變這一行為。
- workQueue。即實際用於儲存任務的佇列,這個可以說是最核心的一個引數了,直接決定了執行緒池的行為,比如說傳入一個有界佇列,那麼佇列滿的時候,執行緒池就會根據core和max引數的設定情況決定是否需要擴容,如果傳入了一個SynchronousQueue,這個佇列只有在另一個執行緒在同步remove的時候才可以put成功,對應到執行緒池中,簡單來說就是如果有執行緒池任務處理完了,呼叫poll或者take方法獲取新的任務的時候,新提交的任務才會put成功,否則如果當前的執行緒都在忙著處理任務,那麼就會put失敗,也就會走擴容的邏輯,如果傳入了一個DelayedWorkQueue,顧名思義,任務就會根據過期時間來決定什麼時候彈出,即為ScheduledThreadPoolExecutor的機制。
- threadFactory。建立執行緒都是通過ThreadFactory來實現的,如果沒指定的話,預設會使用Executors.defaultThreadFactory(),一般來說,我們會在這裡對執行緒設定名稱、異常處理器等。
- handler。即當任務提交失敗的時候,會呼叫這個處理器,ThreadPoolExecutor內建了多個實現,比如拋異常、直接拋棄等。這裡也需要根據業務場景進行設定,比如說當佇列積壓的時候,針對性的對執行緒池擴容或者傳送告警等策略。
看完這幾個引數的含義,我們看一下Executors提供的一些工具方法,只要是為了方便使用,但是我建議最好少用這個類,而是直接用ThreadPoolExecutor的建構函式,多瞭解一下這幾個引數到底是什麼意思,自己的業務場景是什麼樣的,比如執行緒池需不需要擴容、用不用回收空閒的執行緒等。
public class Executors { /* * 提供一個固定大小的執行緒池,並且執行緒不會回收,由於傳入的是一個無界佇列,相當於佇列永遠不會滿 * 也就不會擴容,因此需要特別注意任務積壓在佇列中導致記憶體爆掉的問題 */ public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); } /* *這個執行緒池會一直擴容,由於SynchronousQueue的特性,如果當前所有的執行緒都在處理任務,那麼 *新的請求過來,就會導致建立一個新的執行緒處理任務。如果執行緒一分鐘沒有新任務處理,就會被回 *收掉。特別注意,如果每一個任務都比較耗時,併發又比較高,那麼可能每次任務過來都會建立一個線 *程 */ public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); } }
原始碼分析
既然是個執行緒池,那就必然有其生命週期:執行中、關閉、停止等。ThreadPoolExecutor是用一個AtomicInteger去的前三位表示這個狀態的,另外又重用了低29位用於表示執行緒數,可以支援最大大概5億多,絕逼夠用了,如果以後硬體真的發展到能夠啟動這麼多執行緒,改成AtomicLong就可以了。
狀態這裡主要分為下面幾種:
- RUNNING:表示當前執行緒池正在執行中,可以接受新任務以及處理佇列中的任務
- SHUTDOWN:不再接受新的任務,但會繼續處理佇列中的任務
- STOP:不再接受新的任務,也不處理佇列中的任務了,並且會中斷正在進行中的任務
- TIDYING:所有任務都已經處理完畢,執行緒數為0,轉為為TIDYING狀態之後,會呼叫terminated()回撥
- TERMINATED:terminated()已經執行完畢
同時我們可以看到所有的狀態都是用二進位制位表示的,並且依次遞增,從而方便進行比較,比如想獲取當前狀態是否至少為SHUTDOWN等,同時狀態之前有幾種轉換:
- RUNNING -> SHUTDOWN。呼叫了shutdown()之後,或者執行了finalize()
- (RUNNING 或者 SHUTDOWN) -> STOP。呼叫了shutdownNow()之後會轉換這個狀態
- SHUTDOWN -> TIDYING。當執行緒池和佇列都為空的時候
- STOP -> TIDYING。當執行緒池為空的時候
- IDYING -> TERMINATED。執行完terminated()回撥之後會轉換為這個狀態
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; 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; //由於前三位表示狀態,因此將CAPACITY取反,和進行與操作即可 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; } private static boolean runStateLessThan(int c, int s) { return c < s; } private static boolean runStateAtLeast(int c, int s) { return c >= s; } private static boolean isRunning(int c) { return c < SHUTDOWN; } //下面三個方法,通過CAS修改worker的數目 private boolean compareAndIncrementWorkerCount(int expect) { return ctl.compareAndSet(expect, expect + 1); } //只嘗試一次,失敗了則返回,是否重試由呼叫方決定 private boolean compareAndDecrementWorkerCount(int expect) { return ctl.compareAndSet(expect, expect - 1); } //跟上一個不一樣,會一直重試 private void decrementWorkerCount() { do {} while (! compareAndDecrementWorkerCount(ctl.get())); }
下面是比較核心的欄位,這裡workers採用的是非執行緒安全的HashSet,而不是執行緒安全的版本,主要是因為這裡有些複合的操作,比如說將worker新增到workers後,我們還需要判斷是否需要更新largestPoolSize等,workers只在獲取到mainLock的情況下才會進行讀寫,另外這裡的mainLock也用於在中斷執行緒的時候序列執行,否則如果不加鎖的話,可能會造成併發去中斷執行緒,引起不必要的中斷風暴。
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;
核心方法
拿到一個執行緒池之後,我們就可以開始提交任務,讓它去執行了,那麼我們看一下submit方法是如何實現的。
public Future<?> submit(Runnable task) { if (task == null) throw new NullPointerException(); RunnableFuture<Void> ftask = newTaskFor(task, null); execute(ftask); return ftask; } public <T> Future<T> submit(Callable<T> task) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task); execute(ftask); return ftask; }
這兩個方法都很簡單,首先將提交過來的任務(有兩種形式:Callable、Runnable)都包裝成統一的RunnableFuture,然後呼叫execute方法,execute可以說是執行緒池最核心的一個方法。
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); /* 獲取當前worker的數目,如果小於corePoolSize那麼就擴容, 這裡不會判斷是否已經有core執行緒,而是隻要小於corePoolSize就會直接增加worker */ if (workerCountOf(c) < corePoolSize) { /* 呼叫addWorker(Runnable firstTask, boolean core)方法擴容 firstTask表示為該worker啟動之後要執行的第一個任務,core表示要增加的為core執行緒 */ if (addWorker(command, true)) return; //如果增加失敗了那麼重新獲取ctl的快照,比如可能執行緒池在這期間關閉了 c = ctl.get(); } /* 如果當前執行緒池正在執行中,並且將任務丟到佇列中成功了, 那麼就會進行一次double check,看下在這期間執行緒池是否關閉了, 如果關閉了,比如處於SHUTDOWN狀態,如上文所講的,SHUTDOWN狀態的時候, 不再接受新任務,remove成功後呼叫拒絕處理器。而如果仍然處於執行中的狀態, 那麼這裡就double check下當前的worker數,如果為0,有可能在上述邏輯的執行 過程中,有worker銷燬了,比如說任務丟擲了未捕獲異常等,那麼就會進行一次擴容, 但不同於擴容core執行緒,這裡由於任務已經丟到佇列中去了,因此就不需要再傳遞firstTask了, 同時要注意,這裡擴容的是非core執行緒 */ if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false); } else if (!addWorker(command, false)) /* 如果在上一步中,將任務丟到佇列中失敗了,那麼就進行一次擴容, 這裡會將任務傳遞到firstTask引數中,並且擴容的是非core執行緒, 如果擴容失敗了,那麼就執行拒絕策略。 */ reject(command); }
這裡要特別注意下防止佇列失敗的邏輯,不同的佇列丟任務的邏輯也不一樣,例如說無界佇列,那麼就永遠不會put失敗,也就是說擴容也永遠不會執行,如果是有界佇列,那麼當佇列滿的時候,會擴容非core執行緒,如果是SynchronousQueue,這個佇列比較特殊,當有另外一個執行緒正在同步獲取任務的時候,你才能put成功,因此如果當前執行緒池中所有的worker都忙著處理任務的時候,那麼後續的每次新任務都會導致擴容,當然如果worker沒有任務處理了,阻塞在獲取任務這一步的時候,新任務的提交就會直接丟到佇列中去,而不會擴容。
上文中多次提到了擴容,那麼我們下面看一下執行緒池具體是如何進行擴容的:
private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { int c = ctl.get(); //獲取當前執行緒池的狀態 int rs = runStateOf(c); /* 如果狀態為大於SHUTDOWN, 比如說STOP,STOP上文說過佇列中的任務不處理了,也不接受新任務, 因此可以直接返回false不擴容了,如果狀態為SHUTDOWN並且firstTask為null,同時佇列非空, 那麼就可以擴容 */ if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { int wc = workerCountOf(c); /* 若worker的數目大於CAPACITY則直接返回, 然後根據要擴容的是core執行緒還是非core執行緒,進行判斷worker數目 是否超過設定的值,超過則返回 */ if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; /* 通過CAS的方式自增worker的數目,成功了則直接跳出迴圈 */ if (compareAndIncrementWorkerCount(c)) break retry; //重新讀取狀態變數,如果狀態改變了,比如執行緒池關閉了,那麼就跳到最外層的for迴圈, //注意這裡跳出的是retry。 c = ctl.get();// Re-read ctl if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } } boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { //建立Worker w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { /* 獲取鎖,並判斷執行緒池是否已經關閉 */ int rs = runStateOf(ctl.get()); if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // 若執行緒已經啟動了,比如說已經呼叫了start()方法,那麼就拋異常, throw new IllegalThreadStateException(); //新增到workers中 workers.add(w); int s = workers.size(); if (s > largestPoolSize) //更新largestPoolSize largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { //若Worker建立成功,則啟動執行緒,這麼時候worker就會開始執行任務了 t.start(); workerStarted = true; } } } finally { if (! workerStarted) //新增失敗 addWorkerFailed(w); } return workerStarted; } private void addWorkerFailed(Worker w) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { if (w != null) workers.remove(w); decrementWorkerCount(); //每次減少worker或者從佇列中移除任務的時候都需要呼叫這個方法 tryTerminate(); } finally { mainLock.unlock(); } }
這裡有個貌似不太起眼的方法tryTerminate,這個方法會在所有可能導致執行緒池終結的地方呼叫,比如說減少worker的數目等,如果滿足條件的話,那麼將執行緒池轉換為TERMINATED狀態。另外這個方法沒有用private修飾,因為ScheduledThreadPoolExecutor繼承自ThreadPoolExecutor,而ScheduledThreadPoolExecutor也會呼叫這個方法。
final void tryTerminate() { for (;;) { int c = ctl.get(); /* 如果當前執行緒處於執行中、TIDYING、TERMINATED狀態則直接返回,執行中的沒 什麼好說的,後面兩種狀態可以說執行緒池已經正在終結了,另外如果處於SHUTDOWN狀態, 並且workQueue非空,表明還有任務需要處理,也直接返回 */ if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) return; //可以退出,但是執行緒數非0,那麼就中斷一個執行緒,從而使得關閉的訊號能夠傳遞下去, //中斷worker後,worker捕獲異常後,會嘗試退出,並在這裡繼續執行tryTerminate()方法, //從而使得訊號傳遞下去 if (workerCountOf(c) != 0) { interruptIdleWorkers(ONLY_ONE); return; } final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //嘗試轉換成TIDYING狀態,執行完terminated回撥之後 //會轉換為TERMINATED狀態,這個時候執行緒池已經完整關閉了, //通過signalAll方法,喚醒所有阻塞在awaitTermination上的執行緒 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 } } /** * 中斷空閒的執行緒 * @param onlyOne */ private void interruptIdleWorkers(boolean onlyOne) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) { //遍歷所有worker,若之前沒有被中斷過, //並且獲取鎖成功,那麼就嘗試中斷。 //鎖能夠獲取成功,那麼表明當前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(); } }
Worker
下面看一下Worker類,也就是這個類實際負責執行任務,Worker類繼承自AbstractQueuedSynchronizer,AQS可以理解為一個同步框架,提供了一些通用的機制,利用模板方法模式,讓你能夠原子的管理同步狀態、blocking和unblocking執行緒、以及佇列,具體的內容之後有時間會再寫,還是比較複雜的。這裡Worker對AQS的使用相對比較簡單,使用了狀態變數state表示是否獲得鎖,0表示解鎖、1表示已獲得鎖,同時通過exclusiveOwnerThread儲存當前持有鎖的執行緒。另外再簡單提一下,比如說CountDownLatch, 也是基於AQS框架實現的,countdown方法遞減state,await阻塞等待state為0。
private final class Worker extends AbstractQueuedSynchronizer implements Runnable { /** Thread this worker is running in.Null if factory fails. */ final Thread thread; /** Initial task to run.Possibly null. */ Runnable firstTask; /** Per-thread task counter */ volatile long completedTasks; Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } /** Delegates main run loop to outer runWorker*/ public void run() { runWorker(this); } 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) { 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) { } } } }
注意這裡Worker初始化的時候,會通過setState(-1)將state設定為-1,並在runWorker()方法中置為0,上文說過Worker是利用state這個變數來表示鎖的狀態,那麼加鎖的操作就是通過CAS將state從0改成1,那麼初始化的時候改成-1,也就是表示在Worker啟動之前,都不允許加鎖操作,我們再看interruptIfStarted()以及interruptIdleWorkers()方法,這兩個方法在嘗試中斷Worker之前,都會先加鎖或者判斷state是否大於0,因此這裡的將state設定為-1,就是為了禁止中斷操作,並在runWorker中置為0,也就是說只能在Worker啟動之後才能夠中斷Worker。
另外執行緒啟動之後,其實就是呼叫了runWorker方法,下面我們看一下具體是如何實現的。
final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // 呼叫unlock()方法,將state置為0,表示其他操作可以獲得鎖或者中斷worker boolean completedAbruptly = true; try { /* 首先嚐試執行firstTask,若沒有的話,則呼叫getTask()從佇列中獲取任務 */ while (task != null || (task = getTask()) != null) { w.lock(); /* 如果執行緒池正在關閉,那麼中斷執行緒。 */ if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { //執行beforeExecute回撥 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回撥 afterExecute(task, thrown); } } finally { task = null; //這裡加了鎖,因此沒有執行緒安全的問題,volatile修飾保證其他執行緒的可見性 w.completedTasks++; w.unlock();//解鎖 } } completedAbruptly = false; } finally { //拋異常了,或者當前佇列中已沒有任務需要處理等 processWorkerExit(w, completedAbruptly); } } private void processWorkerExit(Worker w, boolean completedAbruptly) { //如果是異常終止的,那麼減少worker的數目 if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted decrementWorkerCount(); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //將當前worker中workers中刪除掉,並累加當前worker已執行的任務到completedTaskCount中 completedTaskCount += w.completedTasks; workers.remove(w); } finally { mainLock.unlock(); } //上文說過,減少worker的操作都需要呼叫這個方法 tryTerminate(); /* 如果當前執行緒池仍然是執行中的狀態,那麼就看一下是否需要新增另外一個worker替換此worker */ int c = ctl.get(); if (runStateLessThan(c, STOP)) { /* 如果是異常結束的則直接擴容,否則的話則為正常退出,比如當前佇列中已經沒有任務需要處理, 如果允許core執行緒超時的話,那麼看一下當前佇列是否為空,空的話則不用擴容。否則話看一下 是否少於corePoolSize個worker在執行。 */ 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); } } private Runnable getTask() { boolean timedOut = false; // 上一次poll()是否超時了 for (;;) { int c = ctl.get(); int rs = runStateOf(c); // 若執行緒池關閉了(狀態大於STOP) // 或者執行緒池處於SHUTDOWN狀態,但是佇列為空,那麼返回null if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); /* 如果允許core執行緒超時 或者 不允許core執行緒超時但當前worker的數目大於core執行緒數, 那麼下面的poll()則超時呼叫 */ boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; /* 獲取任務超時了並且(當前執行緒池中還有不止一個worker 或者 佇列中已經沒有任務了),那麼就嘗試 減少worker的數目,若失敗了則重試 */ if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { //從佇列中抓取任務 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; //走到這裡表明,poll呼叫超時了 timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }
關閉執行緒池
關閉執行緒池一般有兩種形式,shutdown()和shutdownNow()。
public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); //通過CAS將狀態更改為SHUTDOWN,這個時候執行緒池不接受新任務,但會繼續處理佇列中的任務 advanceRunState(SHUTDOWN); //中斷所有空閒的worker,也就是說除了正在處理任務的worker,其他阻塞在getTask()上的worker //都會被中斷 interruptIdleWorkers(); //執行回撥 onShutdown(); // hook for ScheduledThreadPoolExecutor } finally { mainLock.unlock(); } tryTerminate(); //這個方法不會等待所有的任務處理完成才返回 } public List<Runnable> shutdownNow() { List<Runnable> tasks; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); /* 不同於shutdown(),會轉換為STOP狀態,不再處理新任務,佇列中的任務也不處理, 而且會中斷所有的worker,而不只是空閒的worker */ advanceRunState(STOP); interruptWorkers(); tasks = drainQueue();//將所有的任務從佇列中彈出 } finally { mainLock.unlock(); } tryTerminate(); return tasks; } private List<Runnable> drainQueue() { BlockingQueue<Runnable> q = workQueue; ArrayList<Runnable> taskList = new ArrayList<Runnable>(); /* 將佇列中所有的任務remove掉,並新增到taskList中, 但是有些佇列比較特殊,比如說DelayQueue,如果第一個任務還沒到過期時間,則不會彈出, 因此這裡通過呼叫toArray方法,然後再一個一個的remove掉 */ q.drainTo(taskList); if (!q.isEmpty()) { for (Runnable r : q.toArray(new Runnable[0])) { if (q.remove(r)) taskList.add(r); } } return taskList; }
從上文中可以看到,呼叫了shutdown()方法後,不會等待所有的任務處理完畢才返回,因此需要呼叫awaitTermination()來實現。
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; //阻塞在訊號量上,等待執行緒池終結,但是要注意這個方法可能會因為一些未知原因隨時喚醒當前執行緒, //因此需要重試,在tryTerminate()方法中,執行完terminated()回撥後,表明執行緒池已經終結了, //然後會通過termination.signalAll()喚醒當前執行緒 nanos = termination.awaitNanos(nanos); } } finally { mainLock.unlock(); } } 一些統計相關的方法 public int getPoolSize() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //若執行緒已終結則直接返回0,否則計算works中的數目 //想一下為什麼不用workerCount呢? return runStateAtLeast(ctl.get(), TIDYING) ? 0 : workers.size(); } finally { mainLock.unlock(); } } public int getActiveCount() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { int n = 0; for (Worker w : workers) if (w.isLocked())//上鎖的表明worker當前正在處理任務,也就是活躍的worker ++n; return n; } finally { mainLock.unlock(); } } public int getLargestPoolSize() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { return largestPoolSize; } finally { mainLock.unlock(); } } //獲取任務的總數,這個方法慎用,若是個無解佇列,或者佇列擠壓比較嚴重,會很蛋疼 public long getTaskCount() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { long n = completedTaskCount;//比如有些worker被銷燬後,其處理完成的任務就會疊加到這裡 for (Worker w : workers) { n += w.completedTasks;//疊加歷史處理完成的任務 if (w.isLocked())//上鎖表明正在處理任務,也算一個 ++n; } return n + workQueue.size();//獲取佇列中的數目 } finally { mainLock.unlock(); } } public long getCompletedTaskCount() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { long n = completedTaskCount; for (Worker w : workers) n += w.completedTasks; return n; } finally { mainLock.unlock(); } }
總結
這篇部落格基本上覆蓋了執行緒池的方方面面,但仍然有非常多的細節可以深究,比如說異常的處理,可以參照之前的一篇部落格:《 深度解析Java執行緒池的異常處理機制 》,另外還有AQS、unsafe等可以之後再單獨總結。