一、總覽

執行緒池類ThreadPoolExecutor的相關類需要先了解：

 （圖片來自：https://javadoop.com/post/java-thread-pool#%E6%80%BB%E8%A7%88）

Executor：位於最頂層，只有一個 execute(Runnable runnable) 方法，用於提交任務。

ExecutorService ：在 Executor 介面的基礎上添加了很多的介面方法，提交任務，獲取結果，關閉執行緒池。

AbstractExecutorService：實現了ExecutorService 介面，然後在其基礎上實現了幾個實用的方法，這些方法提供給子類進行呼叫。

ThreadPoolExecutor：執行緒池類

Executors：最常用的用於生成 ThreadPoolExecutor 的例項的工具類

FutureTask：Runnable, Future -> RunnableFuture -> FutureTask

FutureTask 通過 RunnableFuture 間接實現了 Runnable 介面， 所以每個 Runnable 通常都先包裝成 FutureTask， 然後呼叫 executor.execute(Runnable command) 將其提交給執行緒池

Runnable 的 void run() 方法是沒有返回值的，如果我們需要的話，會在 submit 中指定第二個引數作為返回值。

Callable：Callable 也是因為執行緒池的需要，所以才有了這個介面。它和 Runnable 的區別在於 run() 沒有返回值，而 Callable 的 call() 方法有返回值

BlockingQueue：Java併發（十八）：阻塞佇列BlockingQueue

二、執行緒池狀態

執行緒池中的各個狀態：

• RUNNING：這個沒什麼好說的，這是最正常的狀態：接受新的任務，處理等待佇列中的任務
• SHUTDOWN：不接受新的任務提交，但是會繼續處理等待佇列中的任務
• STOP：不接受新的任務提交，不再處理等待佇列中的任務，中斷正在執行任務的執行緒
• TIDYING：所有的任務都銷燬了，workCount 為 0。執行緒池的狀態在轉換為 TIDYING 狀態時，會執行鉤子方法 terminated()
• TERMINATED：terminated() 方法結束後，執行緒池的狀態就會變成這個

狀態轉換：

• RUNNING -> SHUTDOWN：當呼叫了 shutdown() 後，會發生這個狀態轉換，這也是最重要的
• (RUNNING or SHUTDOWN) -> STOP：當呼叫 shutdownNow() 後，會發生這個狀態轉換，這下要清楚 shutDown() 和 shutDownNow() 的區別了
• SHUTDOWN -> TIDYING：當任務佇列和執行緒池都清空後，會由 SHUTDOWN 轉換為 TIDYING
• STOP -> TIDYING：當任務佇列清空後，發生這個轉換
• TIDYING -> TERMINATED：這個前面說了，當 terminated() 方法結束後

ThreadPoolExecutor採用一個 32 位的整數來存放執行緒池的狀態和當前池中的執行緒數，其中高 3 位用於存放執行緒池狀態，低 29 位表示執行緒數。

    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

    // 這裡 COUNT_BITS 設定為 29(32-3)，意味著前三位用於存放執行緒狀態，後29位用於存放執行緒數
    // 很多初學者很喜歡在自己的程式碼中寫很多 29 這種數字，或者某個特殊的字串，然後分佈在各個地方，這是非常糟糕的
    private static final int COUNT_BITS = Integer.SIZE - 3;

    // 000 11111111111111111111111111111
    // 這裡得到的是 29 個 1，也就是說執行緒池的最大執行緒數是 2^29-1=536870911
    // 以我們現在計算機的實際情況，這個數量還是夠用的
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

    // 我們說了，執行緒池的狀態存放在高 3 位中
    // 運算結果為 111跟29個0：111 00000000000000000000000000000
    private static final int RUNNING    = -1 << COUNT_BITS;
    // 000 00000000000000000000000000000
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    // 001 00000000000000000000000000000
    private static final int STOP       =  1 << COUNT_BITS;
    // 010 00000000000000000000000000000
    private static final int TIDYING    =  2 << COUNT_BITS;
    // 011 00000000000000000000000000000
    private static final int TERMINATED =  3 << COUNT_BITS;

    // 將整數 c 的低 29 位修改為 0，就得到了執行緒池的狀態
    private static int runStateOf(int c)     { return c & ~CAPACITY; }
    // 將整數 c 的高 3 為修改為 0，就得到了執行緒池中的執行緒數
    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;
    }

三、執行緒池引數

通過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.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

corePoolSize：執行緒池中核心執行緒的數量。當提交一個任務時，執行緒池會新建一個執行緒來執行任務，直到當前執行緒數等於corePoolSize。如果呼叫了執行緒池的prestartAllCoreThreads()方法，執行緒池會提前建立並啟動所有基本執行緒。

