Java 線程池 實現原理與源碼深度解析
正文java
史上最清晰的線程池源碼分析數組
鼎鼎大名的線程池。不須要多說!!!!!架構
這篇博客深刻分析 Java 中線程池的實現。ide
總覽
下圖是 java 線程池幾個相關類的繼承結構：工具
先簡單說說這個繼承結構，Executor 位於最頂層，也是最簡單的，就一個 execute(Runnable runnable) 接口方法定義。oop
ExecutorService 也是接口，在 Executor 接口的基礎上添加了不少的接口方法，因此通常來講咱們會使用這個接口。源碼分析
而後再下來一層是 AbstractExecutorService，從名字咱們就知道，這是抽象類，這裏實現了很是有用的一些方法供子類直接使用，以後咱們再細說。學習
而後纔到咱們的重點部分 ThreadPoolExecutor 類，這個類提供了關於線程池所需的很是豐富的功能。this
線程池中的 BlockingQueue 也是很是重要的概念，若是線程數達到 corePoolSize，咱們的每一個任務會提交到等待隊列中，等待線程池中的線程來取任務並執行。這裏的 BlockingQueue 一般咱們使用其實現類 LinkedBlockingQueue、ArrayBlockingQueue 和 SynchronousQueue，每一個實現類都有不一樣的特徵，使用場景以後會慢慢分析。想要詳細瞭解各個 BlockingQueue 的讀者，能夠參考個人前面的一篇對 BlockingQueue 的各個實現類進行詳細分析的文章。spa
使用示例
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() 方法結束後，線程池的狀態就會變成這個
- 在此我向你們推薦一個架構學習交流圈：830478757 幫助突破瓶頸 提高思惟能力
看了這幾種狀態的介紹，讀者大致也能夠猜到十之八九的狀態轉換了，各個狀態的轉換過程有如下幾種：
- 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 的話，那就會指定使用這個
在此我向你們推薦一個架構學習交流圈：830478757 幫助突破瓶頸 提高思惟能力
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 算是分析得差很少了
相關文章
- 1. Java 線程池 實現原理與源碼深度解析
- 2. Java線程池實現原理與源碼解析(jdk1.8)
- 3. 深刻源碼分析Java線程池的實現原理
- 4. 深刻源碼分析-線程池的實現原理
- 5. Java 線程池的原理與實現
- 6. 深刻源碼分析Java線程池的實現原理(轉載)
- 7. 深刻Java源碼理解線程池原理
- 8. Java線程池ThreadPoolExecutor深度探索及源碼解析
- 9. Java線程池實現原理詳解
- 10. 併發編程之 源碼剖析 線程池 實現原理
- 更多相關文章...
- • C# 多線程 - C#教程
- • BASE原理與最終一致性 - NoSQL教程
- • Java Agent入門實戰（三）-JVM Attach原理與使用
- • Java Agent入門實戰（二）-Instrumentation源碼概述
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. 「插件」Runner更新Pro版，幫助設計師遠離996
- 2. 錯誤 707 Could not load file or assembly ‘Newtonsoft.Json, Version=12.0.0.0, Culture=neutral, PublicKe
- 3. Jenkins 2018 報告速覽，Kubernetes使用率躍升235%！
- 4. TVI-Android技術篇之註解Annotation
- 5. android studio啓動項目
- 6. Android的ADIL
- 7. Android卡頓的檢測及優化方法彙總（線下+線上）
- 8. 登錄註冊的業務邏輯流程梳理
- 9. NDK(1)創建自己的C/C++文件
- 10. 小菜的系統框架界面設計-你的評估是我的決策
歡迎關注本站公眾號,獲取更多信息
