前言
ThreadPoolExecutor在concurrent包下，是我們最常用的類之一。無論是做大數據的，還是寫業務開發，對其透徹的理解以及如何發揮更好的性能，成為了我們在更好的coding道路上必不可少的基礎。
為什么用線程池？
如果並發的請求數量非常多，但每個線程執行的時間很短，這樣就會頻繁的創建和銷毀線程，如此一來會大大降低系統的效率。這就是線程池的目的了。線程池為線程生命周期的開銷和資源不足問題提供了解決方案。通過對多個任務重用線程，線程創建的開銷被分攤到了多個任務上。
創建ThreadPoolExecutor
ThreadPoolExecutor總共有四個構造方法，下面選擇了一個最全參數的構造方法進行分析。
corePoolSize：
核心線程數量，當有新任務在execute()方法提交時，會執行以下判斷：
如果運行的線程少於 corePoolSize，則創建新線程來處理任務，即使線程池中的其他線程是空閑的；
如果線程池中的線程數量大於等於 corePoolSize 且小於 maximumPoolSize，則只有當workQueue滿時才創建新的線程去處理任務；
如果設置的corePoolSize 和 maximumPoolSize相同，則創建的線程池的大小是固定的，這時如果有新任務提交，若workQueue未滿，則將請求放入workQueue中，等待有空閑的線程去從workQueue中取任務並處理；
如果運行的線程數量大於等於maximumPoolSize，這時如果workQueue已經滿了，則通過handler所指定的策略來處理任務；
所以，任務提交時，判斷的順序為 corePoolSize –> workQueue –> maximumPoolSize。
maximumPoolSize：最大線程數量；
keepAliveTime:線程的存活時間。當線程池里的線程數大於corePoolSize時，如果等了keepAliveTime時長還沒有任務可執行，則線程退出。
unit：這個用來指定keepAliveTime的單位，比如秒:TimeUnit.SECONDS。
workQueue：保存等待執行的任務的阻塞隊列，當提交一個新的任務到線程池以后, 線程池會根據當前線程池中正在運行着的線程的數量來決定對該任務的處理方式，主要有以下幾種處理方式:
直接切換：這種方式常用的隊列是SynchronousQueue，但現在還沒有研究過該隊列，這里暫時還沒法介紹；
使用無界隊列：一般使用基於鏈表的阻塞隊列LinkedBlockingQueue。如果使用這種方式，那么線程池中能夠創建的最大線程數就是corePoolSize，而maximumPoolSize就不會起作用了（后面也會說到）。當線程池中所有的核心線程都是RUNNING狀態時，這時一個新的任務提交就會放入等待隊列中。
使用有界隊列：一般使用ArrayBlockingQueue。使用該方式可以將線程池的最大線程數量限制為maximumPoolSize，這樣能夠降低資源的消耗，但同時這種方式也使得線程池對線程的調度變得更困難，因為線程池和隊列的容量都是有限的值，所以要想使線程池處理任務的吞吐率達到一個相對合理的范圍，又想使線程調度相對簡單，並且還要盡可能的降低線程池對資源的消耗，就需要合理的設置這兩個數量。
1⃣️ 如果要想降低系統資源的消耗（包括CPU的使用率，操作系統資源的消耗，上下文環境切換的開銷等）, 可以設置較大的隊列容量和較小的線程池容量, 但這樣也會降低線程處理任務的吞吐量。
2⃣️ 如果提交的任務經常發生阻塞，那么可以考慮通過調用 setMaximumPoolSize() 方法來重新設定線程池的容量。（思路類似於BackPressure動態調節，原理不同）
3⃣️ 如果隊列的容量設置的較小，通常需要將線程池的容量設置大一點，這樣CPU的使用率會相對的高一些。但如果線程池的容量設置的過大，則在提交的任務數量太多的情況下，並發量會增加，那么線程之間的調度就是一個要考慮的問題，因為這樣反而有可能降低處理任務的吞吐量。
threadFactory：它是ThreadFactory類型的變量，用來創建新線程。默認使用Executors.defaultThreadFactory()來創建線程。使用默認的ThreadFactory來創建線程時，會使新創建的線程具有相同的NORM_PRIORITY優先級並且是非守護線程，同時也設置了線程的名稱。
handler：它是RejectedExecutionHandler類型的變量，表示線程池的飽和策略。如果阻塞隊列滿了並且沒有空閑的線程，這時如果繼續提交任務，就需要采取一種策略處理該任務。
線程池提供了4種策略：
AbortPolicy：直接拋出異常，這是默認策略；
CallerRunsPolicy：用調用者所在的線程來執行任務；
DiscardOldestPolicy：丟棄阻塞隊列中靠最前的任務，並執行當前任務；
DiscardPolicy：直接丟棄任務；
執行流程
任務被提交到線程池，會先判斷當前線程數量是否小於corePoolSize，如果小於則創建線程來執行提交的任務，否則將任務放入workQueue隊列，如果workQueue滿了，則判斷當前線程數量是否小於maximumPoolSize,如果小於則創建線程執行任務，否則就會調用handler，以表示線程池拒絕接收任務。
運行狀態
RUNNING：能接受新提交的任務，並且也能處理阻塞隊列中的任務；
SHUTDOWN：關閉狀態，不再接受新提交的任務，但卻可以繼續處理阻塞隊列中已保存的任務。在線程池處於 RUNNING 狀態時，調用 shutdown()方法會使線程池進入到該狀態。（finalize() 方法在執行過程中也會調用shutdown()方法進入該狀態）；
STOP：不能接受新任務，也不處理隊列中的任務，會中斷正在處理任務的線程。在線程池處於 RUNNING 或 SHUTDOWN 狀態時，調用 shutdownNow() 方法會使線程池進入到該狀態；
TIDYING：如果所有的任務都已終止了，workerCount (有效線程數) 為0，線程池進入該狀態后會調用 terminated() 方法進入TERMINATED 狀態。
TERMINATED：在terminated() 方法執行完后進入該狀態。
核心代碼
execute
/**
* Executes the given task sometime in the future. The task
* may execute in a new thread or in an existing pooled thread.
*
* If the task cannot be submitted for execution, either because this
* executor has been shutdown or because its capacity has been reached,
* the task is handled by the current {@code RejectedExecutionHandler}.
*
* @param command the task to execute
* @throws RejectedExecutionException at discretion of
* {@code RejectedExecutionHandler}, if the task
* cannot be accepted for execution
* @throws NullPointerException if {@code command} is null
*/
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
// clt記錄着runState和workerCount，即線程池狀態和有效線程數
int c = ctl.get();
// workerCountOf方法取出低29位的值，表示當前活動的線程數
// 如果當前活動線程數小於corePoolSize，則新建一個線程放入線程池中，並把任務添加到該線程中（調用addWorker創建線程執行任務）
if (workerCountOf(c) < corePoolSize) {
// addWorker中的第二個參數表示限制添加線程的數量是根據corePoolSize來判斷還是maximumPoolSize來判斷；
// 如果為true，根據corePoolSize來判斷；
// 如果為false，則根據maximumPoolSize來判斷
if (addWorker(command, true))
return;
// 如果添加失敗，則重新獲取ctl值
c = ctl.get();
}
// 如果不小於corePoolSize，則將任務添加到workQueue隊列
// 如果當前線程池是運行狀態並且任務添加到隊列成功
if (isRunning(c) && workQueue.offer(command)) {
// 重新獲取ctl值
int recheck = ctl.get();
// 再次判斷線程池的運行狀態，如果不是運行狀態，由於之前已經把command添加到workQueue中了，
// 需要移除該command，執行過后通過handler使用拒絕策略對該任務進行處理，整個方法返回
if (! isRunning(recheck) && remove(command))
reject(command);
/*
* 獲取線程池中的有效線程數，如果數量是0，則執行addWorker方法
* 這里傳入的參數表示：
* 1. 第一個參數為null，表示在線程池中創建一個線程，但不去啟動；
* 2. 第二個參數為false，將線程池的有限線程數量的上限設置為maximumPoolSize，添加線程時根據maximumPoolSize來判斷；
* 如果判斷workerCount不等於0，則直接返回，在workQueue中新增的command會在將來的某個時刻被執行。
*/
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
/*
* 如果執行到這里，有兩種情況：
* 1. 線程池已經不是RUNNING狀態；
* 2. 線程池是RUNNING狀態，但workerCount >= corePoolSize並且workQueue已滿。
* 這時，再次調用addWorker方法，但第二個參數傳入為false，將線程池的有限線程數量的上限設置為maximumPoolSize；
* 如果失敗則拒絕該任務
*/
else if (!addWorker(command, false))
reject(command);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
簡單來說，在執行execute()方法時如果狀態一直是RUNNING時，的執行過程如下：
如果workerCount < corePoolSize，則創建並啟動一個線程來執行新提交的任務；
如果workerCount >= corePoolSize，且線程池內的阻塞隊列未滿，則將任務添加到該阻塞隊列中；
如果workerCount >= corePoolSize && workerCount < maximumPoolSize，且線程池內的阻塞隊列已滿，則創建並啟動一個線程來執行新提交的任務；
如果workerCount >= maximumPoolSize，並且線程池內的阻塞隊列已滿, 則根據拒絕策略來處理該任務,
默認的處理方式是直接拋異常。
這里要注意一下addWorker(null, false);，也就是創建一個線程，但並沒有傳入任務，因為任務已經被添加到workQueue中了，所以worker在執行的時候，會直接從workQueue中獲取任務。所以，在workerCountOf(recheck) == 0時執行addWorker(null, false);也是為了保證線程池在RUNNING狀態下必須要有一個線程來執行任務。
addWorker
addWorker方法的主要工作是在線程池中創建一個新的線程並執行，firstTask參數 用於指定新增的線程執行的第一個任務，core參數為true表示在新增線程時會判斷當前活動線程數是否少於corePoolSize，false表示新增線程前需要判斷當前活動線程數是否少於maximumPoolSize，代碼如下：
/**
* Checks if a new worker can be added with respect to current
* pool state and the given bound (either core or maximum). If so,
* the worker count is adjusted accordingly, and, if possible, a
* new worker is created and started, running firstTask as its
* first task. This method returns false if the pool is stopped or
* eligible to shut down. It also returns false if the thread
* factory fails to create a thread when asked. If the thread
* creation fails, either due to the thread factory returning
* null, or due to an exception (typically OutOfMemoryError in
* Thread.start()), we roll back cleanly.
*
* @param firstTask the task the new thread should run first (or
* null if none). Workers are created with an initial first task
* (in method execute()) to bypass queuing when there are fewer
* than corePoolSize threads (in which case we always start one),
* or when the queue is full (in which case we must bypass queue).
* Initially idle threads are usually created via
* prestartCoreThread or to replace other dying workers.
*
* @param core if true use corePoolSize as bound, else
* maximumPoolSize. (A boolean indicator is used here rather than a
* value to ensure reads of fresh values after checking other pool
* state).
* @return true if successful
*/
private boolean addWorker(Runnable firstTask, boolean core) {
// 該處retry跳轉標簽的用法解析：[java標簽跳轉用法解析](https://blog.csdn.net/zyzzxycj/article/details/90295537)
retry:
for (;;) {
int c = ctl.get();
// 獲取運行狀態
int rs = runStateOf(c);
// Check if queue empty only if necessary.
/*
* 這個if判斷
* 如果rs >= SHUTDOWN，則表示線程池狀態為STOP或者TIDYING或者TERMINATED，此時不再接收新任務；
* 接着判斷以下3個條件，只要有1個不滿足，則返回false：
* 1. rs == SHUTDOWN，這時表示關閉狀態，不再接受新提交的任務，但卻可以繼續處理阻塞隊列中已保存的任務
* 2. firsTask為空
* 3. 阻塞隊列不為空
*
* 首先考慮rs == SHUTDOWN的情況
* 這種情況下不會接受新提交的任務，所以在firstTask不為空的時候會返回false；
* 然后，如果firstTask為空，並且workQueue也為空，則返回false，
* 因為隊列中已經沒有任務了，不需要再添加線程了
*/
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
// 獲取線程數
int wc = workerCountOf(c);
// 如果wc超過CAPACITY，也就是ctl的低29位的最大值（二進制是29個1），返回false；
// 在創建非核心線程時，即core等於false時，判斷當前線程數是否大於等於maximumPoolSize，
// 在core等於true時，判斷當前線程數是否大於等於corePoolSize，
// 如果大於等於則返回false。
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 嘗試CAS增加workerCount，如果成功，則跳出所有for循環
if (compareAndIncrementWorkerCount(c))
break retry;
// 如果增加workerCount失敗，則重新獲取ctl的值
c = ctl.get(); // Re-read ctl
// 如果當前的運行狀態不等於rs，說明狀態已被改變，返回到第一個for循環外繼續執行
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
// 根據firstTask來創建Worker對象
w = new Worker(firstTask);
// 每一個Worker對象都會創建一個線程
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int rs = runStateOf(ctl.get());
// rs < SHUTDOWN表示是RUNNING狀態；
// 如果rs是RUNNING狀態或者rs是SHUTDOWN狀態並且firstTask為null，向線程池中添加線程。
// 因為在SHUTDOWN時不會在添加新的任務，但還是會執行workQueue中的任務
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
// workers是一個HashSet<Worker>();
workers.add(w);
int s = workers.size();
// largestPoolSize記錄着線程池中出現過的最大線程數量
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
// 啟動這個線程
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
注意一下這里的t.start()這個語句，啟動時會調用Worker類中的run方法，Worker本身實現了Runnable接口，所以一個Worker類型的對象也是一個線程。
worker
線程池中的每一個線程被封裝成一個Worker對象，ThreadPool維護的其實就是一組Worker對象，看一下Worker的定義：
/**
* Class Worker mainly maintains interrupt control state for
* threads running tasks, along with other minor bookkeeping.
* This class opportunistically extends AbstractQueuedSynchronizer
* to simplify acquiring and releasing a lock surrounding each
* task execution. This protects against interrupts that are
* intended to wake up a worker thread waiting for a task from
* instead interrupting a task being run. We implement a simple
* non-reentrant mutual exclusion lock rather than use
* ReentrantLock because we do not want worker tasks to be able to
* reacquire the lock when they invoke pool control methods like
* setCorePoolSize. Additionally, to suppress interrupts until
* the thread actually starts running tasks, we initialize lock
* state to a negative value, and clear it upon start (in
* runWorker).
*/
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/**
* This class will never be serialized, but we provide a
* serialVersionUID to suppress a javac warning.
*/
private static final long serialVersionUID = 6138294804551838833L;
/** Thread this worker is running in. Null if factory fails. */
// thread是在調用構造方法時通過ThreadFactory來創建的線程，是用來處理任務的線程
final Thread thread;
/** Initial task to run. Possibly null. */
// firstTask用它來保存傳入的任務
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
/**
* Creates with given first task and thread from ThreadFactory.
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
// 禁止在執行任務前對線程進行中斷
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
// 可以看到在創建Worker時會調用threadFactory來創建一個線程
// newThread方法傳入的參數是this，因為Worker本身繼承了Runnable接口，也就是一個線程，所以一個Worker對象在啟動的時候會調用Worker類中的run方法
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
public void run() {
// addWorker中的t.start()就會觸發該run方法
runWorker(this);
}
// Lock methods
//
// The value 0 represents the unlocked state.
// The value 1 represents the locked state.
protected boolean isHeldExclusively() {
return getState() != 0;
}
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
public void lock() { acquire(1); }
public boolean tryLock() { return tryAcquire(1); }
public void unlock() { release(1); }
public boolean isLocked() { return isHeldExclusively(); }
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
Worker類繼承了AQS，並實現了Runnable接口，注意其中的firstTask和thread屬性：firstTask用它來保存傳入的任務；thread是在調用構造方法時通過ThreadFactory來創建的線程，是用來處理任務的線程。
在調用構造方法時，需要把任務傳入，這里通過getThreadFactory().newThread(this);來新建一個線程，newThread方法傳入的參數是this，因為Worker本身繼承了Runnable接口，也就是一個線程，所以一個Worker對象在啟動的時候會調用Worker類中的run方法。
Worker繼承了AQS，使用AQS來實現獨占鎖的功能。為什么不使用ReentrantLock來實現呢？可以看到tryAcquire方法，它是不允許重入的，而ReentrantLock是允許重入的：
lock方法一旦獲取了獨占鎖，表示當前線程正在執行任務中；
如果正在執行任務，則不應該中斷線程；
如果該線程現在不是獨占鎖的狀態，也就是空閑的狀態，說明它沒有在處理任務，這時可以對該線程進行中斷；
線程池在執行shutdown方法或tryTerminate方法時會調用interruptIdleWorkers方法來中斷空閑的線程，interruptIdleWorkers方法會使用tryLock方法來判斷線程池中的線程是否是空閑狀態；
之所以設置為不可重入，是因為我們不希望任務在調用像setCorePoolSize這樣的線程池控制方法時重新獲取鎖。如果使用ReentrantLock，它是可重入的，這樣如果在任務中調用了如setCorePoolSize這類線程池控制的方法，會中斷正在運行的線程。
所以，Worker繼承自AQS，用於判斷線程是否空閑以及是否可以被中斷。
此外，在構造方法中執行了setState(-1);，把state變量設置為-1，為什么這么做呢？是因為AQS中默認的state是0，如果剛創建了一個Worker對象，還沒有執行任務時，這時就不應該被中斷，看一下tryAquire方法：
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
1
2
3
4
5
6
7
tryAcquire方法是根據state是否是0來判斷的，所以，setState(-1);將state設置為-1是為了禁止在執行任務前對線程進行中斷。
正因為如此，在runWorker方法中會先調用Worker對象的unlock方法將state設置為0.
runWorker
/**
* Main worker run loop. Repeatedly gets tasks from queue and
* executes them, while coping with a number of issues:
*
* 1. We may start out with an initial task, in which case we
* don't need to get the first one. Otherwise, as long as pool is
* running, we get tasks from getTask. If it returns null then the
* worker exits due to changed pool state or configuration
* parameters. Other exits result from exception throws in
* external code, in which case completedAbruptly holds, which
* usually leads processWorkerExit to replace this thread.
*
* 2. Before running any task, the lock is acquired to prevent
* other pool interrupts while the task is executing, and then we
* ensure that unless pool is stopping, this thread does not have
* its interrupt set.
*
* 3. Each task run is preceded by a call to beforeExecute, which
* might throw an exception, in which case we cause thread to die
* (breaking loop with completedAbruptly true) without processing
* the task.
*
* 4. Assuming beforeExecute completes normally, we run the task,
* gathering any of its thrown exceptions to send to afterExecute.
* We separately handle RuntimeException, Error (both of which the
* specs guarantee that we trap) and arbitrary Throwables.
* Because we cannot rethrow Throwables within Runnable.run, we
* wrap them within Errors on the way out (to the thread's
* UncaughtExceptionHandler). Any thrown exception also
* conservatively causes thread to die.
*
* 5. After task.run completes, we call afterExecute, which may
* also throw an exception, which will also cause thread to
* die. According to JLS Sec 14.20, this exception is the one that
* will be in effect even if task.run throws.
*
* The net effect of the exception mechanics is that afterExecute
* and the thread's UncaughtExceptionHandler have as accurate
* information as we can provide about any problems encountered by
* user code.
*
* @param w the worker
*/
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
// 獲取第一個任務
Runnable task = w.firstTask;
w.firstTask = null;
// 允許在執行任務前對線程進行中斷
w.unlock(); // allow interrupts
// 是否因為異常退出循環
boolean completedAbruptly = true;
try {
// 線程調用runWoker，會while循環調用getTask方法從workerQueue里讀取任務，然后執行任務。
// // 如果task為空，則通過getTask來獲取任務，只要getTask方法不返回null,此線程就不會退出。
while (task != null || (task = getTask()) != null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
// 如果線程池正在停止，那么要保證當前線程是中斷狀態；如果不是的話，則要保證當前線程不是中斷狀態
// 這里要考慮在執行該if語句期間可能也執行了shutdownNow方法，shutdownNow方法會把狀態設置為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) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
上方的w.lock()和w.unlock()是通過AQS的acquire和release方法實現的。
AQS可以參考這篇文章：
這里說明一下第一個if判斷，目的是：
如果線程池正在停止，那么要保證當前線程是中斷狀態；
如果不是的話，則要保證當前線程不是中斷狀態；
這里要考慮在執行該if語句期間可能也執行了shutdownNow方法，shutdownNow方法會把狀態設置為STOP，回顧一下STOP狀態：
不能接受新任務，也不處理隊列中的任務，會中斷正在處理任務的線程。在線程池處於 RUNNING 或 SHUTDOWN 狀態時，調用 shutdownNow() 方法會使線程池進入到該狀態。
STOP狀態要中斷線程池中的所有線程，而這里使用Thread.interrupted()來判斷是否中斷是為了確保在RUNNING或者SHUTDOWN狀態時線程是非中斷狀態的，因為Thread.interrupted()方法會復位中斷的狀態。
總結一下runWorker方法的執行過程：
while循環不斷地通過getTask()方法獲取任務；
getTask()方法從阻塞隊列中取任務；
如果線程池正在停止，那么要保證當前線程是中斷狀態，否則要保證當前線程不是中斷狀態；
調用task.run()執行任務；
如果task為null則跳出循環，執行processWorkerExit()方法；
runWorker方法執行完畢，也代表着Worker中的run方法執行完畢，銷毀線程。
這里的beforeExecute方法和afterExecute方法在ThreadPoolExecutor類中是空的，留給子類來實現。
completedAbruptly變量來表示在執行任務過程中是否出現了異常，在processWorkerExit方法中會對該變量的值進行判斷。
getTask
getTask方法用來從阻塞隊列中取任務，代碼如下：
/**
* Performs blocking or timed wait for a task, depending on
* current configuration settings, or returns null if this worker
* must exit because of any of:
* 1. There are more than maximumPoolSize workers (due to
* a call to setMaximumPoolSize).
* 2. The pool is stopped.
* 3. The pool is shutdown and the queue is empty.
* 4. This worker timed out waiting for a task, and timed-out
* workers are subject to termination (that is,
* {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
* both before and after the timed wait, and if the queue is
* non-empty, this worker is not the last thread in the pool.
*
* @return task, or null if the worker must exit, in which case
* workerCount is decremented
*/
private Runnable getTask() {
// timeOut變量的值表示上次從阻塞隊列中取任務時是否超時
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
/*
* 如果線程池狀態rs >= SHUTDOWN，也就是非RUNNING狀態，再進行以下判斷：
* 1. rs >= STOP，線程池是否正在stop；
* 2. 阻塞隊列是否為空。
* 如果以上條件滿足，則將workerCount減1並返回null。
* 因為如果當前線程池狀態的值是SHUTDOWN或以上時，不允許再向阻塞隊列中添加任務。
*/
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
// timed變量用於判斷是否需要進行超時控制。
// allowCoreThreadTimeOut默認是false，也就是核心線程不允許進行超時；
// wc > corePoolSize，表示當前線程池中的線程數量大於核心線程數量；
// 對於超過核心線程數量的這些線程，需要進行超時控制
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
/*
* wc > maximumPoolSize的情況是因為可能在此方法執行階段同時執行了setMaximumPoolSize方法；
* timed && timedOut 如果為true，表示當前操作需要進行超時控制，並且上次從阻塞隊列中獲取任務發生了超時
* 接下來判斷，如果有效線程數量大於1，或者阻塞隊列是空的，那么嘗試將workerCount減1；
* 如果減1失敗，則返回重試。
* 如果wc == 1時，也就說明當前線程是線程池中唯一的一個線程了。
*/
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
//如果當前線程數大於corePoolSize，則會調用workQueue的poll方法獲取任務，超時時間是keepAliveTime
//如果超過keepAliveTime時長，poll返回了null，上邊提到的while循序就會退出，線程也就執行完了
/*
* 根據timed來判斷，如果為true，則通過阻塞隊列的poll方法進行超時控制，如果在keepAliveTime時間內沒有獲取到任務，則返回null；
* 否則通過take方法，如果這時隊列為空，則take方法會阻塞直到隊列不為空。
*/
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
//如果當前線程數小於corePoolSize，則會調用workQueue的take方法阻塞在當前
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
這里重要的地方是第二個if判斷，目的是控制線程池的有效線程數量。由上文中的分析可以知道，在執行execute方法時，如果當前線程池的線程數量超過了corePoolSize且小於maximumPoolSize，並且workQueue已滿時，則可以增加工作線程，但這時如果超時沒有獲取到任務，也就是timedOut為true的情況，說明workQueue已經為空了，也就說明了當前線程池中不需要那么多線程來執行任務了，可以把多於corePoolSize數量的線程銷毀掉，保持線程數量在corePoolSize即可。
什么時候會銷毀？當然是runWorker方法執行完之后，也就是Worker中的run方法執行完，由JVM自動回收。
getTask方法返回null時，在runWorker方法中會跳出while循環，然后會執行processWorkerExit方法。
processWorkerExit
private void processWorkerExit(Worker w, boolean completedAbruptly) {
// 如果completedAbruptly值為true，則說明線程執行時出現了異常，需要將workerCount減1；
// 如果線程執行時沒有出現異常，說明在getTask()方法中已經已經對workerCount進行了減1操作，這里就不必再減了。
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//統計完成的任務數
completedTaskCount += w.completedTasks;
// 從workers中移除，也就表示着從線程池中移除了一個工作線程
workers.remove(w);
} finally {
mainLock.unlock();
}
// 根據線程池狀態進行判斷是否結束線程池
tryTerminate();
int c = ctl.get();
/*
* 當線程池是RUNNING或SHUTDOWN狀態時，如果worker是異常結束，那么會直接addWorker；
* 如果allowCoreThreadTimeOut=true，並且等待隊列有任務，至少保留一個worker；
* 如果allowCoreThreadTimeOut=false，workerCount不少於corePoolSize。
*/
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
addWorker(null, false);
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
至此，processWorkerExit執行完之后，工作線程被銷毀，以上就是整個工作線程的生命周期，從execute方法開始，Worker使用ThreadFactory創建新的工作線程，runWorker通過getTask獲取任務，然后執行任務，如果getTask返回null，進入processWorkerExit方法，整個線程結束，如圖所示：
tryTerminate
tryTerminate方法根據線程池狀態進行判斷是否結束線程池，代碼如下：
final void tryTerminate() {
for (;;) {
int c = ctl.get();
/*
* 當前線程池的狀態為以下幾種情況時，直接返回：
* 1. RUNNING，因為還在運行中，不能停止；
* 2. TIDYING或TERMINATED，因為線程池中已經沒有正在運行的線程了；
* 3. SHUTDOWN並且等待隊列非空，這時要執行完workQueue中的task；
*/
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
// 如果線程數量不為0，則中斷一個空閑的工作線程，並返回
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 這里嘗試設置狀態為TIDYING，如果設置成功，則調用terminated方法
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
// terminated方法默認什么都不做，留給子類實現
terminated();
} finally {
// 設置狀態為TERMINATED
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
interruptIdleWorkers(ONLY_ONE);的作用是因為在getTask方法中執行workQueue.take()時，如果不執行中斷會一直阻塞。在下面介紹的shutdown方法中，會中斷所有空閑的工作線程，如果在執行shutdown時工作線程沒有空閑，然后又去調用了getTask方法，這時如果workQueue中沒有任務了，調用workQueue.take()時就會一直阻塞。所以每次在工作線程結束時調用tryTerminate方法來嘗試中斷一個空閑工作線程，避免在隊列為空時取任務一直阻塞的情況。
shutdown
shutdown方法要將線程池切換到SHUTDOWN狀態，並調用interruptIdleWorkers方法請求中斷所有空閑的worker，最后調用tryTerminate嘗試結束線程池。
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 安全策略判斷
checkShutdownAccess();
// 切換狀態為SHUTDOWN
advanceRunState(SHUTDOWN);
// 中斷空閑線程
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
// 嘗試結束線程池
tryTerminate();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
這里思考一個問題：在runWorker方法中，執行任務時對Worker對象w進行了lock操作，為什么要在執行任務的時候對每個工作線程都加鎖呢？
下面仔細分析一下：
在getTask方法中，如果這時線程池的狀態是SHUTDOWN並且workQueue為空，那么就應該返回null來結束這個工作線程，而使線程池進入SHUTDOWN狀態需要調用shutdown方法；
shutdown方法會調用interruptIdleWorkers來中斷空閑的線程，interruptIdleWorkers持有mainLock，會遍歷workers來逐個判斷工作線程是否空閑。但getTask方法中沒有mainLock；
在getTask中，如果判斷當前線程池狀態是RUNNING，並且阻塞隊列為空，那么會調用workQueue.take()進行阻塞；
如果在判斷當前線程池狀態是RUNNING后，這時調用了shutdown方法把狀態改為了SHUTDOWN，這時如果不進行中斷，那么當前的工作線程在調用了workQueue.take()后會一直阻塞而不會被銷毀，因為在SHUTDOWN狀態下不允許再有新的任務添加到workQueue中，這樣一來線程池永遠都關閉不了了；
由上可知，shutdown方法與getTask方法（從隊列中獲取任務時）存在競態條件；
解決這一問題就需要用到線程的中斷，也就是為什么要用interruptIdleWorkers方法。在調用workQueue.take()時，如果發現當前線程在執行之前或者執行期間是中斷狀態，則會拋出InterruptedException，解除阻塞的狀態；
但是要中斷工作線程，還要判斷工作線程是否是空閑的，如果工作線程正在處理任務，就不應該發生中斷；
所以Worker繼承自AQS，在工作線程處理任務時會進行lock，interruptIdleWorkers在進行中斷時會使用tryLock來判斷該工作線程是否正在處理任務，如果tryLock返回true，說明該工作線程當前未執行任務，這時才可以被中斷。
interruptIdleWorkers
private void interruptIdleWorkers() {
interruptIdleWorkers(false);
}
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
Thread t = w.thread;
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
interruptIdleWorkers遍歷workers中所有的工作線程，若線程沒有被中斷tryLock成功，就中斷該線程。
為什么需要持有mainLock？因為workers是HashSet類型的，不能保證線程安全。
shutdownNow
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;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
shutdownNow方法與shutdown方法類似，不同的地方在於：
設置狀態為STOP；
中斷所有工作線程，無論是否是空閑的；
取出阻塞隊列中沒有被執行的任務並返回。
shutdownNow方法執行完之后調用tryTerminate方法，該方法在上文已經分析過了，目的就是使線程池的狀態設置為TERMINATED。
線程池的監控
通過線程池提供的參數進行監控。線程池里有一些屬性在監控線程池的時候可以使用
getTaskCount：線程池已經執行的和未執行的任務總數；
getCompletedTaskCount：線程池已完成的任務數量，該值小於等於taskCount；
getLargestPoolSize：線程池曾經創建過的最大線程數量。通過這個數據可以知道線程池是否滿過，也就是達到了maximumPoolSize；
getPoolSize：線程池當前的線程數量；
getActiveCount：當前線程池中正在執行任務的線程數量。
通過這些方法，可以對線程池進行監控，在ThreadPoolExecutor類中提供了幾個空方法，如beforeExecute方法，afterExecute方法和terminated方法，可以擴展這些方法在執行前或執行后增加一些新的操作，例如統計線程池的執行任務的時間等，可以繼承自ThreadPoolExecutor來進行擴展。
小結
本文比較詳細的分析了線程池的工作流程，總體來說有如下幾個內容：
分析了線程的創建，任務的提交，狀態的轉換以及線程池的關閉；
這里通過execute方法來展開線程池的工作流程，execute方法通過corePoolSize，maximumPoolSize以及阻塞隊列的大小來判斷決定傳入的任務應該被立即執行，還是應該添加到阻塞隊列中，還是應該拒絕任務。
介紹了線程池關閉時的過程，也分析了shutdown方法與getTask方法存在競態條件；
在獲取任務時，要通過線程池的狀態來判斷應該結束工作線程還是阻塞線程等待新的任務，也解釋了為什么關閉線程池時要中斷工作線程以及為什么每一個worker都需要lock。
在向線程池提交任務時，除了execute方法，還有一個submit方法，submit方法會返回一個Future對象用於獲取返回值，有關Future和Callable請看Callable和Future詳解。
線程池優化策略
CPU密集型
說到優化，肯定離不開業務。比如我們的任務是計算密集型(以CPU計算為主)、多內存計算的，那么可能4c的機器開了4-8個線程，負載就打滿了。那么我們在設置最大線程數maximumPoolSize的時候，最好使用Runtime.availableProcessors方法獲取可用處理器的個數N，並設置maximumPoolSize=N+1。(額外+1的原因是當計算密集型線程偶爾由於頁缺失故障或其他原因而暫停時，這個“額外的”線程也能確保這段時間內的CPU始終周期不會被浪費)
IO密集型
像讀寫磁盤文件、讀寫數據庫、網絡請求等阻塞操作，執行IO操作時，CPU處於等待狀態，等待過程中操作系統會把CPU時間片分給其他線程。我們可以使用newCachedThreadPool。
newCachedThreadPool默認最大線程數為Integer.MAX_VALUE，keepAliveTime只有60s，隊列也是無界隊列。
這種就適合用於一些生命周期較短，密集而又頻繁的操作。
也可以參考newCachedThreadPool，根據實際情況（內存上線的控制很關鍵），適當縮小maximumPoolSize的大小，增加或減小keepAliveTime。
CPU/IO混合型任務
大多數任務並不是單一的計算型或IO型，而是IO伴隨計算兩者混合執行的任務——即使簡單的Http請求也會有請求的構造過程。
混合型任務要根據任務等待阻塞時間與CPU計算時間的比重來決定線程數量：
比如一個任務包含一次數據庫讀寫(0.1ms)，並在內存中對讀取的數據進行分組過濾等操作(5μs)，那么線程數應該為80左右(假設為4c的機器)。
阻塞的時間(waitTime)對計算的時間(computeTime)占比越大，則開放的線程數也應該越多。
小結
線程池的大小取決於任務的類型以及系統的特性，避免“過大”和“過小”兩種極端。線程池過大，大量的線程將在相對更少的CPU和有限的內存資源上競爭，這不僅影響並發性能，還會因過高的內存消耗導致OOM；線程池過小，將導致處理器得不到充分利用，降低吞吐率。
要想正確的設置線程池大小，需要了解部署的系統中有多少個CPU，多大的內存，提交的任務是計算密集型、IO密集型還是兩者兼有。
雖然線程池和JDBC連接池的目的都是對稀缺資源的重復利用，但通常一個應用只需要一個JDBC連接池，而線程池通常不止一個。如果一個系統要執行不同類型的任務，並且它們的行為差異較大，那么應該考慮使用多個線程池，使每個線程池可以根據各自的任務類型以及工作負載來調整。
————————————————
版權聲明：本文為CSDN博主「Deegue」的原創文章，遵循CC 4.0 BY-SA版權協議，轉載請附上原文出處鏈接及本聲明。
原文鏈接：https://blog.csdn.net/zyzzxycj/article/details/90290176