史上最清晰的執行緒池原始碼分析

鼎鼎大名的執行緒池。不需要多說!!!!!

這篇部落格深入分析 Java 中執行緒池的實現。

總覽

下圖是 java 執行緒池幾個相關類的繼承結構：

先簡單說說這個繼承結構，Executor 位於最頂層，也是最簡單的，就一個 execute(Runnable runnable) 介面方法定義。

ExecutorService 也是介面，在 Executor 介面的基礎上添加了很多的介面方法，所以一般來說我們會使用這個介面。

然後再下來一層是 AbstractExecutorService，從名字我們就知道，這是抽象類，這裡實現了非常有用的一些方法供子類直接使用，之後我們再細說。

然後才到我們的重點部分 ThreadPoolExecutor 類，這個類提供了關於執行緒池所需的非常豐富的功能。

執行緒池中的 BlockingQueue 也是非常重要的概念，如果執行緒數達到 corePoolSize，我們的每個任務會提交到等待佇列中，等待執行緒池中的執行緒來取任務並執行。這裡的 BlockingQueue 通常我們使用其實現類 LinkedBlockingQueue、ArrayBlockingQueue 和 SynchronousQueue，每個實現類都有不同的特徵，使用場景之後會慢慢分析。想要詳細瞭解各個 BlockingQueue 的讀者，可以參考我的前面的一篇對 BlockingQueue 的各個實現類進行詳細分析的文章。

使用示例

 1 package main.java.Juc;
 2 
 3 import java.util.concurrent.ExecutorService;
 4 import java.util.concurrent.Executors;
 5 
 6 class MyRunnable implements Runnable {
 7     @Override
 8     public void run() {
 9         for (int x = 0; x < 100; x++) {
10             System.out.println(Thread.currentThread().getName() + ":" + x);
 
11         }
12     }
13 }
14 
15 public class TestThreadPool {
16     public static void main(String[] args) {
17         // 建立一個執行緒池物件，控制要建立幾個執行緒物件。
18         ExecutorService pool = Executors.newFixedThreadPool(2);
19 
20         // 可以執行Runnable物件或者Callable物件代表的執行緒
21         pool.execute(new MyRunnable());
22         pool.execute(new MyRunnable());
23 
24         //結束執行緒池
25         pool.shutdown();
26     }
27 }

執行結果：

Executor 介面

1 public interface Executor {
2     void execute(Runnable command);
3 }

我們可以看到 Executor 介面非常簡單，就一個 void execute(Runnable command) 方法，代表提交一個任務。

 當然了，Executor 這個介面只有提交任務的功能，太簡單了，我們想要更豐富的功能，比如我們想知道執行結果、我們想知道當前執行緒池有多少個執行緒活著、已經完成了多少任務等等，這些都是這個介面的不足的地方。接下來我們要介紹的是繼承自 Executor 介面的 ExecutorService 介面，這個介面提供了比較豐富的功能，也是我們最常使用到的介面。

ExecutorService

那麼我們簡單初略地來看一下這個介面中都有哪些方法：

 1 public interface ExecutorService extends Executor {
 2 
 3     // 關閉執行緒池，已提交的任務繼續執行，不接受繼續提交新任務
 4     void shutdown();
 5 
 6     // 關閉執行緒池，嘗試停止正在執行的所有任務，不接受繼續提交新任務
 7     // 它和前面的方法相比，加了一個單詞“now”，區別在於它會去停止當前正在進行的任務
 8     List<Runnable> shutdownNow();
 9 
10     // 執行緒池是否已關閉
11     boolean isShutdown();
12 
13     // 如果呼叫了 shutdown() 或 shutdownNow() 方法後，所有任務結束了，那麼返回true
14     // 這個方法必須在呼叫shutdown或shutdownNow方法之後呼叫才會返回true
15     boolean isTerminated();
16 
17     // 等待所有任務完成，並設定超時時間
18     // 我們這麼理解，實際應用中是，先呼叫 shutdown 或 shutdownNow，
19     // 然後再調這個方法等待所有的執行緒真正地完成，返回值意味著有沒有超時
20     boolean awaitTermination(long timeout, TimeUnit unit)
21             throws InterruptedException;
22 
23     // 提交一個 Callable 任務
24     <T> Future<T> submit(Callable<T> task);
25 
26     // 提交一個 Runnable 任務，第二個引數將會放到 Future 中，作為返回值，
27     // 因為 Runnable 的 run 方法本身並不返回任何東西
28     <T> Future<T> submit(Runnable task, T result);
29 
30     // 提交一個 Runnable 任務
31     Future<?> submit(Runnable task);
32     
33     ......
34 }