maximumPoolSize：執行緒池中允許的最大執行緒數。執行緒池的阻塞佇列滿了之後，如果還有任務提交，如果當前的執行緒數小於maximumPoolSize，則會新建執行緒來執行任務。注意，如果使用的是無界佇列，該引數也就沒有什麼效果了。

keepAliveTime：空閒執行緒的保活時間，如果某執行緒的空閒時間超過這個值都沒有任務給它做，那麼可以被關閉了。注意這個值並不會對所有執行緒起作用，如果執行緒池中的執行緒數少於等於核心執行緒數 corePoolSize，那麼這些執行緒不會因為空閒太長時間而被關閉，當然，也可以通過呼叫 allowCoreThreadTimeOut(true)使核心執行緒數內的執行緒也可以被回收

unit：keepAliveTime的單位。TimeUnit

workQueue：

用來儲存等待執行的任務的阻塞佇列，等待的任務必須實現Runnable介面。我們可以選擇如下幾種：

• ArrayBlockingQueue：基於陣列結構的有界阻塞佇列，FIFO。
  
• LinkedBlockingQueue：基於連結串列結構的有界阻塞佇列，FIFO。
• SynchronousQueue：不儲存元素的阻塞佇列，每個插入操作都必須等待一個移出操作，反之亦然。

threadFactory：用於設定建立執行緒的工廠。

handler：

RejectedExecutionHandler，執行緒池的拒絕策略。

所謂拒絕策略，是指將任務新增到執行緒池中時，執行緒池拒絕該任務所採取的相應策略。當向執行緒池中提交任務時，如果此時執行緒池中的執行緒已經飽和了，而且阻塞佇列也已經滿了，則執行緒池會選擇一種拒絕策略來處理該任務。

執行緒池提供了四種拒絕策略：（重寫RejectedExecutionHandler.rejectedExecution(Runnable, ThreadPoolExecutor)）

　　AbortPolicy：直接丟擲異常，預設策略；

　　CallerRunsPolicy：用呼叫者所在的執行緒來執行任務；

　　DiscardOldestPolicy：丟棄阻塞佇列中靠最前的任務，並執行當前任務；

　　DiscardPolicy：直接丟棄任務； 

當然我們也可以實現自己的拒絕策略，例如記錄日誌等等，實現RejectedExecutionHandler介面寫rejectedExecution方法即可。

四、執行緒池建立

Executor工具類提供了三種執行緒池建立方式：

FixedThreadPool ：可重用固定執行緒數的執行緒池

    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

corePoolSize 和 maximumPoolSize都設定為建立FixedThreadPool時指定的引數nThreads，意味著當執行緒池滿時且阻塞佇列也已經滿時，如果繼續提交任務，則會直接走拒絕策略，該執行緒池不會再新建執行緒來執行任務，而是直接走拒絕策略。FixedThreadPool使用的是預設的拒絕策略，即AbortPolicy，則直接丟擲異常。

keepAliveTime設定為0L，表示空閒的執行緒會立刻終止。

workQueue則是使用LinkedBlockingQueue，但是沒有設定範圍，那麼則是最大值（Integer.MAX_VALUE），這基本就相當於一個無界隊列了。使用該“無界佇列”則會帶來哪些影響呢？當執行緒池中的執行緒數量等於corePoolSize 時，如果繼續提交任務，該任務會被新增到阻塞佇列workQueue中，當阻塞佇列也滿了之後，則執行緒池會新建執行緒執行任務直到maximumPoolSize。由於FixedThreadPool使用的是“無界佇列”LinkedBlockingQueue，那麼maximumPoolSize引數無效，同時指定的拒絕策略AbortPolicy也將無效。而且該執行緒池也不會拒絕提交的任務，如果客戶端提交任務的速度快於任務的執行，那麼keepAliveTime也是一個無效引數。

SingleThreadExecutor：只有一個執行緒的固定執行緒池

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

為單一worker執行緒的執行緒池，SingleThreadExecutor把corePool和maximumPoolSize均被設定為1，和FixedThreadPool一樣使用的是無界佇列LinkedBlockingQueue,所以帶來的影響和FixedThreadPool一樣。

CachedThreadPool：根據需要建立新執行緒的執行緒池

  public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

CachedThreadPool的corePool為0，maximumPoolSize為Integer.MAX_VALUE，這就意味著所有的任務一提交就會加入到阻塞佇列中。

keepAliveTime這是為60L，unit設定為TimeUnit.SECONDS，意味著空閒執行緒等待新任務的最長時間為60秒，空閒執行緒超過60秒後將會被終止。

阻塞佇列採用的SynchronousQueue，SynchronousQueue是一個沒有元素的阻塞佇列，加上corePool = 0 ，maximumPoolSize = Integer.MAX_VALUE，這樣就會存在一個問題，如果主執行緒提交任務的速度遠遠大於CachedThreadPool的處理速度，則CachedThreadPool會不斷地建立新執行緒來執行任務，這樣有可能會導致系統耗盡CPU和記憶體資源，所以在使用該執行緒池是，一定要注意控制併發的任務數，否則建立大量的執行緒可能導致嚴重的效能問題。

