­

深入淺出Java執行緒池：源碼篇

前言

在上一篇文章深入淺出Java執行緒池：理論篇中，已經介紹了什麼是執行緒池以及基本的使用。（本來寫作的思路是使用篇，但經網友建議後，感覺改為理論篇會更加合適）。本文則深入執行緒池的源碼，主要是介紹ThreadPoolExecutor內部的源碼是如何實現的，對ThreadPoolExecutor有一個更加清晰的認識。

ThreadPoolExecutor的源碼相對而言比較好理解，沒有特別難以讀懂的地方。相信沒有閱讀源碼習慣的讀者，跟著本文，也可以很輕鬆地讀懂ThreadPoolExecutor的核心源碼邏輯。

本文源碼jdk版本為8，該類版本為jdk1.5，也就是在1.5之後，ThreadPoolExecutor的源碼沒有做修改。

執行緒池家族

Java中的執行緒池繼承結構如下圖：(類圖中只寫了部分方法且省略參數)

  • 頂層介面Executor表示一個執行器，他只有一個介面：execute() ，表示可以執行任務
  • ExecutorService在Executor的基礎上拓展了更多的執行方法，如submit() shutdown() 等等,表示一個任務執行服務。
  • AbstarctExecutorService是一個抽象類，他實現了ExecutorService的部分核心方法，如submit等
  • ThreadPoolExecutor是最核心的類，也就是執行緒池，他繼承了抽象類AbstarctExecutorService
  • 此外還有ScheduledExecutorService介面，他表示一個可以按照指定時間或周期執行的執行器服務，內部定義了如schedule() 等方法來執行任務
  • ScheduledThreadPoolExecutor實現了ScheduledExecutorService介面，同時繼承於ThreadPoolExecutor，內部的執行緒池相關邏輯使用自ThreadPoolExecutor，在此基礎上拓展了延遲、周期執行等功能特性

ScheduledThreadPoolExecutor相對來說用的是比較少。延時任務在我們Android中有更加熟悉的方案：Handler；而周期任務則用的非常少。現在android的後台限制非常嚴格，基本上一退出應用，應用進程很容易被系統幹掉。當然ScheduledThreadPoolExecutor也不是完全沒有用處，例如桌面小部件需要設置定時刷新，那麼他就可以派上用場了。

因此，我們本文的源碼，主要針對ThreadPoolExecutor。在閱讀源碼之前，我們先來看一下ThreadPoolExecutor內部的結構以及關鍵角色。

內部結構

閱讀源碼前，我們先把ThreadPoolExecutor整個源碼結構講解一下，形成一個整體概念，再閱讀源碼就不會迷失在源碼中了。先來看一下ThreadPoolExecutor的內部結構：

yGOUkq.md.png

  • ThreadPoolExecutor內部有三個關鍵的角色：阻塞隊列、執行緒、以及RejectExecutionHandler（這裡寫個中文名純粹因為不知道怎麼翻譯這個名字），他們的作用在理論篇有詳細介紹，這裡不再贅述。
  • 在ThreadPoolExecutor中，一個執行緒對應一個worker對象，工人，非常形象。每個worker內部有一個獨立的執行緒，他會不斷去阻塞隊列獲取任務來執行，也就是調用阻塞隊列的 poll 或者 take 方法，他們區別後面會講。如果隊列沒有任務了，那麼就會阻塞在這裡。
  • workQueue，就是阻塞隊列，當核心執行緒已滿之後，任務就會被放置在這裡等待被工人worker領取執行
  • RejectExecutionHandler本身是一個介面，ThreadPoolExecutor內部有這樣的一個介面對象，當任務無法被執行會調用這個對象的方法。ThreadPoolExecutor提供了該介面的4種實現方案，我們可以直接拿來用，或者自己繼承介面，實現自定義邏輯。在構造執行緒池的時候可以傳入RejectExecutionHandler對象。
  • 整個ThreadPoolExecutor中最核心的方法就是execute，他會根據具體的情況來選擇不同的執行方案或者拒絕執行。

這樣，我們就清楚ThreadPoolExecutor的內部結構了，然後，我們開始 Read the fucking code 吧。

源碼分析

內部關鍵屬性

ThreadPoolExecutor內部有很多的變數，他們包含的資訊非常重要，先來了解一下。