這些方法都很好理解，一個簡單的執行緒池主要就是這些功能，能提交任務，能獲取結果，能關閉執行緒池，這也是為什麼我們經常用這個介面的原因。

AbstractExecutorService

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

這個抽象類實現了 ExecutorService 中的 submit 方法，newTaskFor 方法用於將任務包裝成 FutureTask。定義於最上層介面 Executor中的 void execute(Runnable command) 由於不需要獲取結果，不會進行 FutureTask 的包裝。

 1 public abstract class AbstractExecutorService implements ExecutorService {
 2 
 3     // RunnableFuture 是用於獲取執行結果的，我們常用它的子類 FutureTask
 4     // 下面兩個 newTaskFor 方法用於將我們的任務包裝成 FutureTask 提交到執行緒池中執行
 5     protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
 6         return new FutureTask<T>(runnable, value);
 7     }
 8 
 9     protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
10         return new FutureTask<T>(callable);
11     }
12 
13     // 提交任務
14     public Future<?> submit(Runnable task) {
15         if (task == null) throw new NullPointerException();
16         // 1. 將任務包裝成 FutureTask
17         RunnableFuture<Void> ftask = newTaskFor(task, null);
18         // 2. 交給執行器執行，execute 方法由具體的子類來實現
19         // 前面也說了，FutureTask 間接實現了Runnable 介面。
20         execute(ftask);
21         return ftask;
22     }
23 
24     public <T> Future<T> submit(Runnable task, T result) {
25         if (task == null) throw new NullPointerException();
26         // 1. 將任務包裝成 FutureTask
27         RunnableFuture<T> ftask = newTaskFor(task, result);
28         // 2. 交給執行器執行
29         execute(ftask);
30         return ftask;
31     }
32 
33     public <T> Future<T> submit(Callable<T> task) {
34         if (task == null) throw new NullPointerException();
35         // 1. 將任務包裝成 FutureTask
36         RunnableFuture<T> ftask = newTaskFor(task);
37         // 2. 交給執行器執行
38         execute(ftask);
39         return ftask;
40     }
41 }

到這裡，我們發現，這個抽象類包裝了一些基本的方法，可是 submit等方法，它們都沒有真正開啟執行緒來執行任務，它們都只是在方法內部呼叫了 execute 方法，所以最重要的 execute(Runnable runnable) 方法還沒出現，這裡我們要說的就是 ThreadPoolExecutor 類了。

ThreadPoolExecutor

我們經常會使用 Executors 這個工具類來快速構造一個執行緒池，對於初學者而言，這種工具類是很有用的，開發者不需要關注太多的細節，只要知道自己需要一個執行緒池，僅僅提供必需的引數就可以了，其他引數都採用作者提供的預設值。

 1 public static ExecutorService newFixedThreadPool(int nThreads) {
 2     return new ThreadPoolExecutor(nThreads, nThreads,
 3                                   0L, TimeUnit.MILLISECONDS,
 4                                   new LinkedBlockingQueue<Runnable>());
 5 }
 6 public static ExecutorService newCachedThreadPool() {
 7     return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
 8                                   60L, TimeUnit.SECONDS,
 9                                   new SynchronousQueue<Runnable>());
10 }