五、執行過程

提交任務：

執行緒池根據業務不同的需求提供了兩種方式提交任務：Executor.execute()、ExecutorService.submit()。其中ExecutorService.submit()可以獲取該任務執行的Future。

execute()

執行流程如下：

（1）如果執行緒池當前執行緒數小於corePoolSize，則呼叫addWorker建立新執行緒執行任務，成功返回true，失敗執行步驟2。

（2）如果執行緒池處於RUNNING狀態，則嘗試加入阻塞佇列，如果加入阻塞佇列成功，則嘗試進行Double Check，如果加入失敗，則執行步驟3。

　　如果加入阻塞佇列成功了，則會進行一個Double Check的過程。Double Check過程的主要目的是判斷加入到阻塞隊裡中的執行緒是否可以被執行。如果執行緒池不是RUNNING狀態，則呼叫remove()方法從阻塞佇列中刪除該任務，然後呼叫reject()方法處理任務。否則需要確保還有執行緒執行。

（3）如果執行緒池不是RUNNING狀態或者加入阻塞佇列失敗，則嘗試建立新執行緒直到maxPoolSize，如果失敗，則呼叫reject()方法執行相應的拒絕策略。

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();

    // 前面說的那個表示 “執行緒池狀態” 和 “執行緒數” 的整數
    int c = ctl.get();

    // 如果當前執行緒數少於核心執行緒數，那麼直接新增一個 worker 來執行任務，
    // 建立一個新的執行緒，並把當前任務 command 作為這個執行緒的第一個任務(firstTask)
    if (workerCountOf(c) < corePoolSize) {
        // 新增任務成功，那麼就結束了。提交任務嘛，執行緒池已經接受了這個任務，這個方法也就可以返回了
        // 至於執行的結果，到時候會包裝到 FutureTask 中。
        // 返回 false 代表執行緒池不允許提交任務
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 到這裡說明，要麼當前執行緒數大於等於核心執行緒數，要麼剛剛 addWorker 失敗了

    // 如果執行緒池處於 RUNNING 狀態，把這個任務新增到任務佇列 workQueue 中
    if (isRunning(c) && workQueue.offer(command)) {
        /* 這裡面說的是，如果任務進入了 workQueue，我們是否需要開啟新的執行緒
         * 因為執行緒數在 [0, corePoolSize) 是無條件開啟新的執行緒
         * 如果執行緒數已經大於等於 corePoolSize，那麼將任務新增到佇列中，然後進到這裡
         */
        int recheck = ctl.get();
        // 如果執行緒池已不處於 RUNNING 狀態，那麼移除已經入隊的這個任務，並且執行拒絕策略
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 如果執行緒池還是 RUNNING 的，並且執行緒數為 0，那麼開啟新的執行緒
        // 到這裡，我們知道了，這塊程式碼的真正意圖是：擔心任務提交到佇列中了，但是執行緒都關閉了
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 如果 workQueue 佇列滿了，那麼進入到這個分支
    // 以 maximumPoolSize 為界建立新的 worker，
    // 如果失敗，說明當前執行緒數已經達到 maximumPoolSize，執行拒絕策略
    else if (!addWorker(command, false))
        reject(command);
}

addWorker

在這裡需要好好理論addWorker中的引數，在execute()方法中，有三處呼叫了該方法：
第一次：workerCountOf(c) < corePoolSize ==> addWorker(command, true)，這個很好理解，當然執行緒池的執行緒數量小於 corePoolSize ，則新建執行緒執行任務即可，在執行過程core == true，內部與corePoolSize比較即可。
第二次：加入阻塞佇列進行Double Check時，else if (workerCountOf(recheck) == 0) ==>addWorker(null, false)。如果執行緒池中的執行緒==0，按照道理應該該任務應該新建執行緒執行任務，但是由於已經該任務已經新增到了阻塞佇列，那麼就線上程池中新建一個空執行緒，然後從阻塞佇列中取執行緒即可。
第三次：執行緒池不是RUNNING狀態或者加入阻塞佇列失敗：else if (!addWorker(command, false))，這裡core == fase，則意味著是與maximumPoolSize比較。

執行流程：

（1）判斷當前執行緒是否可以新增任務，如果可以則進行下一步，否則return false；

1. rs >= SHUTDOWN ，表示當前執行緒處於SHUTDOWN ，STOP、TIDYING、TERMINATED狀態
2. rs == SHUTDOWN , firstTask != null時不允許新增執行緒，因為執行緒處於SHUTDOWN 狀態，不允許新增任務
3. rs == SHUTDOWN , firstTask == null，但workQueue.isEmpty() == true，不允許新增執行緒，因為firstTask == null是為了新增一個沒有任務的執行緒然後再從workQueue中獲取任務的，如果workQueue == null，則說明新增的任務沒有任何意義。

（2）內嵌迴圈，通過CAS worker + 1

（3）獲取主鎖mailLock，如果執行緒池處於RUNNING狀態獲取處於SHUTDOWN狀態且 firstTask == null，則將任務新增到workers Queue中，然後釋放主鎖mainLock，然後啟動執行緒，然後return true，如果中途失敗導致workerStarted= false，則呼叫addWorkerFailed()方法進行處理。

// 第一個引數是準備提交給這個執行緒執行的任務，之前說了，可以為 null
// 第二個引數為 true 代表使用核心執行緒數 corePoolSize 作為建立執行緒的界線，也就說建立這個執行緒的時候，
//         如果執行緒池中的執行緒總數已經達到 corePoolSize，那麼不能響應這次建立執行緒的請求
//         如果是 false，代表使用最大執行緒數 maximumPoolSize 作為界線
private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // 這個非常不好理解
        // 如果執行緒池已關閉，並滿足以下條件之一，那麼不建立新的 worker：
        // 1. 執行緒池狀態大於 SHUTDOWN，其實也就是 STOP, TIDYING, 或 TERMINATED
        // 2. firstTask != null
        // 3. workQueue.isEmpty()
        // 簡單分析下：
        // 還是狀態控制的問題，當執行緒池處於 SHUTDOWN 的時候，不允許提交任務，但是已有的任務繼續執行
        // 當狀態大於 SHUTDOWN 時，不允許提交任務，且中斷正在執行的任務
        // 多說一句：如果執行緒池處於 SHUTDOWN，但是 firstTask 為 null，且 workQueue 非空，那麼是允許建立 worker 的
        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;
            // 由於有併發，重新再讀取一下 ctl
            c = ctl.get();
            // 正常如果是 CAS 失敗的話，進到下一個裡層的for迴圈就可以了
            // 可是如果是因為其他執行緒的操作，導致執行緒池的狀態發生了變更，如有其他執行緒關閉了這個執行緒池
            // 那麼需要回到外層的for迴圈
            if (runStateOf(c) != rs)
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }

    /* 
     * 到這裡，我們認為在當前這個時刻，可以開始建立執行緒來執行任務了，
     * 因為該校驗的都校驗了，至於以後會發生什麼，那是以後的事，至少當前是滿足條件的
     */

    // worker 是否已經啟動
    boolean workerStarted = false;
    // 是否已將這個 worker 新增到 workers 這個 HashSet 中
    boolean workerAdded = false;
    Worker w = null;
    try {
        final ReentrantLock mainLock = this.mainLock;
        // 把 firstTask 傳給 worker 的構造方法
        w = new Worker(firstTask);
        // 取 worker 中的執行緒物件，之前說了，Worker的構造方法會呼叫 ThreadFactory 來建立一個新的執行緒
        final Thread t = w.thread;
        if (t != null) {
            // 這個是整個類的全域性鎖，持有這個鎖才能讓下面的操作“順理成章”，
            // 因為關閉一個執行緒池需要這個鎖，至少我持有鎖的期間，執行緒池不會被關閉
            mainLock.lock();
            try {

                int c = ctl.get();
                int rs = runStateOf(c);

                // 小於 SHUTTDOWN 那就是 RUNNING，這個自不必說，是最正常的情況
                // 如果等於 SHUTDOWN，前面說了，不接受新的任務，但是會繼續執行等待佇列中的任務
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    // worker 裡面的 thread 可不能是已經啟動的
                    if (t.isAlive())
                        throw new IllegalThreadStateException();
                    // 加到 workers 這個 HashSet 中
                    workers.add(w);
                    int s = workers.size();
                    // largestPoolSize 用於記錄 workers 中的個數的最大值
                    // 因為 workers 是不斷增加減少的，通過這個值可以知道執行緒池的大小曾經達到的最大值
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            // 新增成功的話，啟動這個執行緒
            if (workerAdded) {
                // 啟動執行緒
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        // 如果執行緒沒有啟動，需要做一些清理工作，如前面 workCount 加了 1，將其減掉
        if (! workerStarted)
            addWorkerFailed(w);
    }
    // 返回執行緒是否啟動成功
    return workerStarted;

Woker內部類

從Worker的原始碼中我們可以看到Woker繼承AQS，實現Runnable介面，所以可以認為Worker既是一個可以執行的任務，也可以達到獲取鎖釋放鎖的效果。這裡繼承AQS主要是為了方便執行緒的中斷處理。這裡注意兩個地方：建構函式、run()。建構函式主要是做三件事：1.設定同步狀態state為-1，同步狀態大於0表示就已經獲取了鎖，2.設定將當前任務task設定為firstTask，3.利用Worker本身物件this和ThreadFactory建立執行緒物件。

    private final class Worker extends AbstractQueuedSynchronizer
            implements Runnable {
        private static final long serialVersionUID = 6138294804551838833L;

        // task 的thread
        final Thread thread;

        // 執行的任務task
        Runnable firstTask;

        volatile long completedTasks;

        Worker(Runnable firstTask) {

            //設定AQS的同步狀態private volatile int state，是一個計數器，大於0代表鎖已經被獲取
            setState(-1);
            this.firstTask = firstTask;

            // 利用ThreadFactory和 Worker這個Runnable建立的執行緒物件
            this.thread = getThreadFactory().newThread(this);
        }

        // 任務執行
        public void run() {
            runWorker(this);
        }

    }

runWorker

執行流程：

（1）根據worker獲取要執行的任務task，然後呼叫unlock()方法釋放鎖，這裡釋放鎖的主要目的在於中斷，因為在new Worker時，設定的state為-1，呼叫unlock()方法可以將state設定為0，這裡主要原因就在於interruptWorkers()方法只有在state >= 0時才會執行；

（2）通過getTask()獲取執行的任務，呼叫task.run()執行，當然在執行之前會呼叫worker.lock()上鎖，執行之後呼叫worker.unlock()放鎖；

（3）在任務執行前後，可以根據業務場景自定義beforeExecute() 和 afterExecute()方法，則兩個方法在ThreadPoolExecutor中是空實現；

（4）如果執行緒執行完成，則會呼叫getTask()方法從阻塞佇列中獲取新任務，如果阻塞佇列為空，則根據是否超時來判斷是否需要阻塞；

（5）task == null或者丟擲異常（beforeExecute()、task.run()、afterExecute()均有可能）導致worker執行緒終止，則呼叫processWorkerExit()方法處理worker退出流程。

// 此方法由 worker 執行緒啟動後呼叫，這裡用一個 while 迴圈來不斷地從等待佇列中獲取任務並執行
// 前面說了，worker 在初始化的時候，可以指定 firstTask，那麼第一個任務也就可以不需要從佇列中獲取
final void runWorker(Worker w) {
    // 
    Thread wt = Thread.currentThread();
    // 該執行緒的第一個任務(如果有的話)
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        // 迴圈呼叫 getTask 獲取任務
        while (task != null || (task = getTask()) != null) {
            w.lock();          
            // 如果執行緒池狀態大於等於 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) {
                    // 這裡不允許丟擲 Throwable，所以轉換為 Error
                    thrown = x; throw new Error(x);
                } finally {
                    // 也是一個鉤子方法，將 task 和異常作為引數，留給需要的子類實現
                    afterExecute(task, thrown);
                }
            } finally {
                // 置空 task，準備 getTask 獲取下一個任務
                task = null;
                // 累加完成的任務數
                w.completedTasks++;
                // 釋放掉 worker 的獨佔鎖
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        // 如果到這裡，需要執行執行緒關閉：
        // 1. 說明 getTask 返回 null，也就是說，這個 worker 的使命結束了，執行關閉
        // 2. 任務執行過程中發生了異常
        // 第一種情況，已經在程式碼處理了將 workCount 減 1，這個在 getTask 方法分析中會說
        // 第二種情況，workCount 沒有進行處理，所以需要在 processWorkerExit 中處理
        // 限於篇幅，我不準備分析這個方法了，感興趣的讀者請自行分析原始碼
        processWorkerExit(w, completedAbruptly);
    }
}

getTask()

// 此方法有三種可能：
// 1. 阻塞直到獲取到任務返回。我們知道，預設 corePoolSize 之內的執行緒是不會被回收的，
//      它們會一直等待任務
// 2. 超時退出。keepAliveTime 起作用的時候，也就是如果這麼多時間內都沒有任務，那麼應該執行關閉
// 3. 如果發生了以下條件，此方法必須返回 null:
//    - 池中有大於 maximumPoolSize 個 workers 存在(通過呼叫 setMaximumPoolSize 進行設定)
//    - 執行緒池處於 SHUTDOWN，而且 workQueue 是空的，前面說了，這種不再接受新的任務
//    - 執行緒池處於 STOP，不僅不接受新的執行緒，連 workQueue 中的執行緒也不再執行
private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?

    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // 兩種可能
        // 1. rs == SHUTDOWN && workQueue.isEmpty()
        // 2. rs >= STOP
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            // CAS 操作，減少工作執行緒數
            decrementWorkerCount();
            return null;
        }

        boolean timed;      // Are workers subject to culling?
        for (;;) {
            int wc = workerCountOf(c);
            // 允許核心執行緒數內的執行緒回收，或當前執行緒數超過了核心執行緒數，那麼有可能發生超時關閉
            timed = allowCoreThreadTimeOut || wc > corePoolSize;

            // 這裡 break，是為了不往下執行後一個 if (compareAndDecrementWorkerCount(c))
            // 兩個 if 一起看：如果當前執行緒數 wc > maximumPoolSize，或者超時，都返回 null
            // 那這裡的問題來了，wc > maximumPoolSize 的情況，為什麼要返回 null？
            //    換句話說，返回 null 意味著關閉執行緒。
            // 那是因為有可能開發者呼叫了 setMaximumPoolSize 將執行緒池的 maximumPoolSize 調小了
            if (wc <= maximumPoolSize && ! (timedOut && timed))
                break;
            if (compareAndDecrementWorkerCount(c))
                return null;
            c = ctl.get();  // Re-read ctl
            // compareAndDecrementWorkerCount(c) 失敗，執行緒池中的執行緒數發生了改變
            if (runStateOf(c) != rs)
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
        // wc <= maximumPoolSize 同時沒有超時
        try {
            // 到 workQueue 中獲取任務
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            // 如果此 worker 發生了中斷，採取的方案是重試
            // 解釋下為什麼會發生中斷，這個讀者要去看 setMaximumPoolSize 方法，
            // 如果開發者將 maximumPoolSize 調小了，導致其小於當前的 workers 數量，
            // 那麼意味著超出的部分執行緒要被關閉。重新進入 for 迴圈，自然會有部分執行緒會返回 null
            timedOut = false;
        }
    }
}