ThreadPoolExecutor的狀態和執行緒數整合在同一個int變數中，類似於view測量中MeasureSpec。他的高三位表示執行緒池的狀態，低29位表示執行緒池中執行緒的數量,如下：

// AtomicInteger對象可以利用CAS實現執行緒安全的修改，其中包含了執行緒池狀態和執行緒數量資訊
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// COUNT_BITS=29,（對於int長度為32來說）表示執行緒數量的位元組位數
private static final int COUNT_BITS = Integer.SIZE - 3;
// 狀態掩碼，高三位是1，低29位全是0，可以通過 ctl&COUNT_MASK 運算來獲取執行緒池狀態
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;

執行緒池的狀態一共有5個：

  • 運行running：執行緒池創建之後即是運行狀態
  • 關閉shutdown：調用shutdown方法之後執行緒池處於shutdown狀態，該狀態會停止接收任何任務，阻塞隊列中的任務執行完成之後會自動終止執行緒池
  • 停止stop：調用shutdownNow方法之後執行緒池處於stop狀態。和shutdown的區別是這個狀態下的執行緒池不會去執行隊列中剩下的任務
  • 整理tidying：在執行緒池stop之後，進入tidying狀態，然後執行 terminated() 方法，再進入terminated狀態
  • 終止terminated：執行緒池中沒有任何執行緒在執行任務，執行緒池完全終止。

在源碼中這幾個狀態分別對應：

// 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;

上面的位操作不夠直觀，轉化後如下：

private static final int RUNNING    = 111 00000 00000000 00000000 00000000;
private static final int SHUTDOWN   = 000 00000 00000000 00000000 00000000; 
private static final int STOP       = 001 00000 00000000 00000000 00000000;
private static final int TIDYING    = 010 00000 00000000 00000000 00000000;
private static final int TERMINATED = 011 00000 00000000 00000000 00000000;

可以看到除了running是負數，其他的狀態都是正數，且狀態越靠後，數值越大。因此我們可以通過判斷 ctl&COUNT_MASK > SHUTDOWN 來判斷狀態是否處於 stop、tidying、terminated之一。後續源碼中會有很多的這樣的判斷，舉其中的一個方法：

// 這裡來判斷執行緒池的狀態
if(runStateAtLeast(ctl,SHUTDOWN)) {
    ...
}
// 這裡執行邏輯，直接判斷兩個數的大小
private static boolean runStateAtLeast(int c, int s) {
    return c >= s;
}

ps：這裡怎麼沒有使用掩碼COUNT_MASK ？因為狀態是處於高位，低位的數值不影響高位的大小判斷。當然如果要判斷相等，就還是需要使用掩碼COUNT_MASK的。

接下來是ThreadPoolExecutor內部的三個關鍵角色對象：

// 阻塞隊列
private final BlockingQueue<Runnable> workQueue;
// 存儲worker的hashSet,worker被創建之後會被存儲到這裡
private final HashSet<Worker> workers = new HashSet<>();
// RejectedExecutionHandler默認的實現是AbortPolicy
private volatile RejectedExecutionHandler handler;
private static final RejectedExecutionHandler defaultHandler =
        new AbortPolicy();

內部使用的鎖對象：

// 這裡是兩個鎖。ThreadPoolExecutor內部並沒有使用Synchronize關鍵字來保持同步
// 而是使用Lock；和Synchronize的區別就是他是應用層的鎖，而synchronize是jvm層的鎖
private final ReentrantLock mainLock = new ReentrantLock();
private final Condition termination = mainLock.newCondition();

最後是內部一些參數的配置，前面都介紹過，把源碼貼出來再回顧一下：

// 執行緒池歷史達到的最大執行緒數
private int largestPoolSize;
// 執行緒池完成的任務數。
// 該數並不是實時更新的，在獲取執行緒池完成的任務數時，需要去統計每個worker完成的任務並累加起來
// 當一個worker被銷毀之後，他的任務數就會被累加到這個數據中
private long completedTaskCount;
// 執行緒工廠，用於創建執行緒
private volatile ThreadFactory threadFactory;
// 空閑執行緒存儲的時間
private volatile long keepAliveTime;
// 是否允許核心執行緒被回收
private volatile boolean allowCoreThreadTimeOut;
// 核心執行緒數限額
private volatile int corePoolSize;
// 執行緒總數限額
private volatile int maximumPoolSize;