這裡先不說有什麼區別，它們最終都會導向這個構造方法：

 1 public ThreadPoolExecutor(int corePoolSize,
 2                           int maximumPoolSize,
 3                           long keepAliveTime,
 4                           TimeUnit unit,
 5                           BlockingQueue<Runnable> workQueue,
 6                           ThreadFactory threadFactory,
 7                           RejectedExecutionHandler handler) {
 8     if (corePoolSize < 0 ||
 9         maximumPoolSize <= 0 ||
10         maximumPoolSize < corePoolSize ||
11         keepAliveTime < 0)
12         throw new IllegalArgumentException();
13     // 這幾個引數都是必須要有的
14     if (workQueue == null || threadFactory == null || handler == null)
15         throw new NullPointerException();
16 
17     this.corePoolSize = corePoolSize;
18     this.maximumPoolSize = maximumPoolSize;
19     this.workQueue = workQueue;
20     this.keepAliveTime = unit.toNanos(keepAliveTime);
21     this.threadFactory = threadFactory;
22     this.handler = handler;
23 }

上面的構造方法中列出了我們最需要關心的幾個屬性了，下面逐個介紹下構造方法中出現的這幾個屬性：

• corePoolSize

　　　　執行緒池中的核心執行緒數。

• maximumPoolSize

　　　　最大執行緒數，執行緒池允許建立的最大執行緒數。如果當前阻塞佇列滿了，且繼續提交任務，則建立新的執行緒執行任務，前提是當前執行緒數小於maximumPoolSize；當阻塞佇列是無界佇列, 則maximumPoolSize則不起作用, 因為無法提交至核心執行緒池的執行緒會一直持續地放入workQueue

• workQueue

　　　　用來儲存等待被執行的任務的阻塞佇列. 在JDK中提供瞭如下阻塞佇列：

　　　　(1) ArrayBlockingQueue：基於陣列結構的有界阻塞佇列，按FIFO排序任務；
　　　　(2) LinkedBlockingQuene：基於連結串列結構的阻塞佇列，按FIFO排序任務，吞吐量通常要高於ArrayBlockingQuene；
　　　　(3) SynchronousQuene：一個不儲存元素的阻塞佇列，每個插入操作必須等到另一個執行緒呼叫移除操作，否則插入操作一直處於阻塞狀態，吞吐量通常要高於LinkedBlockingQuene；
　　　　(4) priorityBlockingQuene：具有優先順序的無界阻塞佇列；

　　　　有興趣的可以看看我前面關於BlockingQuene的文章

• keepAliveTime

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

• unit

　　　　keepAliveTime的單位

• threadFactory

　　　　用於生成執行緒，一般我們可以用預設的就可以了。通常，我們可以通過它將我們的執行緒的名字設定得比較可讀一些，如 Message-Thread-1， Message-Thread-2 類似這樣。

• handler

　　　　執行緒池的飽和策略，當阻塞佇列滿了，且沒有空閒的工作執行緒，如果繼續提交任務，必須採取一種策略處理該任務，執行緒池提供了4種策略： 

　　　　　　AbortPolicy：直接丟擲異常，預設策略；
　　　　　　CallerRunsPolicy：用呼叫者所在的執行緒來執行任務；
　　　　　　DiscardOldestPolicy：丟棄阻塞佇列中靠最前的任務，並執行當前任務；
　　　　　　DiscardPolicy：直接丟棄任務；
　　　　當然也可以根據應用場景實現RejectedExecutionHandler介面，自定義飽和策略，如記錄日誌或持久化儲存不能處理的任務。

除了上面幾個屬性外，我們再看看其他重要的屬性。

 1 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
 2 
 3 // 這裡 COUNT_BITS 設定為 29(32-3)，意味著前三位用於存放執行緒狀態，後29位用於存放執行緒數
 4 private static final int COUNT_BITS = Integer.SIZE - 3;
 5 
 6 // 000 11111111111111111111111111111
 7 // 這裡得到的是 29 個 1，也就是說執行緒池的最大執行緒數是 2^29-1=536870911
 8 // 以我們現在計算機的實際情況，這個數量還是夠用的
 9 private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