processWorkerExit()

在runWorker()方法中，無論最終結果如何，都會執行processWorkerExit()方法對worker進行退出處理。

首先completedAbruptly的值來判斷是否需要對執行緒數-1處理，如果completedAbruptly == true，說明在任務執行過程中出現了異常，那麼需要進行減1處理，否則不需要，因為減1處理在getTask()方法中處理了。然後從HashSet中移出該worker，過程需要獲取mainlock。然後呼叫tryTerminate()方法處理，該方法是對最後一個執行緒退出做終止執行緒池動作。如果執行緒池沒有終止，那麼執行緒池需要保持一定數量的執行緒，則通過addWorker(null,false)新增一個空的執行緒。

 private void processWorkerExit(Worker w, boolean completedAbruptly) {

        // true：使用者執行緒執行異常,需要扣減
        // false：getTask方法中扣減執行緒數量
        if (completedAbruptly)
            decrementWorkerCount();

        // 獲取主鎖
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            completedTaskCount += w.completedTasks;
            // 從HashSet中移出worker
            workers.remove(w);
        } finally {
            mainLock.unlock();
        }

        // 有worker執行緒移除，可能是最後一個執行緒退出需要嘗試終止執行緒池
        tryTerminate();

        int c = ctl.get();
        // 如果執行緒為running或shutdown狀態，即tryTerminate()沒有成功終止執行緒池，則判斷是否有必要一個worker
        if (runStateLessThan(c, STOP)) {
            // 正常退出，計算min：需要維護的最小執行緒數量
            if (!completedAbruptly) {
                // allowCoreThreadTimeOut 預設false：是否需要維持核心執行緒的數量
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                // 如果min ==0 或者workerQueue為空，min = 1
                if (min == 0 && ! workQueue.isEmpty())
                    min = 1;

                // 如果執行緒數量大於最少數量min，直接返回，不需要新增執行緒
                if (workerCountOf(c) >= min)
                    return; // replacement not needed
            }
            // 新增一個沒有firstTask的worker
            addWorker(null, false);
        }
    }