不是吧sir？源碼還沒看到魂呢，整出來這麼無聊的變數？
咳咳，別急嘛，源碼解析馬上來。這些變數會貫穿整個源碼過程始終，先對他們有個印象，後續閱讀源碼就會輕鬆暢通很多。

關鍵方法：execute()

這個方法的主要任務就是根據執行緒池的當前狀態，選擇任務的執行策略。該方法的核心邏輯思路是：

  1. 在執行緒數沒有達到核心執行緒數時，會創建一個核心執行緒來執行任務

    public void execute(Runnable command) {
    // 不能傳入空任務
    if (command == null)
        throw new NullPointerException();

    // 獲取ctl變數，就是上面我們講的將狀態和執行緒數合在一起的一個變數
    int c = ctl.get();
    // 判斷核心執行緒數是否超過限額，否則創建一個核心執行緒來執行任務
    if (workerCountOf(c) < corePoolSize) {
        // addWorker方法是創建一個worker，也就是創建一個執行緒，參數true表示這是一個核心執行緒
        // 如果添加成功則直接返回
        // 否則意味著中間有其他的worker被添加了，導致超出核心執行緒數;或者執行緒池被關閉了等其他情況
        // 需要進入下一步繼續判斷
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    ...
}

  2. 當執行緒數達到核心執行緒數時，新任務會被放入到等待隊列中等待被執行

  3. 當等待隊列已經滿了之後，如果執行緒數沒有到達總的執行緒數上限，那麼會創建一個非核心執行緒來執行任務

  4. 當執行緒數已經到達總的執行緒數限制時，新的任務會被拒絕策略者處理，執行緒池無法執行該任務。

    public void execute(Runnable command) {
    ...
    // 如果執行緒池還在運行,則嘗試添加任務到隊列中
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 再次檢查如果執行緒池被關閉了，那麼把任務移出隊列
        // 如果移除成功則拒絕本次任務
        // 這裡主要是判斷在插入隊列的過程中，執行緒池有沒有被關閉了
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 否則再次檢查執行緒數是否為0，如果是，則創建一個沒有任務的非主執行緒worker
        // 這裡對應核心執行緒為0的情況，指定任務為null，worker會去隊列拿任務來執行
        // 這裡表示執行緒池至少有一個執行緒來執行隊列中的任務
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 如果上面添加到隊列中失敗，則嘗試創建一個非核心執行緒來執行任務
    // 如果創建失敗，則拒絕任務
    else if (!addWorker(command, false))
        reject(command);
}

源碼中還設計到兩個關鍵方法：addWorker創建一個新的worker，也就是創建一個執行緒；reject拒絕一個任務。後者比較簡單我們先看一下。

拒絕任務：reject()

// 拒絕任務，調用rejectedExecutionHandler來處理
final void reject(Runnable command) {
    handler.rejectedExecution(command, this);
}

默認的實現類有4個，我們依次來看一下：

  • AbortPolicy是默認實現，會拋出一個RejectedExecutionException異常：

    public static class AbortPolicy implements RejectedExecutionHandler {
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    }
}

  • DiscardPolicy最簡單，就是：什麼都不做，直接拋棄任務。（這是非常渣男不負責任的行為，咱們不能學他，所以也不要用它 [此處狗頭] ）

    public static class DiscardPolicy implements RejectedExecutionHandler {
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
}

  • DiscardOldestPolicy會刪除隊列頭的一個任務，然後再次執行自己（擠掉原位，自己上位，綠茶行為？）

    public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            e.getQueue().poll();
            e.execute(r);
        }
    }
}

  • CallerRunsPolicy最猛，他乾脆在自己的執行緒執行run方法，不依靠執行緒池了，自己動手豐衣足食。

    public static class CallerRunsPolicy implements RejectedExecutionHandler {
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

上面4個ThreadPoolExecutor已經幫我們實現了，他的靜態內部類，在創建ThreadPoolExecutor的時候我們可以直接拿來用。也可以自己繼承介面實現自己的邏輯。具體選擇哪個需要根據實際的業務需求來決定。

那麼接下來看創建worker的方法。

創建worker：addWorker()

方法的目的很簡單：創建一個worker。前面我們講到，worker內部創建了一個執行緒，每一個worker則代表了一個執行緒，非常類似android中的looper。looper的loop()方法會不斷地去MessageQueue獲取message，而Worker的run()方法會不斷地去阻塞隊列獲取任務，這個我們後面講。

addWorker() 方法的邏輯整體上分為兩個部分：

  1. 檢查執行緒狀態和執行緒數是否滿足條件：

    // 第一個參數是創建的執行緒首次要執行的任務，可以是null，則表示初始化一個執行緒
// 第二參數表示是否是一個核心執行緒
private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (int c = ctl.get();;) {
        // 還記不記得我們前面講到執行緒池的狀態控制？
        // runStateAtLeast(c, SHUTDOWN)表示狀態至少為shutdown，後面類同
        // 如果執行緒池處於stop及以上，不會再創建worker
        // 如果執行緒池狀態在shutdown時，如果隊列不為空或者任務!=null，則還會創建worker
        if (runStateAtLeast(c, SHUTDOWN)
            && (runStateAtLeast(c, STOP)
                || firstTask != null
                || workQueue.isEmpty()))
            // 其他情況返回false，表示拒絕創建worker
            return false;
        
		// 這裡採用CAS輪詢，也就是循環鎖的策略來讓執行緒總數+1
        for (;;) {
            // 檢查是否超出執行緒數限制
            // 這裡根據core參數判斷是核心執行緒還是非核心執行緒
            if (workerCountOf(c)
                >= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
                return false;
            // 利用CAS讓ctl變數自增，表示worker+1
            // 如果CAS失敗，則表示發生了競爭，則再來一次
            if (compareAndIncrementWorkerCount(c))
                // 成功則跳出最外層循環
                break retry;
            // 如果這個期間ctl被改變了，則獲取ctl，再嘗試一次
            c = ctl.get();  
            // 如果執行緒池被shutdown了，那麼重複最外層的循環，重新判斷狀態是否可以創建worker
            if (runStateAtLeast(c, SHUTDOWN))
                // 繼續最外層循環
                continue retry;
        }
    }
    
    // 創建worker邏輯
    ...
}

    不知道讀者對於源碼中的retry: 有沒有疑惑，畢竟平時很少用到。他的作用是標記一個循環，這樣我們在內層的循環就可以跳轉到任意一個外層的循環。這裡的retry只是一個名字，改成 repeat: 甚至 a: 都是可以的。他的本質就是：一個循環的標記 。

  2. 創建worker對象，並調用其內部執行緒的start()方法來啟動執行緒：

    private boolean addWorker(Runnable firstTask, boolean core) {
	...
    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 c = ctl.get();
				// 只有運行狀態或者shutDown&&task==null才會被執行
                if (isRunning(c) ||
                    (runStateLessThan(c, STOP) && firstTask == null)) {
                    // 如果這個執行緒不是剛創建的，則拋出異常
                    if (t.getState() != Thread.State.NEW)
                        throw new IllegalThreadStateException(); 
                    // 添加到workerSet中
                    workers.add(w);
                    workerAdded = true;
                    int s = workers.size();
                    // 跟蹤執行緒池到達的最多執行緒數量
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                }
            } finally {
                // 釋放鎖
                mainLock.unlock();
            }
            // 如果添加成功，啟動執行緒
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        // 如果執行緒沒有啟動，表示添加worker失敗，可能在添加的過程中執行緒池被關閉了
        if (! workerStarted)
            // 把worker從workerSet中移除
            addWorkerFailed(w);
    }
    return workerStarted;
}