10 
11 // 我們說了，執行緒池的狀態存放在高 3 位中
12 // 運算結果為 111跟29個0：111 00000000000000000000000000000
13 private static final int RUNNING    = -1 << COUNT_BITS;
14 // 000 00000000000000000000000000000
15 private static final int SHUTDOWN   =  0 << COUNT_BITS;
16 // 001 00000000000000000000000000000
17 private static final int STOP       =  1 << COUNT_BITS;
18 // 010 00000000000000000000000000000
19 private static final int TIDYING    =  2 << COUNT_BITS;
20 // 011 00000000000000000000000000000
21 private static final int TERMINATED =  3 << COUNT_BITS;
22 
23 // 將整數 c 的低 29 位修改為 0，就得到了執行緒池的狀態
24 private static int runStateOf(int c)     { return c & ~CAPACITY; }
25 // 將整數 c 的高 3 為修改為 0，就得到了執行緒池中的執行緒數
26 private static int workerCountOf(int c)  { return c & CAPACITY; }
27 
28 private static int ctlOf(int rs, int wc) { return rs | wc; }
29 
30 private static boolean runStateLessThan(int c, int s) {
31     return c < s;
32 }
33 
34 private static boolean runStateAtLeast(int c, int s) {
35     return c >= s;
36 }
37 
38 private static boolean isRunning(int c) {
39     return c < SHUTDOWN;
40 }

在這裡，介紹下執行緒池中的各個狀態和狀態變化的轉換過程：

• 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() 方法結束後

另外，我們還要看看一個內部類 Worker，因為 Doug Lea 把執行緒池中的執行緒包裝成了一個個 Worker，翻譯成工人，就是執行緒池中做任務的執行緒。所以到這裡，我們知道任務是 Runnable（內部叫 task 或 command），執行緒是 Worker。

 1 private final class Worker
 2     extends AbstractQueuedSynchronizer
 3     implements Runnable{
 4     private static final long serialVersionUID = 6138294804551838833L;
 5 
 6     // 這個是真正的執行緒，任務靠你啦
 7     final Thread thread;
 8 
 9     // 前面說了，這裡的 Runnable 是任務。為什麼叫 firstTask？因為在建立執行緒的時候，如果同時指定了
10     // 這個執行緒起來以後需要執行的第一個任務，那麼第一個任務就是存放在這裡的(執行緒可不止執行這一個任務)
11     // 當然了，也可以為 null，這樣執行緒起來了，自己到任務佇列（BlockingQueue）中取任務（getTask 方法）就行了
12     Runnable firstTask;
13 
14     // 用於存放此執行緒完全的任務數，注意了，這裡用了 volatile，保證可見性
15     volatile long completedTasks;
16 
17     // Worker 只有這一個構造方法，傳入 firstTask，也可以傳 null
18     Worker(Runnable firstTask) {
19         setState(-1); // inhibit interrupts until runWorker
20         this.firstTask = firstTask;
21         // 呼叫 ThreadFactory 來建立一個新的執行緒,這裡建立的執行緒到時候用來執行任務
22         this.thread = getThreadFactory().newThread(this);
23     }
24 
25     // 這裡呼叫了外部類的 runWorker 方法
26     public void run() {
27         runWorker(this);
28     }
29 
30     ...
31 }