六、關閉執行緒池

tryTerminate()

當執行緒池涉及到要移除worker時候都會呼叫tryTerminate()，該方法主要用於判斷執行緒池中的執行緒是否已經全部移除了，如果是的話則關閉執行緒池。

  final void tryTerminate() {
        for (;;) {
            int c = ctl.get();
            // 執行緒池處於Running狀態
            // 執行緒池已經終止了
            // 執行緒池處於ShutDown狀態，但是阻塞佇列不為空
            if (isRunning(c) ||
                    runStateAtLeast(c, TIDYING) ||
                    (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
                return;

            // 執行到這裡，就意味著執行緒池要麼處於STOP狀態，要麼處於SHUTDOWN且阻塞佇列為空
            // 這時如果執行緒池中還存線上程，則會嘗試中斷執行緒
            if (workerCountOf(c) != 0) {
                // /執行緒池還有執行緒，但是佇列沒有任務了，需要中斷喚醒等待任務的執行緒
                // （runwoker的時候首先就通過w.unlock設定執行緒可中斷，getTask最後面的catch處理中斷）
                interruptIdleWorkers(ONLY_ONE);
                return;
            }

            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // 嘗試終止執行緒池
                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                    try {
                        terminated();
                    } finally {
                        // 執行緒池狀態轉為TERMINATED
                        ctl.set(ctlOf(TERMINATED, 0));
                        termination.signalAll();
                    }
                    return;
                }
            } finally {
                mainLock.unlock();
            }
        }
    }