經過前面兩步，如果沒有出現異常，則創建worker成功。最後還涉及到一個方法： addWorkerFailed(w) ，他的內容比較簡答，順便提一下吧：

// 添加worker失敗
private void addWorkerFailed(Worker w) {
    final ReentrantLock mainLock = this.mainLock;
    // 加鎖
    mainLock.lock();
    try {
        if (w != null)
            workers.remove(w);
        // 這裡會讓執行緒總數-1
        decrementWorkerCount();
        // 嘗試設置執行緒池的狀態為terminad
        // 因為添加失敗有可能是執行緒池在添加worker的過程中被shutdown
        // 那麼這個時候如果沒有任務正在執行就需要設置狀態為terminad
        // 這個方法後面會詳細講
        tryTerminate();
    } finally {
        mainLock.unlock();
    }
}

那麼到這裡，execute()方法中的一些調用方法就分析完了。阻塞隊列相關的方法不屬於本文的範疇，就不展開了。那麼還有一個問題：worker是如何工作的呢？worker內部有一個執行緒，當執行緒啟動時，初始化執行緒的runnable對象的run方法會被調用，那麼這個runnable對象是什麼？我直接來看worker。

打工人：Worker

首先我們看到他的構造方法：

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

源碼很簡單，把傳進來的任務設置給內部變數firstTask，然後把自己傳給執行緒工廠去創建一個執行緒。所以執行緒啟動時，Worker本身的run方法會被調用，那麼我們看到Worker的 run()方法。

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