有了上面的這些基礎後，我們終於可以看看 ThreadPoolExecutor 的 execute 方法了，前面原始碼分析的時候也說了，各種方法都最終依賴於 execute 方法：

 1 public void execute(Runnable command) {
 2     if (command == null)
 3         throw new NullPointerException();
 4 
 5     // 前面說的那個表示 "執行緒池狀態" 和 "執行緒數" 的整數
 6     int c = ctl.get();
 7 
 8     // 如果當前執行緒數少於核心執行緒數，那麼直接新增一個 worker 來執行任務，
 9     // 建立一個新的執行緒，並把當前任務 command 作為這個執行緒的第一個任務(firstTask)
10     if (workerCountOf(c) < corePoolSize) {
11         // 新增任務成功，那麼就結束了。提交任務嘛，執行緒池已經接受了這個任務，這個方法也就可以返回了
12         // 至於執行的結果，到時候會包裝到 FutureTask 中。
13         // 這裡的true代表當前執行緒數小於corePoolSize，表示以corePoolSize為執行緒數界限
14         if (addWorker(command, true))
15             return;
16         c = ctl.get();
17     }
18     // 到這裡說明，要麼當前執行緒數大於等於核心執行緒數，要麼剛剛 addWorker 失敗了
19     // 如果執行緒池處於 RUNNING 狀態，把這個任務新增到任務佇列 workQueue 中
20     if (isRunning(c) && workQueue.offer(command)) {
21         int recheck = ctl.get();
22         // 如果執行緒池已不處於 RUNNING 狀態，那麼移除已經入隊的這個任務，並且執行拒絕策略
23         if (! isRunning(recheck) && remove(command))
24             reject(command);
25         else if (workerCountOf(recheck) == 0)
26             addWorker(null, false);
27     }
28     // 如果 workQueue 佇列滿了，那麼進入到這個分支
29     // 這裡的false代表當前執行緒數大於corePoolSize，表示以 maximumPoolSize 為界建立新的 worker
30     // 如果失敗，說明當前執行緒數已經達到 maximumPoolSize，執行拒絕策略
31     else if (!addWorker(command, false))
32         reject(command);
33 }

 我們可以看看大體的執行流程

這個方法非常重要 addWorker(Runnable firstTask, boolean core) 方法，我們看看它是怎麼建立新的執行緒的：

 1 // 第一個引數是準備提交給這個執行緒執行的任務，之前說了，可以為 null
 2 // 第二個引數為 true 代表使用核心執行緒數 corePoolSize 作為建立執行緒的界線，也就說建立這個執行緒的時候，
 3 //         如果執行緒池中的執行緒總數已經達到 corePoolSize，那麼返回false
 4 //         如果是 false，代表使用最大執行緒數 maximumPoolSize 作為界線，執行緒池中的執行緒總數已經達到 maximumPoolSize，那麼返回false
 5 private boolean addWorker(Runnable firstTask, boolean core) {
 6     retry:
 7     for (;;) {
 8         int c = ctl.get();
 9         int rs = runStateOf(c);
10 
11         // 如果執行緒池已關閉，並滿足以下條件之一，那麼不建立新的 worker：
12         // 1. 執行緒池狀態大於 SHUTDOWN，其實也就是 STOP, TIDYING, 或 TERMINATED
13         // 2. firstTask != null
14         // 3. workQueue.isEmpty()
15         if (rs >= SHUTDOWN &&
16             ! (rs == SHUTDOWN &&
17                firstTask == null &&
18                ! workQueue.isEmpty()))
19             return false;
20 
21         for (;;) {
22             int wc = workerCountOf(c);
23             //這裡就是通過core引數對當前執行緒數的判斷
24             if (wc >= CAPACITY ||
25                 wc >= (core ? corePoolSize : maximumPoolSize))
26                 return false;
27             if (compareAndIncrementWorkerCount(c))
28                 break retry;
29             c = ctl.get();
30             if (runStateOf(c) != rs)
31                 continue retry;
32             // else CAS failed due to workerCount change; retry inner loop
33         }
34     }
35 
36     /* 
37      * 到這裡，我們認為在當前這個時刻，可以開始建立執行緒來執行任務了，
38      */
39 
40     // worker 是否已經啟動
41     boolean workerStarted = false;
42     // 是否已將這個 worker 新增到 workers 這個 HashSet 中
43     boolean workerAdded = false;
44     Worker w = null;
45     try {
46         final ReentrantLock mainLock = this.mainLock;
47         // 把 firstTask 傳給 worker 的構造方法
48         w = new Worker(firstTask);
49         // 取 worker 中的執行緒物件，之前說了，Worker的構造方法會呼叫 ThreadFactory 來建立一個新的執行緒
50         final Thread t = w.thread;
51         if (t != null) {
52             // 這個是整個類的全域性鎖，因為關閉一個執行緒池需要這個鎖，至少我持有鎖的期間，執行緒池不會被關閉
53             mainLock.lock();
54             try {
55 
56                 int c = ctl.get();
57                 int rs = runStateOf(c);
58 
59                 // 小於 SHUTTDOWN 那就是 RUNNING
60                 // 如果等於 SHUTDOWN，前面說了，不接受新的任務，但是會繼續執行等待佇列中的任務
61                 if (rs < SHUTDOWN ||
62                     (rs == SHUTDOWN && firstTask == null)) {
63                     // worker 裡面的 thread 可不能是已經啟動的
64                     if (t.isAlive())
65                         throw new IllegalThreadStateException();
66                     // 加到 workers 這個 HashSet 中
67                     workers.add(w);
68                     int s = workers.size();
69                     // largestPoolSize 用於記錄 workers 中的個數的最大值
70                     // 因為 workers 是不斷增加減少的，通過這個值可以知道執行緒池的大小曾經達到的最大值
71                     if (s > largestPoolSize)
72                         largestPoolSize = s;
73                     workerAdded = true;
74                 }
75             } finally {
76                 mainLock.unlock();
77             }
78             // 新增成功的話，啟動這個執行緒
79             if (workerAdded) {
80                 // 啟動執行緒，最重要的就是這裡，下面我們會講解如何執行任務
81                 t.start();
82                 workerStarted = true;
83             }
84         }
85     } finally {
86         // 如果執行緒沒有啟動，需要做一些清理工作，如前面 workCount 加了 1，將其減掉
87         if (! workerStarted)
88             addWorkerFailed(w);
89     }
90     // 返回執行緒是否啟動成功
91     return workerStarted;
92 }