執行緒池ThreadPoolExecutor提供了shutdown()和shutDownNow()用於關閉執行緒池。

shutdown()：按過去執行已提交任務的順序發起一個有序的關閉，但是不接受新任務。

 public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            // 推進執行緒狀態
            advanceRunState(SHUTDOWN);
            // 中斷空閒的執行緒
            interruptIdleWorkers();
            // 交給子類實現
            onShutdown();
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
    }

shutdownNow() :嘗試停止所有的活動執行任務、暫停等待任務的處理，並返回等待執行的任務列表。

shutdownNow會呼叫interruptWorkers()方法中斷所有執行緒，同時會呼叫drainQueue()方法返回等待執行到任務列表。

public List<Runnable> shutdownNow() {
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            advanceRunState(STOP);
            // 中斷所有執行緒
            interruptWorkers();
            // 返回等待執行的任務列表
            tasks = drainQueue();
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
        return tasks;
    }

private void interruptWorkers() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (Worker w : workers)
                w.interruptIfStarted();
        } finally {
            mainLock.unlock();
        }
    }

    private List<Runnable> drainQueue() {
        BlockingQueue<Runnable> q = workQueue;
        ArrayList<Runnable> taskList = new ArrayList<Runnable>();
        q.drainTo(taskList);
        if (!q.isEmpty()) {
            for (Runnable r : q.toArray(new Runnable[0])) {
                if (q.remove(r))
                    taskList.add(r);
            }
        }
        return taskList;
    }