Worker是ThreadPoolExecutor的內部類，這裡直接調用到了ThreadPoolExecutor的方法： runWorker()來開始執行。那麼接下來，我們就看到這個方法。

啟動worker：runWorker()

這個方法是worker執行的方法，在執行緒被銷毀前他會一直執行，類似於Handler的looper，不斷去隊列獲取消息來執行：

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    // 獲取worker初始化時設置的任務，可以為null。如果為null則表示僅僅創建執行緒
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); 
    // 這個參數的作用後面解釋，需要結合其他的源碼
    boolean completedAbruptly = true;
    try {
        // 如果自身的task不為null，那麼會執行自身的task
        // 否則調用getTask去隊列獲取一個task來執行
        // 這個getTask最終會去調用隊列的方法來獲取任務
        // 而隊列如果為空他的獲取方法會進行阻塞，這裡也就阻塞了,後面深入講
        while (task != null || (task = getTask()) != null) {
            try{
            // 執行任務
            ...
            } finally {
                // 任務執行完成，把task設置為null
                task = null;
                // 任務總數+1
                w.completedTasks++;
                // 釋放鎖
                w.unlock();
            }
        }
    	// 這裡設置為false，先記住他
        completedAbruptly = false;
    } finally {
    	// 如果worker退出，那麼需要執行後續的善後工作
        processWorkerExit(w, completedAbruptly);
    }
}

可以看到這個方法的整體框架還是比較簡單的，核心就在於 while (task != null || (task = getTask()) != null) 這個循環中，如果 getTask() 返回null，則表示執行緒該結束了，這和Handler機制也是一樣的。

上面的源碼省略了具體執行任務的邏輯，他的邏輯也是很簡單：判斷狀態+運行任務。我們來看一下：

final void runWorker(Worker w) {
    ...;
    try {
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // 如果執行緒池已經設置為stop狀態，那麼保證執行緒是interrupted標誌
            // 如果執行緒池沒有在stop狀態，那麼保證執行緒不是interrupted標誌
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                // 回調方法，這個方法是一個空實現
                beforeExecute(wt, task);
                try {
                    // 運行任務
                    task.run();
                    // 回調方法，也是一個空實現
                    afterExecute(task, null);
                } catch (Throwable ex) {
                    afterExecute(task, ex);
                    throw ex;
                }
            }
            ...
        }
    	completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

在獲取到一個任務後，就會去執行該任務的run方法，然後再回去繼續獲取新的任務。

我們會發現其中有很多的空實現方法，他是給子類去實現的，有點類似於Activity的生命周期，子類需要重寫這些方法，在具體的情況做一些工作。當然，一般的使用是不需要去重寫這些方法。接下來需要來看看 getTask() 是如何獲取任務的。

獲取任務：getTask()

這個方法的內容可以分為兩個部分：判斷當前執行緒池的狀態+阻塞地從隊列中獲取一個任務。

第一部分是判斷當前執行緒池的狀況，如果處於關閉狀態那麼直接返回null來讓worker結束，否則需要判斷當前執行緒是否超時或者超出最大限制的執行緒數：