上面第81行程式碼處已經啟動了執行緒，w = new Worker(firstTask);  t = w.thread，我們接著看看Worker這個類

 1 private final class Worker
 2     extends AbstractQueuedSynchronizer
 3     implements Runnable{
 4     private static final long serialVersionUID = 6138294804551838833L;
 5     final Thread thread;
 6     Runnable firstTask;
 7     volatile long completedTasks;
 8 
 9     // Worker 只有這一個構造方法，傳入 firstTask
10     Worker(Runnable firstTask) {
11         setState(-1); // inhibit interrupts until runWorker
12         this.firstTask = firstTask;
13         // 呼叫 ThreadFactory 來建立一個新的執行緒,這裡建立的執行緒到時候用來執行任務
14         // 我們發現建立執行緒的時候傳入的值是this，我們知道建立執行緒可以通過繼承Runnable的方法，
15         // Worker繼承了Runnable，並且下面重寫了run()方法
16         this.thread = getThreadFactory().newThread(this);
17     }
18 
19     // 由上面建立執行緒時傳入的this,上面的thread啟動後，會執行這裡的run()方法，並且此時runWorker傳入的也是this
20     public void run() {
21         runWorker(this);
22     }
23 }

繼續往下看 runWorker 方法：

 1 // 此方法由 worker 執行緒啟動後呼叫，這裡用一個 while 迴圈來不斷地從等待佇列中獲取任務並執行
 2 // 前面說了，worker 在初始化的時候，可以指定 firstTask，那麼第一個任務也就可以不需要從佇列中獲取
 3 final void runWorker(Worker w) {
 4     Thread wt = Thread.currentThread();
 5     // 該執行緒的第一個任務(如果有的話)
 6     Runnable task = w.firstTask;
 7     w.firstTask = null;
 8     w.unlock(); // allow interrupts
 9     boolean completedAbruptly = true;
10     try {
11         // 迴圈呼叫 getTask 獲取任務
12         while (task != null || (task = getTask()) != null) {
13             w.lock();          
14             // 如果執行緒池狀態大於等於 STOP，那麼意味著該執行緒也要中斷
15             if ((runStateAtLeast(ctl.get(), STOP) ||
16                  (Thread.interrupted() &&
17                   runStateAtLeast(ctl.get(), STOP))) &&
18                 !wt.isInterrupted())
19                 wt.interrupt();
20             try {
21                 beforeExecute(wt, task);
22                 Throwable thrown = null;
23                 try {
24                     // 到這裡終於可以執行任務了，這裡是最重要的，task是什麼？是Worker 中的firstTask屬性
25                     // 也就是上面我們使用示例裡面的 new MyRunnable()例項，這裡就是真正的執行run方法裡面的程式碼
26                     task.run();
27                 } catch (RuntimeException x) {
28                     thrown = x; throw x;
29                 } catch (Error x) {
30                     thrown = x; throw x;
31                 } catch (Throwable x) {
32                     thrown = x; throw new Error(x);
33                 } finally {
34                     afterExecute(task, thrown);
35                 }
36             } finally {
37                 // 一個任務執行完了，這個執行緒還可以複用，接著去佇列中拉取任務執行
38                 // 置空 task，準備 getTask 獲取下一個任務
39                 task = null;
40                 // 累加完成的任務數
41                 w.completedTasks++;
42                 // 釋放掉 worker 的獨佔鎖
43                 w.unlock();
44             }
45         }
46         completedAbruptly = false;
47     } finally {
48         // 如果到這裡，需要執行執行緒關閉：
49         // 說明 getTask 返回 null，也就是超過corePoolSize的執行緒過了超時時間還沒有獲取到任務，也就是說，這個 worker 的使命結束了，執行關閉
50         processWorkerExit(w, completedAbruptly);
51     }
52 }