七、其他問題

1、任務拒絕策略

execute()方法中addWorker()失敗會呼叫reject(command) 來處理任務

執行緒池此時不能接受這個任務，所以需要執行拒絕策略

此處的 handler 我們需要在構造執行緒池的時候就傳入這個引數，它是 RejectedExecutionHandler 的例項。

RejectedExecutionHandler 在 ThreadPoolExecutor 中有四個已經定義好的實現類可供我們直接使用，當然，我們也可以實現自己的策略，不過一般也沒有必要。

final void reject(Runnable command) {
    // 執行拒絕策略
    handler.rejectedExecution(command, this);
}

// 只要執行緒池沒有被關閉，那麼由提交任務的執行緒自己來執行這個任務。
public static class CallerRunsPolicy implements RejectedExecutionHandler {
    public CallerRunsPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

// 不管怎樣，直接丟擲 RejectedExecutionException 異常
// 這個是預設的策略，如果我們構造執行緒池的時候不傳相應的 handler 的話，那就會指定使用這個
public static class AbortPolicy implements RejectedExecutionHandler {
    public AbortPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    }
}

// 不做任何處理，直接忽略掉這個任務
public static class DiscardPolicy implements RejectedExecutionHandler {
    public DiscardPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
}

// 這個相對霸道一點，如果執行緒池沒有被關閉的話，
// 把佇列隊頭的任務(也就是等待了最長時間的)直接扔掉，然後提交這個任務到等待佇列中
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    public DiscardOldestPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            e.getQueue().poll();
            e.execute(r);
        }
    }
}

2、執行緒池中的執行緒初始化

預設情況下，建立執行緒池之後，執行緒池中是沒有執行緒的，需要提交任務之後才會建立執行緒。

在實際中如果需要執行緒池建立之後立即建立執行緒，可以通過以下兩個方法辦到：

prestartCoreThread()：初始化一個核心執行緒；

prestartAllCoreThreads()：初始化所有核心執行緒

3、任務快取佇列及排隊策略

workQueue的型別為BlockingQueue<Runnable>，通常可以取下面三種類型：

（1）ArrayBlockingQueue：基於陣列的先進先出佇列，此佇列建立時必須指定大小；

（2）LinkedBlockingQueue：基於連結串列的先進先出佇列，如果建立時沒有指定此佇列大小，則預設為Integer.MAX_VALUE；

（3）synchronousQueue：這個佇列比較特殊，它不會儲存提交的任務，而是將直接新建一個執行緒來執行新來的任務。

更多BlockingQueue參考 Java併發（十八）：阻塞佇列BlockingQueue

4、執行緒池容量的動態調整

ThreadPoolExecutor提供了動態調整執行緒池容量大小的方法：setCorePoolSize()和setMaximumPoolSize()，

setCorePoolSize：設定核心池大小

setMaximumPoolSize：設定執行緒池最大能建立的執行緒數目大小

當上述引數從小變大時，ThreadPoolExecutor進行執行緒賦值，還可能立即建立新的執行緒來執行任務。

5、執行緒池的監控

（1）通過執行緒池提供的引數進行監控。

　　taskCount：執行緒池需要執行的任務數量。

　　completedTaskCount：執行緒池在執行過程中已完成的任務數量。小於或等於taskCount。

