Java併發包原始碼學習系列：執行緒池ThreadPoolExecutor原始碼解析
阿新 • • 發佈：2021-02-11
[toc]
系列傳送門：
- [Java併發包原始碼學習系列：AbstractQueuedSynchronizer](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/112254373)
- [Java併發包原始碼學習系列：CLH同步佇列及同步資源獲取與釋放](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/112301359)
- [Java併發包原始碼學習系列：AQS共享式與獨佔式獲取與釋放資源的區別](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/112386838)
- [Java併發包原始碼學習系列：ReentrantLock可重入獨佔鎖詳解](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/112454874)
- [Java併發包原始碼學習系列：ReentrantReadWriteLock讀寫鎖解析](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/112689635)
- [Java併發包原始碼學習系列：詳解Condition條件佇列、signal和await](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/112727669)
- [Java併發包原始碼學習系列：掛起與喚醒執行緒LockSupport工具類](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/112757098)
- [Java併發包原始碼學習系列：JDK1.8的ConcurrentHashMap原始碼解析](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/113059783)
- [Java併發包原始碼學習系列：阻塞佇列BlockingQueue及實現原理分析](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/113186979)
- [Java併發包原始碼學習系列：阻塞佇列實現之ArrayBlockingQueue原始碼解析](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/113252384)
- [Java併發包原始碼學習系列：阻塞佇列實現之LinkedBlockingQueue原始碼解析](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/113329416)
- [Java併發包原始碼學習系列：阻塞佇列實現之PriorityBlockingQueue原始碼解析](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/113358710)
- [Java併發包原始碼學習系列：阻塞佇列實現之DelayQueue原始碼解析](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/113440013)
- [Java併發包原始碼學習系列：阻塞佇列實現之SynchronousQueue原始碼解析](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/113528143)
- [Java併發包原始碼學習系列：阻塞佇列實現之LinkedTransferQueue原始碼解析](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/113576816)
- [Java併發包原始碼學習系列：阻塞佇列實現之LinkedBlockingDeque原始碼解析](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/113620915)
- [Java併發包原始碼學習系列：基於CAS非阻塞併發佇列ConcurrentLinkedQueue原始碼解析](https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/113728300)
## ThreadPoolExecutor概述
### 執行緒池解決的優點
1. 當執行大量非同步任務時執行緒池能夠提供較好的效能，因為執行緒池中的執行緒是可複用的，不需要每次執行非同步任務時都建立和銷燬執行緒。
2. 提供資源限制和管理的手段，比如可以限制執行緒的個數，動態新增執行緒等等。
### 執行緒池處理流程
ThreadPoolExecutor執行execute時，流程如下：
1. 如果當前執行的執行緒少於corePoolSize，則建立新執行緒來執行任務，這裡需要加全域性鎖。
2. 如果執行的執行緒數>=corePoolSize，則將任務加入BlockingQueue。
3. 如果此時BlockingQueue已滿，則建立新的執行緒來處理任務，這裡也需要加全域性鎖。
4. 如果建立新執行緒將使當前執行的執行緒超出maximumPoolSize，則按照拒絕策略拒絕任務。
當然啦，這篇文章意在從原始碼角度學習執行緒池這些核心步驟的具體實現啦，執行緒池概念性的東西，可以參考一些其他的部落格：
- [【Java併發程式設計】執行緒池相關知識點整理](https://tqbx.gitee.io/javablog/#/%E9%9D%A2%E8%AF%95%E7%9B%B8%E5%85%B3/%E5%B9%B6%E5%8F%91/%E7%BA%BF%E7%A8%8B%E6%B1%A0%E7%9A%84%E7%9F%A5%E8%AF%86%E7%82%B9)
- [閃客sun : 圖解 | 原來這就是執行緒池](https://www.cnblogs.com/flashsun/p/14368520.html)
### 建立執行緒池
建立執行緒池有幾種方法，一種是使用Executors工具類快速建立內建的幾種執行緒池，也可以自定義。
一、通過Executor框架的工具類Executors可以建立三種類型的ThreadPoolExecutor。
二、使用ThreadPoolExecutor的各種構造方法。
《阿里巴巴 Java 開發手冊》中：**強制執行緒池不允許使用 Executors 去建立，而是通過 ThreadPoolExecutor 的方式**，這樣的處理方式讓寫的同學更加明確執行緒池的執行規則，規避資源耗盡的風險
> Executors 返回執行緒池物件的弊端如下：
>
> - **FixedThreadPool 和 SingleThreadExecutor** ： 允許請求的佇列長度為 Integer.MAX_VALUE ，可能堆積大量的請求，從而導致 OOM。
> - **CachedThreadPool 和 ScheduledThreadPool** ： 允許建立的執行緒數量為 Integer.MAX_VALUE ，可能會建立大量執行緒，從而導致 OOM。
本篇的重點就是這個ThreadPoolExecutor。
## 重要常量及欄位
```java
public class ThreadPoolExecutor extends AbstractExecutorService {
// 原子的Integer變數ctl，用於記錄執行緒池狀態【高3位】和執行緒池中執行緒個數【低29位】，這裡假設Integer是32位的
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 其實並不是每個平臺的Integer二進位制都是32位的，實際上是，二進位制位-3代表執行緒個數
private static final int COUNT_BITS = Integer.SIZE - 3;
// 執行緒最大個數【約5億】 低COUNT_BITS都是1 000 11111111111111111111111111111
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// runState is stored in the high-order bits
// 111 00000000000000000000000000000 高3位是 111
private static final int RUNNING = -1 << COUNT_BITS;
// 000 00000000000000000000000000000 高3位是 000
private static final int SHUTDOWN = 0 << COUNT_BITS;
// 001 00000000000000000000000000000 高3位是 001
private static final int STOP = 1 << COUNT_BITS;
// 010 00000000000000000000000000000 高3位是 110
private static final int TIDYING = 2 << COUNT_BITS;
// 011 00000000000000000000000000000 高3位是 011
private static final int TERMINATED = 3 << COUNT_BITS;
// Packing and unpacking ctl
// 獲取高3位的執行狀態
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 獲取低29位的執行緒個數
private static int workerCountOf(int c) { return c & CAPACITY; }
// 通過RunState和WorkCount計算ctl的值
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;
}
// 只有RUNNING 是小於SHUTDOWN的
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
// ...
// 阻塞佇列
private final BlockingQueue workQueue;
// 獨佔鎖 同步保證
private final ReentrantLock mainLock = new ReentrantLock();
// 存放 執行緒池中的工作執行緒
private final HashSet workers = new HashSet();
// 條件佇列，執行緒呼叫awaitTermination時存放阻塞的執行緒
private final Condition termination = mainLock.newCondition();
// ...
// 繼承AQS和Runnable，任務執行緒
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{ /*.. */}
}
```
- ThreadPoolExecutor通過AtomicInteger類的變數ctl記錄**執行緒池狀態**和**執行緒池中執行緒個數**，這裡以Integer為32為例。
- 高3位表示執行緒池的狀態，低29位表示執行緒個數，分別通過`runStateOf`和`workerCountOf`計算。
## 執行緒池的五種狀態及轉換
- **執行緒池的狀態**有五種，他們提供了執行緒池宣告週期的控制：
- RUNNING：能夠接收新任務，並且處理阻塞佇列裡的任務。
- SHUTDOWN：拒絕新任務，但會處理阻塞佇列裡的任務。
- STOP：拒絕新任務，並且拋棄阻塞佇列裡的任務，同時會中斷正在處理的任務。
- TIDYING：所有任務都執行完後當前執行緒池workerCount為0，將呼叫terminated()這個鉤子方法。
- TERMINATED：終止狀態。terminated方法呼叫完成。
- 執行緒池的狀態是有規律的，保證單調遞增，但是不一定每個狀態都會經歷，比如有以下幾種轉換：
- RUNNING -> SHUTDOWN：可能是顯式呼叫了`shutdown()`方法，也可能在`finalize()`裡隱式呼叫。
- RUNNING或SHUTDOWN -> STOP：呼叫了`shutdownNow()`方法。
- SHUTDOWN -> TIDYING：佇列和執行緒池都為空的時候。
- STOP -> TIDYING：執行緒池為空的時候。
- TIDYING -> TERMINATED：鉤子方法`terminated()`呼叫完成的時候。
> 由於awaitTermination()方法而阻塞在條件佇列中的執行緒將會線上程池TERMINATED的時候返回。
## ThreadPoolExecutor構造引數及引數意義
ThreadPoolExecutor方法的構造引數有很多，我們看看最長的那個就可以了：
```java
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue 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.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
```
- `corePoolSize`：核心執行緒數定義了**最小可以同時執行的執行緒數量**。
- `maximumPoolSize`：**當佇列中存放的任務達到佇列容量的時候**，當前可以同時執行的執行緒數量變為**最大執行緒數**。【如果使用的無界佇列，這個引數就沒啥效果】
- `workQueue`: 阻塞佇列，當新任務來的時候會先判斷當前執行的執行緒數量是否達到核心執行緒數，**如果達到核心執行緒數的話，新任務就會被存放在佇列中**。
- `keepAliveTime`:當執行緒池中的執行緒數量大於 `corePoolSize` 的時候，如果這時沒有新的任務提交，核心執行緒外的執行緒不會立即銷燬，而是會等待，直到等待的時間超過了 `keepAliveTime`才會被回收銷燬。
- `unit`：`keepAliveTime` 的時間單位。
- `threadFactory`：用於設定建立執行緒的工廠，可以通過執行緒工廠給每個創建出來的執行緒設定更有意義的名字，預設使用Executors的靜態內部類`DefaultThreadFactory`。
- `handler`：飽和策略，當前同時執行的執行緒數量達到最大執行緒數量【`maximumPoolSize`】並且佇列也已經被放滿時，執行飽和策略。
關於各個引數的意義，強烈推薦這篇部落格：[閃客sun : 圖解 | 原來這就是執行緒池](https://www.cnblogs.com/flashsun/p/14368520.html)
## Work類
```java
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
private static final long serialVersionUID = 6138294804551838833L;
/** 具體執行任務的執行緒 */
final Thread thread;
/** 執行的第一個任務 */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
/** 執行緒啟動時，執行runWorker方法 */
public void 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;
}
// 不可重入的，state = 1表示已獲取
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
// state = 0 表示鎖未被獲取
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(); }
// 如果執行緒啟動，則中斷執行緒 state只有初始化的時候才是-1，其他的時間都是滿足> shutdownNow() {
List tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 檢查許可權
checkShutdownAccess();
// 設定為STOP
advanceRunState(STOP);
// 設定中斷標誌
interruptWorkers();
// 將佇列任務移到tasks中
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
```
### void interruptWorkers()
```java
private void interruptWorkers() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers)
// 如果執行緒啟動，則中斷執行緒【正在執行 + 空閒的所有執行緒都會被中斷】
w.interruptIfStarted();
} finally {
mainLock.unlock();
}
}
```
## boolean awaitTermination(timeout, unit)
當該方法被呼叫時，當前執行緒會被阻塞，直到超時時間到了，返回false。或者執行緒池狀態為TERMINATED時，返回true。
```java
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (;;) {
// 執行緒池狀態為TERMINATED 返回true
if (runStateAtLeast(ctl.get(), TERMINATED))
return true;
// 超時了， 返回false
if (nanos <= 0)
return false;
nanos = termination.awaitNanos(nanos);
}
} finally {
mainLock.unlock();
}
}
```
## 參考閱讀
- 《Java併發程式設計之美》
- 《Java併發程式設計的藝術》
- [【Java併發程式設計】執行緒池相關知識點整理](https://tqbx.gitee.io/javablog/#/%E9%9D%A2%E8%AF%95%E7%9B%B8%E5%85%B3/%E5%B9%B6%E5%8F%91/%E7%BA%BF%E7%A8%8B%E6%B1%A0%E7%9A%84%E7%9F%A5%E8%AF%86%E7%82%B9)
- [閃客sun : 圖解 | 原來這就是執行緒池](https://www.cnblogs.com/flashsun/p/143685