我們看看 getTask() 是怎麼獲取任務的

 1 // 此方法有三種可能：
 2 // 1. 阻塞直到獲取到任務返回。我們知道，預設 corePoolSize 之內的執行緒是不會被回收的，
 3 //      它們會一直等待任務
 4 // 2. 超時退出。keepAliveTime 起作用的時候，也就是如果這麼多時間內都沒有任務，那麼應該執行關閉
 5 // 3. 如果發生了以下條件，此方法必須返回 null:
 6 //    - 池中有大於 maximumPoolSize 個 workers 存在(通過呼叫 setMaximumPoolSize 進行設定)
 7 //    - 執行緒池處於 SHUTDOWN，而且 workQueue 是空的，前面說了，這種不再接受新的任務
 8 //    - 執行緒池處於 STOP，不僅不接受新的執行緒，連 workQueue 中的執行緒也不再執行
 9 private Runnable getTask() {
10     boolean timedOut = false; // Did the last poll() time out?
11 
12     retry:
13     for (;;) {
14         int c = ctl.get();
15         int rs = runStateOf(c);
16         // 兩種可能
17         // 1. rs == SHUTDOWN && workQueue.isEmpty()
18         // 2. rs >= STOP
19         if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
20             // CAS 操作，減少工作執行緒數
21             decrementWorkerCount();
22             return null;
23         }
24 
25         boolean timed;      // Are workers subject to culling?
26         for (;;) {
27             int wc = workerCountOf(c);
28             // 允許核心執行緒數內的執行緒回收，或當前執行緒數超過了核心執行緒數，那麼有可能發生超時關閉
29             timed = allowCoreThreadTimeOut || wc > corePoolSize;
30             if (wc <= maximumPoolSize && ! (timedOut && timed))
31                 break;
32             if (compareAndDecrementWorkerCount(c))
33                 return null;
34             c = ctl.get();  // Re-read ctl
35             // compareAndDecrementWorkerCount(c) 失敗，執行緒池中的執行緒數發生了改變
36             if (runStateOf(c) != rs)
37                 continue retry;
38             // else CAS failed due to workerCount change; retry inner loop
39         }
40         // wc <= maximumPoolSize 同時沒有超時
41         try {
42             // 到 workQueue 中獲取任務
43             // 如果timed=wc > corePoolSize=false,我們知道核心執行緒數之內的執行緒永遠不會銷燬，則執行workQueue.take();我前面文章中講過，take()方法是阻塞方法，如果隊裡中有任務則取到任務，如果沒有任務，則一直阻塞在這裡知道有任務被喚醒。
44             //如果timed=wc > corePoolSize=true,這裡將執行超時策略，poll(keepAliveTime, TimeUnit.NANOSECONDS)會阻塞keepAliveTime這麼長時間，沒超時就返回任務，超時則返回null.
45             Runnable r = timed ?
46                 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
47                 workQueue.take();
48             if (r != null)
49                 return r;
50             timedOut = true;
51         } catch (InterruptedException retry) {
52             // 如果此 worker 發生了中斷，採取的方案是重試
53             // 解釋下為什麼會發生中斷，這個讀者要去看 setMaximumPoolSize 方法，
54             // 如果開發者將 maximumPoolSize 調小了，導致其小於當前的 workers 數量，
55             // 那麼意味著超出的部分執行緒要被關閉。重新進入 for 迴圈，自然會有部分執行緒會返回 null
56             timedOut = false;
57         }
58     }
59 }