private Runnable getTask() {
    boolean timedOut = false; 
    // 內部使用了CAS，這裡需要有一個循環來不斷嘗試
    for (;;) {
        int c = ctl.get();
        // 如果處於shutdown狀態而且隊列為空，或者處於stop狀態，返回null
        // 這和前面我們討論到不同的執行緒池的狀態的不同行為一致
        if (runStateAtLeast(c, SHUTDOWN)
            && (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
            // 這裡表示讓執行緒總數-1，記住他，後面會繼續聊到
            decrementWorkerCount();
            return null;
        }
        
        // 獲取目前的執行緒總數
        int wc = workerCountOf(c);
        // 判斷該執行緒在空閑情況是否可以被銷毀：允許核心執行緒為null或者當前執行緒超出核心執行緒數
        // 可以看到這裡並沒有去區分具體的執行緒是核心還是非核心，只有執行緒數量處於核心範圍還是非核心範圍
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        // 超出最大執行緒數或者已經超時；
        // 這裡可能是用戶通過 setMaximumPoolSize 改動了數據才會導致這裡超出最大執行緒數
        // 同時還必須保證當前執行緒數量大於1或者隊列已經沒有任務了
        // 這樣就確保了當有任務存在時，一定至少有一個執行緒在執行任務
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            // 使用CAS嘗試讓當前執行緒總數-1，失敗則從來一次上面的邏輯
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
        
        // 獲取任務邏輯
        ...
    }
}

第二部分是獲取一個任務並執行。獲取任務使用的是阻塞隊列的方法，如果隊列中沒有任務，則會被阻塞：

private Runnable getTask() {
    boolean timedOut = false; 
    // 內部使用了CAS，這裡需要有一個循環來不斷嘗試
    for (;;) {
        // 判斷執行緒池狀態邏輯
        ...
        try {
            // 獲取一個任務
            // poll方法等待具體時間之後如果沒有獲取到對象，會返回null
            // take方法會一直等到獲取新對象，除非被interrupt
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            // r==null,說明超時了，重新循環
            timedOut = true;
        } catch (InterruptedException retry) {
            // 被interrupt，說明可能執行緒池被關閉了，重新判斷情況
            timedOut = false;
        }
    }
}

這裡需要重點關注的是阻塞隊列的 poll() 和 take() 方法，他們都會去隊列中獲取一個任務；但是，poll() 方法會阻塞指定時間後返回，而 take() 則是無限期阻塞。這裡對應的就是有存活時間的執行緒和不會被銷毀的核心執行緒。

同時注意 timedOut = true 是在這一部分被賦值的，當賦值為true之後需要再執行一次循環，在上面的判斷中就會被攔截下來並返回false，這在第一部分邏輯介紹了。而如果執行緒在等待的時候被 interrupt 了，說明執行緒池被關閉了，此時也會重走一次上面判斷狀態的邏輯。

到這裡關於執行的邏輯就講得差不多了，下面聊一聊執行緒池關閉以及worker結束的相關邏輯。

worker退出工作：processWorkerExit

前面已經介紹 runWorker() 了方法，這個方法的主要任務就是讓worker動起來，不斷去隊列獲取任務。而當獲取任務的時候返回了null，則表示該worker可以結束了，最後會調用 processWorkerExit() 方法，如下：

final void runWorker(Worker w) {
    ...
    try {
       ...
    } finally {
    	// 如果worker退出，那麼需要執行後續的善後工作
        processWorkerExit(w, completedAbruptly);
    }
}

processWorkerExit() 會完成worker退出的善後工作。具體的內容是：

  1. 把完成的任務數合併到總的任務數,移除worker，嘗試設置執行緒池的狀態為terminated：
private void processWorkerExit(Worker w, boolean completedAbruptly) {
    // 如果不是經過getTask方法返回null正常退出的，那麼需要讓執行緒總數-1
    // 這個參數前面一直讓你們注意一下不知道你們還記不記得
    // 如果是在正常情況下退出，那麼在getTask() 方法中就會執行decrementWorkerCount()了
    // 而如果出現一些特殊的情況突然結束了，並不是通過在getTask返回null結束
    // Abruptly就是突然的意思，那麼completedAbruptly就為true,正常情況下在runWorker方法中會被設置為false
    // 那什麼叫突然結束？用戶的任務拋出了異常，這個時候執行緒就突然結束了，沒有經過getTask方法
    // 這裡就需要讓執行緒總數-1
    if (completedAbruptly) 
        decrementWorkerCount();

    // 獲取鎖，並累加完成的任務總數，從set中移除worker
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        completedTaskCount += w.completedTasks;
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }

    // 嘗試設置執行緒池的狀態為terminated
    // 這個方法前面我們addWorker失敗的時候提到過，後面再展開
    tryTerminate();
	...
}
  1. 移除worker之後，如果執行緒池還沒有被stop，那麼最後必須保證隊列任務至少有一個執行緒在執行隊列中的任務：