　　largestPoolSize：執行緒池曾經建立過的最大執行緒數量。通過這個資料可以知道執行緒池是否滿過。如等於執行緒池的最大大小，則表示執行緒池曾經滿了。

　　getPoolSize:執行緒池的執行緒數量。如果執行緒池不銷燬的話，池裡的執行緒不會自動銷燬，所以這個大小隻增不減。

　　getActiveCount：獲取活動的執行緒數。

（2）通過擴充套件執行緒池進行監控。

通過繼承執行緒池並重寫執行緒池的beforeExecute，afterExecute和terminated方法，我們可以在任務執行前，執行後和執行緒池關閉前幹一些事情。如監控任務的平均執行時間，最大執行時間和最小執行時間等。這幾個方法線上程池裡是空方法

八、執行緒池使用示例

為什麼使用執行緒池？

1、降低資源消耗。通過重複利用已建立的執行緒降低執行緒建立和銷燬造成的消耗。

2、提高響應速度。當任務到達時，任務可以不需要的等到執行緒建立就能立即執行。

3、提高執行緒的可管理性。執行緒是稀缺資源，如果無限制的建立，不僅會消耗系統資源，還會降低系統的穩定性，使用執行緒池可以進行統一的分配，調優和監控。但是要做到合理的利用執行緒池，必須對其原理了如指掌。

示例一：

class Test {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200,
                TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(5));

        for (int i = 0; i < 15; i++) {
            MyTask myTask = new MyTask(i);
            executor.execute(myTask);
            System.out.println("執行緒池中執行緒數目：" + executor.getPoolSize()
                    + "，佇列中等待執行的任務數目：" + executor.getQueue().size()
                    + "，已執行玩別的任務數目：" + executor.getCompletedTaskCount());
        }
        executor.shutdown();
    }
}

class MyTask implements Runnable {
    private int taskNum;

    public MyTask(int num) {
        this.taskNum = num;
    }

    @Override
    public void run() {
        System.out.println("正在執行task " + taskNum);
        try {
            Thread.currentThread().sleep(4000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("task " + taskNum + "執行完畢");
    }
}

執行結果：

正在執行task 0
執行緒池中執行緒數目：1，佇列中等待執行的任務數目：0，已執行玩別的任務數目：0
執行緒池中執行緒數目：2，佇列中等待執行的任務數目：0，已執行玩別的任務數目：0
正在執行task 1
正在執行task 2
執行緒池中執行緒數目：3，佇列中等待執行的任務數目：0，已執行玩別的任務數目：0
執行緒池中執行緒數目：4，佇列中等待執行的任務數目：0，已執行玩別的任務數目：0
正在執行task 3
執行緒池中執行緒數目：5，佇列中等待執行的任務數目：0，已執行玩別的任務數目：0
正在執行task 4
執行緒池中執行緒數目：5，佇列中等待執行的任務數目：1，已執行玩別的任務數目：0
執行緒池中執行緒數目：5，佇列中等待執行的任務數目：2，已執行玩別的任務數目：0
執行緒池中執行緒數目：5，佇列中等待執行的任務數目：3，已執行玩別的任務數目：0
執行緒池中執行緒數目：5，佇列中等待執行的任務數目：4，已執行玩別的任務數目：0
執行緒池中執行緒數目：5，佇列中等待執行的任務數目：5，已執行玩別的任務數目：0
執行緒池中執行緒數目：6，佇列中等待執行的任務數目：5，已執行玩別的任務數目：0
正在執行task 10
執行緒池中執行緒數目：7，佇列中等待執行的任務數目：5，已執行玩別的任務數目：0
正在執行task 11
執行緒池中執行緒數目：8，佇列中等待執行的任務數目：5，已執行玩別的任務數目：0
正在執行task 12
執行緒池中執行緒數目：9，佇列中等待執行的任務數目：5，已執行玩別的任務數目：0
正在執行task 13
執行緒池中執行緒數目：10，佇列中等待執行的任務數目：5，已執行玩別的任務數目：0
正在執行task 14
task 0執行完畢
正在執行task 5
task 1執行完畢
task 2執行完畢
task 3執行完畢
task 4執行完畢
正在執行task 9
正在執行task 8
正在執行task 7
正在執行task 6
task 12執行完畢
task 13執行完畢
task 10執行完畢
task 11執行完畢
task 14執行完畢
task 5執行完畢
task 6執行完畢
task 7執行完畢
task 9執行完畢
task 8執行完畢

View Code

示例二：

需求：從資料庫中獲取url，並利用httpclient迴圈訪問url地址，並對返回結果進行操作

分析：由於是迴圈的對多個url進行訪問並獲取資料，為了執行的效率，考慮使用多執行緒，url數量未知如果每個任務都建立一個執行緒將消耗大量的系統資源，最後決定使用執行緒池。

class