到這裡，基本上也說完了整個流程，讀者這個時候應該回到 execute(Runnable command) 方法，有兩種情況會呼叫 reject(command) 來處理任務，因為按照正常的流程，執行緒池此時不能接受這個任務，所以需要執行我們的拒絕策略。接下來，我們說一說 ThreadPoolExecutor 中的拒絕策略。

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

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

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

 1 // 只要執行緒池沒有被關閉，那麼由提交任務的執行緒自己來執行這個任務。
 2 public static class CallerRunsPolicy implements RejectedExecutionHandler {
 3     public CallerRunsPolicy() { }
 4     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
 5         if (!e.isShutdown()) {
 6             r.run();
 7         }
 8     }
 9 }
10 
11 // 不管怎樣，直接丟擲 RejectedExecutionException 異常
12 // 這個是預設的策略，如果我們構造執行緒池的時候不傳相應的 handler 的話，那就會指定使用這個
13 public static class AbortPolicy implements RejectedExecutionHandler {
14     public AbortPolicy() { }
15     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
16         throw new RejectedExecutionException("Task " + r.toString() +
17                                              " rejected from " +
18                                              e.toString());
19     }
20 }
21 
22 // 不做任何處理，直接忽略掉這個任務
23 public static class DiscardPolicy implements RejectedExecutionHandler {
24     public DiscardPolicy() { }
25     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
26     }
27 }
28 
29 // 這個相對霸道一點，如果執行緒池沒有被關閉的話，
30 // 把佇列隊頭的任務(也就是等待了最長時間的)直接扔掉，然後提交這個任務到等待佇列中
31 public static class DiscardOldestPolicy implements RejectedExecutionHandler {
32     public DiscardOldestPolicy() { }
33     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
34         if (!e.isShutdown()) {
35             e.getQueue().poll();
36             e.execute(r);
37         }
38     }
39 }

到這裡，ThreadPoolExecutor 算是分析得差不多了

推薦部落格

　　https://www.cnblogs.com/chen-haozi/p/10227797.html

總結

我們簡單回顧下執行緒建立的流程

1. 如果當前執行緒數少於 corePoolSize，那麼提交任務的時候建立一個新的執行緒，並由這個執行緒執行這個任務；
2. 如果當前執行緒數已經達到 corePoolSize，那麼將提交的任務新增到佇列中，等待執行緒池中的執行緒去佇列中取任務；
3. 如果佇列已滿，那麼建立新的執行緒來執行任務，需要保證池中的執行緒數不會超過 maximumPoolSize，如果此時執行緒數超過了 maximumPoolSize，那麼執行拒絕策略。