private void processWorkerExit(Worker w, boolean completedAbruptly) {
    ...
	int c = ctl.get();
    // stop及以上的狀態不需要執行剩下的任務
    if (runStateLessThan(c, STOP)) {
        // 如果執行緒是突然終止的，那肯定需要重新創建一個
        // 否則進行判斷是否要保留一個執行緒
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; 
        }
        // 如果此時執行緒數<=核心執行緒數，或者當核心執行緒可被銷毀時，執行緒數==0且隊列不為空
        // 那麼需要創建一個執行緒來執行任務
        addWorker(null, false);
    }
}

程式碼雖然看起來很多，但是具體的邏輯內容還是比較簡單的。前面一直提到一個方法 tryTerminate() 但一直沒有展開解釋，下面來介紹一下。

嘗試終止執行緒池：tryTerminate()

這個方法出現在任何可能讓執行緒池進入終止狀態的地方。如添加worker失敗時，那麼這個時候可能執行緒池已經處於stop狀態，且已經沒有任何正在執行的worker了，那麼此時可以進入terminated狀態；再如worker被銷毀的時候，可能這是最後一個被銷毀的worker，那麼此時執行緒池需要進入terminated狀態。

根據這個方法的使用情況其實就已經差不多可以推斷出這個方法的內容：判斷當前執行緒池的狀態，如果符合條件則設置執行緒池的狀態為terminated 。如果此時不能轉換為terminated狀態，則什麼也不做，直接返回。

  1. 首先判斷當前執行緒池狀態是否符合轉化為terminated。如果處於運行狀態或者tidying以上狀態，則肯定不需要進行狀態轉換。因為running需要先進入stop狀態，而tidying其實已經是準備進入terminated狀態了。如果處於shutdown狀態且隊列不為空，那麼需要執行完隊列中的任務，所以也不適合狀態轉換：
final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        // 如果處於運行狀態或者tidying以上狀態時，直接返回，不需要修改狀態
        // 如果處於stop以下狀態且隊列不為空，那麼需要等隊列中的任務執行完成，直接返回
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateLessThan(c, STOP) && ! workQueue.isEmpty()))
            return;
        // 到這裡說明執行緒池肯定處於stop狀態
        // 執行緒的數量不等於0，嘗試中斷一個空閑的worker執行緒
        // 這裡他只中斷workerSet中的其中一個，當其中的一個執行緒停止時，會再次調用tryTerminate
        // 然後又會再去中斷workerSet中的一個worker，不斷循環下去直到剩下最後一個，workercount==0
        // 這就是 鏈式反應 。
        if (workerCountOf(c) != 0) { 
            interruptIdleWorkers(ONLY_ONE);
            return;
        }
		
        // 設置狀態為terminated邏輯
        ...
    }
}
  1. 經過上面的判斷，能到第二部分邏輯，執行緒池肯定是具備進入terminated狀態的條件了。剩下的程式碼就是把執行緒池的狀態設置為terminated：
final void tryTerminate() {
    for (;;) {
        // 上一部分邏輯
        ...
        // 首先獲取全局鎖
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // 嘗試把執行緒池的狀態從stop修改為tidying
            // 如果修改失敗，說明狀態已經被修改了，那麼外層循環再跑一個
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    // 這個方法是一個空實現，需要子類繼承重寫
                    terminated();
                } finally {
                    // 最後再設置狀態為terminated
                    ctl.set(ctlOf(TERMINATED, 0));
                    // 喚醒所有等待終止鎖的執行緒
                    termination.signalAll();
                }
                return;
            }
        } finally {
            // 釋放鎖
            mainLock.unlock();
        }
        // CAS修改執行緒池的狀態失敗，重新進行判斷
    }
}

當執行緒池被標記為terminated狀態時，那麼這個執行緒池就徹底地終止了。

好了到這裡，恭喜你，關於ThreadPoolExecutor的源碼解析理解得差不多了。接下來剩下幾個常用的api方法：submit() 、 shutdown()/shutdownNow() 順便看一下吧，他們的邏輯也是都非常簡單。

關閉執行緒池：shutdown/shutdownNow

關閉執行緒池有兩個方法：

  • shutdown：設置執行緒池的狀態為shutdown，同時嘗試中斷所有空閑執行緒，但是會等待隊列中的任務執行結束再終止執行緒池。
  • shutdownNow：設置執行緒池的狀態為stop，同時嘗試中斷所有空閑執行緒，不會等待隊列中的任務完成，正在執行中的執行緒執行結束，執行緒池馬上進入terminated狀態。

我們各自來看一下：

// 關閉後隊列中的任務依舊會被執行，但是不會再添加新的任務
public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        // 設置狀態為shutdown
        advanceRunState(SHUTDOWN);
        // 嘗試中斷所有空閑的worker
        interruptIdleWorkers();
        // 回調方法，這個方法是個空方法，ScheduledThreadPoolExecutor中重寫了該方法
        onShutdown(); 
    } finally {
        mainLock.unlock();
    }
    // 嘗試設置執行緒池狀態為terminated
    tryTerminate();
}

再看一下另一個方法shutdownNow：

// 關閉後隊列中剩餘的任務不會被執行
// 會把剩下的任務返回交給開發者去處理
public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 檢查是否可以關閉執行緒
        checkShutdownAccess();
        // 設置狀態為stop
        advanceRunState(STOP);
        // 嘗試中斷所有執行緒
        interruptWorkers();
        // 返回隊列中剩下的任務
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}

最後再來看一下和 execute()不同的提交任務方法：submit。

提交任務：submit()

submit方法並不是ThreadPoolExecutor實現的，而是AbstractExecutorService，如下：

// runnable沒有返回值，創建FutureTask的返回參數傳入null
public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}
// 有參數返回值的runnable
// 最終也是構造一個callable來執行，把返回值設置為result
public <T> Future<T> submit(Runnable task, T result) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task, result);
    execute(ftask);
    return ftask;
}
// callable本身就擁有返回值
public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}

他們的邏輯都幾乎一樣：調用newTaskFor方法來構造一個Future對象並返回。我們看到newTaskFor方法：

// 創建一個FutureTask來返回
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
    return new FutureTask<T>(runnable, value);
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    return new FutureTask<T>(callable);
}

可以看到這個方法很簡單：構造一個FutureTask並返回，FutureTask也是Future介面目前唯一的實現類。

更加具體關於Future的內容就不展開了，有興趣的讀者可以去了解一下。

最後

好了到這裡，關於ThreadPoolExecutor的源碼分析內容就講完了。最後讓我們再回顧一下吧：

yGOUkq.md.png

  • ThreadPoolExecutor的整個執行流程從execute方法開始，他會根據具體的情況，採用合適的執行方案
  • 執行緒被封裝在worker對象中，worker對象通過runWorker方法，會一直不斷地調用getTask方法來調用隊列的poll或take方法獲取任務
  • 當需要退出一個worker時，只要getTask方法返回null即可退出
  • 當執行緒池關閉時，會根據不同的關閉方法，等待所有的執行緒執行完成，然後關閉執行緒池。

執行緒池整體的模型和handler是十分類似的：一個生產者-消費者模型。但和Handler不同的是，ThreadPoolExecutor不支援延時任務，這點在ScheduledThreadPoolExecutor得到了實現；Handler的執行緒安全採用synchronize關鍵字，而ThreadPoolExecutor採用的是Lock和一些利用CAS實現執行緒安全的整型變數；Handler無法拒絕任務，執行緒池可以；Handler拋出異常會直接程式崩潰，而執行緒池不會等等。

了解了執行緒池的內部源碼，對於他更加了解後，那麼可以根據具體的問題，做出更加合適的解決方案。ThreadPoolExecutor還有一些源碼沒有講到，以及ScheduledThreadPoolExecutor、阻塞隊列的源碼，有興趣讀者可以自行去深入了解一下，拓展關於執行緒池的一切。

全文到此，假期肝文不容易啊，如果文章對你有幫助，求一個大拇指yJsExU.png，贊一下再走唄。

參考文獻

  • 《Java並發編程的藝術》：並發編程必讀，作者對一些原理講的很透徹
  • 《Java核心技術卷》：這系列的書主要是講解框架的使用，不會深入原理，適合入門
  • javaGuide：javaGuide，對java知識總結得很不錯的一個部落格
  • Java並發編程：執行緒池的使用：部落格園上一位很優秀的部落客，文章寫得通俗易懂且不失深度

全文到此，原創不易，覺得有幫助可以點贊收藏評論轉發。
筆者才疏學淺，有任何想法歡迎評論區交流指正。
如需轉載請評論區或私信交流。

另外歡迎光臨筆者的個人部落格：傳送門

Tags:

VirMach 便宜 VPS

QNews

QNews