一、引言
線程池技術在網際網路技術使用如此廣泛，幾乎所有的後端技術面試官都要線上程池技術的使用和原理方面對小夥伴們進行 360° 的刁難。
作為一個在網際網路公司面一次拿一次 Offer 的面霸，打敗了無數競争對手，每次都隻能看到無數落寞的身影失望的離開，略感愧疚（請允許我使用一下誇張的修辭手法）。
于是在一個寂寞難耐的夜晚，暖男我痛定思痛，決定開始寫 《吊打面試官》 系列，希望能幫助各位讀者以後面試勢如破竹，對面試官進行 360° 的反擊，吊打問你的面試官，讓一同面試的同僚瞠目結舌，瘋狂收割大廠 Offer！
雖然現在是網際網路寒冬，但乾坤未定，你我皆是黑馬
二、使用
我想大部分人應該都使用過線程池，我們的 JDK 中也提供了一些包裝好的線程池使用，比如：
- newFixedThreadPool：傳回一個核心線程數為 nThreads 的線程池
- public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); } 複制代碼
- newSingleThreadExecutor：傳回一個核心線程數為 1 的線程池
- public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>())); } 複制代碼
- newCachedThreadPool：大同小異
- public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); } 複制代碼
通過上面 JDK 提供的我們可以發現一個共識，他們其實都是調用了 ThreadPoolExecutor 的構造方法來進行線程池的建立
這時候，我們不免有疑問，我們難道不可以直接使用 ThreadPoolExecutor 的構造方法去進行建立嘛
是的，阿裡巴巴Java開發手冊中明确指出，『不允許』使用Executors建立線程池
是以，我們在生産中，一般使用 ThreadPoolExecutor 的構造方法自定義去建立線程池，比如：
public class ThreadPoolTest {
public static void main(String[] args) {
ThreadPoolExecutor executor = new ThreadPoolExecutor(
2, // 核心線程數
5, // 最大線程數
200, // 非核心工作線程在阻塞隊列位置等待的時間
TimeUnit.SECONDS, // 非核心工作線程在阻塞隊列位置等待的機關
new LinkedBlockingQueue<>(), // 阻塞隊列，存放任務的地方
new ThreadPoolExecutor.AbortPolicy() // 拒絕政策：這裡有四種
);
for (int i = 0; i < 10; i++) {
MyTask task = new MyTask();
executor.execute(task);
}
// 關閉線程
executor.shutdown();
}
}
class MyTask implements Runnable {
@Override
public void run() {
System.out.println("我被執行了....");
}
}
複制代碼 三、源碼
整體的流程如下：
1、初始化
聊源碼不從初始化聊的，都是不講道理的
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), handler);
}
// 執行ThreadPoolExecutor的構造方法進行初始化
// corePoolSize: 核心線程數
// maximumPoolSize: 最大線程數
// keepAliveTime: 非核心工作線程在阻塞隊列位置等待的時間
// unit: 非核心工作線程在阻塞隊列位置等待的時間機關
// workQueue: 存放任務的阻塞隊列
// threadFactory: 線程工廠(生産線程的地方)
// RejectedExecutionHandler: 拒絕政策
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
// 核心線程數可以為0
// 最大線程數不為0
// 最大線程數 大于 核心線程數
// 等待時間大于等于0
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;
}
複制代碼 我們上面初始化的過程主要對入參做了一些校驗，然後将方法的入參賦予給成員變量
1.1 拒絕政策
1.1.1 AbortPolicy
簡單粗暴，直接抛出異常
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());
}
}
複制代碼 1.1.2 CallerRunsPolicy
目前拒絕政策會線上程池無法處理任務時，将任務交給調用者處理
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public CallerRunsPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
// 如果目前的
if (!e.isShutdown()) {
r.run();
}
}
}
複制代碼 1.1.3 DiscardOldestPolicy
如果目前的阻塞隊列滿了，彈出時間最久的
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public DiscardOldestPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
// 擷取阻塞隊列,彈出一個時間最久的
e.getQueue().poll();
// 執行目前的
e.execute(r);
}
}
}
複制代碼 1.1.4 DiscardPolicy
簡單粗暴，不做任何操作
public static class DiscardPolicy implements RejectedExecutionHandler {
public DiscardPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
複制代碼 1.1.5 自定義拒絕政策
自己寫的業務邏輯，可以将拒絕的任務放至資料庫等存儲，等後續在執行
public static class MyRejectedExecution implements RejectedExecutionHandler{
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
System.out.println("這是我自己的拒絕政策");
}
}
複制代碼 1.2 其餘變量
// 該數值代表兩個意思：
// 高3位表示目前線程池的狀态
// 低29位表示目前線程池工作線程的個數
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// COUNT_BITS = 29
private static final int COUNT_BITS = Integer.SIZE - 3;
// CAPACITY就是目前工作線程能記錄的工作線程的最大個數
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// 111：代表RUNNING狀态，RUNNING可以處理任務，并且處理阻塞隊列中的任務。
private static final int RUNNING = -1 << COUNT_BITS;
// 000：代表SHUTDOWN狀态，不會接收新任務，正在處理的任務正常進行，阻塞隊列的任務也會做完。
private static final int SHUTDOWN = 0 << COUNT_BITS;
// 001：代表STOP狀态，不會接收新任務，正在處理任務的線程會被中斷，阻塞隊列的任務一個不管。
private static final int STOP = 1 << COUNT_BITS;
// 010：代表TIDYING狀态，這個狀态是否SHUTDOWN或者STOP轉換過來的，代表目前線程池馬上關閉，就是過渡狀态。
private static final int TIDYING = 2 << COUNT_BITS;
// 011：代表TERMINATED狀态，這個狀态是TIDYING狀态轉換過來的，轉換過來隻需要執行一個terminated方法。
private static final int TERMINATED = 3 << COUNT_BITS;
// 基于&運算的特點，保證隻會拿到ctl高三位的值
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 基于&運算的特點，保證隻會拿到ctl低29位的值
private static int workerCountOf(int c) { return c & CAPACITY; }
複制代碼 線程池的狀态變化流程圖：
2、線程池的execute方法
- Step1：目前的線程池個數低于核心線程數，直接添加核心線程即可
- Step2：目前的線程池個數大于核心線程數，将任務添加至阻塞隊列中
- Step3：如果添加阻塞隊列失敗，則需要添加非核心線程數處理任務
- Step4：如果添加非核心線程數失敗（滿了），執行拒絕政策
public void execute(Runnable command) {
// 如果目前傳過來的任務是null,直接抛出異常即可
if (command == null)
throw new NullPointerException();
// 擷取目前的資料值
int c = ctl.get();
//==========================線程池第一階段：啟動核心線程數開始==================================================
// Step1:擷取ctl低29位的數值，與我們的核心線程數相比
if (workerCountOf(c) < corePoolSize) {
// Step2:添加一個核心線程
if (addWorker(command, true)){
return;
}
// 更新一下目前值
c = ctl.get();
}
//==========================線程池第一階段：啟動核心線程數結束==================================================
// 如果走到下面會有兩種情況：
// 1、核心線程數滿了,需要往阻塞隊列裡面扔任務
// 2、核心線程數滿了,阻塞隊列也滿了,執行拒絕政策
//==========================線程池第二階段：任務放至阻塞隊列開始==================================================
// 判斷目前的狀态是不是Running的狀态(RUNNING可以處理任務，并且處理阻塞隊列中的任務)
// 如果是Running的狀态,則可以将任務放至阻塞隊列中
// 這裡如果放阻塞隊列失敗了,證明阻塞隊列滿了
if (isRunning(c) && workQueue.offer(command)) {
// 再次更新數值
int recheck = ctl.get();
// 再次校驗目前的線程池狀态是不是Running
// 如果線程池狀态不是Running的話,需要删除掉剛剛放的任務
if (!isRunning(recheck) && remove(command)){
// 執行拒絕政策
reject(command);
}
// 如果到這裡,說明上面阻塞隊列中已經有資料了
// 如果線程池的個數為0的話,需要建立一個非核心工作線程去執行該任務
// 不能讓人家堵塞着
else if (workerCountOf(recheck) == 0){
addWorker(null, false);
}
}
//==========================線程池第二階段：任務放至阻塞隊列結束==================================================
// 如果走到這裡的邏輯，證明上面的邏輯沒走通，有以下兩種情況：
// 1、線程池的狀态不是Running
// 1.1 如果是這種情況,下面的添加非核心工作線程失敗執行拒絕政策,但這個并不是這個邏輯的重點
// 2、阻塞隊列添加任務失敗(阻塞隊列滿了)
// 2.1 這種情況才是我們需要關心的
// 2.2 阻塞隊列滿了,添加非核心工作線程
// 2.3 若添加非核心工作線程失敗,證明已經到達maximumPoolSize的限制,執行拒絕政策
//==========================線程池第三階段：啟動非核心線程數開始==================================================
// 添加一個非核心工作線程
else if (!addWorker(command, false))
// 工作隊列中添加任務失敗,執行拒絕政策
reject(command);
//==========================線程池第三階段：啟動非核心線程數結束==================================================
}
複制代碼 流程圖如下：
3、線程池的addWorker方法
3.1 校驗
- 校驗目前線程池的狀态
- 校驗目前線程池工作線程的個數（核心線程數、最大工作線程數）
private boolean addWorker(Runnable firstTask, boolean core) {
// 這裡主要是為了結束整個循環
retry:
for (;;) {
// 擷取目前線程池的數值(ctl)
int c = ctl.get();
// runStateOf:基于&運算的特點，保證隻會拿到ctl高三位的值
int rs = runStateOf(c);
//==========================線程池狀态判斷=============================================================
// rs >= SHUTDOWN:代表目前線程池狀态為：SHUTDOWN、STOP、TIDYING、TERMINATED,線程池狀态異常
// 但這裡SHUTDOWN狀态稍許不同(不會接收新任務，正在處理的任務正常進行，阻塞隊列的任務也會做完)
// 如果目前的狀态是SHUTDOWN狀态并且阻塞隊列任務不為空且新任務為空
// 需要新起一個非核心工作線程去執行任務
// 如果不是前面的,直接傳回false即可
if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty())){
return false;
}
//==========================工作線程個數判斷==========================================================
for (;;) {
// 擷取目前線程池中線程的個數
int wc = workerCountOf(c);
// 1、如果線程池線程的個數是否超過了工作線程的最大個數
// 2、core=true(核心線程)=false(工作線程)
// 2.1 根據目前core判斷建立的是核心線程數(corePoolSize)還是非核心線程數(maximumPoolSize)
if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)){
return false;
}
// 嘗試将線程池線程加一
if (compareAndIncrementWorkerCount(c)){
// CAS成功後，直接退出外層循環，代表可以執行添加工作線程操作了。
break retry;
}
// 擷取目前線程池的數值(ctl)
c = ctl.get();
// 擷取目前線程池的狀态
// 判斷目前線程池的狀态等不等于我們上面的rs
// 我們線程池的狀态被人更改了，需要重新跑整個for循環判斷邏輯
if (runStateOf(c) != rs){
continue retry;
}
}
}
// 省略下面的代碼
}
複制代碼 3.2 添加線程
{
// 省略校驗的步驟
// 兩個标記
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
// 将目前的任務封裝成Worker
w = new Worker(firstTask);
// 拿到目前Worker的線程
final Thread t = w.thread;
// 線程不為空
if (t != null) {
// 上鎖，保證線程安全(workers、largestPoolSize)
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 擷取線程池的狀态
int rs = runStateOf(ctl.get());
// 1、rs < SHUTDOWN：保證目前線程池的狀态一定是RUNNING狀态
// 2、rs == SHUTDOWN && firstTask == null：如果目前線程池是SHUTDOWN狀态且新任務為空
if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {
// 判斷線程是否還活着
if (t.isAlive())
throw new IllegalThreadStateException();
// private final HashSet<Worker> workers = new HashSet<Worker>();
// 添加到我們的work隊列中
workers.add(w);
// 擷取works的大小
int s = workers.size();
// largestPoolSize在記錄最大線程個數的記錄
// 如果目前的大小比最大的還要打，替換即可
if (s > largestPoolSize)
largestPoolSize = s;
// worker添加成功
workerAdded = true;
}
} finally {
// 解鎖
mainLock.unlock();
}
// 如果添加成功
if (workerAdded) {
// 啟動線程
t.start();
// 線程啟動标志位
workerStarted = true;
}
}
} finally {
// 如果線程沒有啟動成功,從workers集合中删除掉該worker
if (!workerStarted)
addWorkerFailed(w);
}
// 傳回線程是否啟動成功
return workerStarted;
}
// Worker的初始化
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
// 從線程工廠裡面拿一個線程出來
this.thread = getThreadFactory().newThread(this);
}
// 從workers集合中删除掉該worker
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (w != null)
workers.remove(w);
decrementWorkerCount();
tryTerminate();
} finally {
mainLock.unlock();
}
}
複制代碼 4、線程池的 worker 源碼
// Worker繼承了AQS，目的就是為了控制工作線程的中斷。
// Worker實作了Runnable，内部的Thread對象，在執行start時，必然要執行Worker中斷額一些操作
private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
private static final long serialVersionUID = 6138294804551838833L;
// 目前線程工廠建立的線程(也是執行任務使用的線程)
final Thread thread;
// 目前的第一個任務
Runnable firstTask;
// 記錄執行了多少個任務
volatile long completedTasks;
// 構造方法
Worker(Runnable firstTask) {
// 将State設定為-1，代表目前不允許中斷線程
setState(-1);
// 設定任務
this.firstTask = firstTask;
// 設定線程
this.thread = getThreadFactory().newThread(this);
}
// 線程啟動執行的方法
public void run() {
runWorker(this);
}
// =======================Worker管理中斷================================
// 目前方法是中斷工作線程時，執行的方法
void interruptIfStarted() {
Thread t;
// 隻有Worker中的state >= 0的時候，可以中斷工作線程
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
// 如果狀态正常，并且線程未中斷，這邊就中斷線程
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
複制代碼 5、線程池的 runWorker 方法
public void run() {
runWorker(this);
}
final void runWorker(Worker w) {
// 拿到目前的線程
Thread wt = Thread.currentThread();
// 拿到目前Worker的第一個任務(如果攜帶的話)
Runnable task = w.firstTask;
// 置空
w.firstTask = null;
// 解鎖
w.unlock();
boolean completedAbruptly = true;
try {
// 如果任務不等于空 或者 從阻塞隊列中拿到的任務不等于空
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) {
thrown = x; throw new Error(x);
} finally {
// 任務執行後一些操作：自己實作的鈎子
afterExecute(task, thrown);
}
} finally {
// 任務置空
task = null;
// 執行任務+1
w.completedTasks++;
// 解鎖
w.unlock();
}
}
completedAbruptly = false;
} finally {
// 删除線程的方法
processWorkerExit(w, completedAbruptly);
}
}
複制代碼 6、線程池的 getTask 方法
private Runnable getTask() {
// 逾時的标記
boolean timedOut = false;
// 死循環拿資料
for (;;) {
// 拿到目前的ctl
int c = ctl.get();
// 擷取其線程池狀态
int rs = runStateOf(c);
// 如果線程池狀态是STOP，沒有必要處理阻塞隊列任務，直接傳回null
// 如果線程池狀态是SHUTDOWN，并且阻塞隊列是空的，直接傳回null
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
// 線程池中的線程個數減一
decrementWorkerCount();
return null;
}
// 目前線程池中線程個數
int wc = workerCountOf(c);
// 這裡是個重點
// allowCoreThreadTimeOut:是否允許核心線程數逾時（開啟這個之後），核心線程數也會執行下面逾時的邏輯
// wc > corePoolSize:目前線程池中的線程個數大于核心線程數
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// wc > maximumPoolSize:基本不存在
// timed && timedOut:第一次肯定是失敗的(逾時标記為false)
if ((wc > maximumPoolSize || (timed && timedOut))
// 1、線程個數為1
// 2、阻塞隊列是空的
&& (wc > 1 || workQueue.isEmpty())) {
// 線程池的線程個數減一
if (compareAndDecrementWorkerCount(c)){
return null;
}
continue;
}
try {
// 根據我們前面的timed的值(目前線程池中的線程個數是否大于核心線程數)
// 如果大于,執行workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)帶有時間的等待,超過時間無任務，會傳回null
// 如果小于,執行workQueue.take(),死等任務，不會傳回null
Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
if (r != null){
return r;
}
// 到這裡，說明上面的等待逾時了
// 這裡要注意一下，如果這裡逾時後，我們上面 if ((wc > maximumPoolSize || (timed && timedOut)) 這個判斷要起作用了
// (timed && timedOut) true
// wc > 1 || workQueue.isEmpty():當線程大于1或者阻塞隊列無資料,直接傳回null,讓外部循環删除
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
複制代碼 7、線程池的 processWorkerExit 方法
private void processWorkerExit(Worker w, boolean completedAbruptly) {
// 正常退出的:completedAbruptly=false
// 不是正常退出的:completedAbruptly=true
if (completedAbruptly)
decrementWorkerCount();
// 加鎖——上鎖
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 将目前worker的執行任務數累加到線程池中
completedTaskCount += w.completedTasks;
// 線程池删除該工作線程
workers.remove(w);
} finally {
// 解鎖
mainLock.unlock();
}
tryTerminate();
// 擷取ctl的資料
int c = ctl.get();
// 這裡隻有SHUTDOWN、RUNNING會進入判斷
if (runStateLessThan(c, STOP)) {
// 正常退出的
if (!completedAbruptly) {
// 是否允許逾時
// 允許：0
// 不允許：核心線程數
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
// 如果min=0并且阻塞隊列不為空
// 将min設定成1
if (min == 0 && ! workQueue.isEmpty())
min = 1;
// 目前線程池的大小大于最小值，直接傳回即可
if (workerCountOf(c) >= min){
return;
}
}
// 如果沒有的話，說明線程池中沒有線程了,并且還有阻塞任務
// 隻能添加一個非核心線程去處理這些任務
addWorker(null, false);
}
}
複制代碼 8、線程池的關閉方法
8.1 shutdownNow 方法
- 将線程池狀态修改為Stop（不會接收新任務，正在處理任務的線程會被中斷，阻塞隊列的任務一個不管）
- 将線程池中的線程全部中斷
- 删除目前線程池所有的工作線程
- 将線程池的狀态從：Stop --> TIDYING --> TERMINATED，正式标記線程池的結束（喚醒一下等待的主線程）
public List<Runnable> shutdownNow() {
// 聲明傳回結果
List<Runnable> tasks;
// 加鎖
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
// 将線程池狀态修改為STOP
advanceRunState(STOP);
// 将線程池中的線程全部中斷
interruptWorkers();
// 删除目前所有的工作線程
tasks = drainQueue();
} finally {
// 解鎖
mainLock.unlock();
}
// 檢視目前線程池是否可以變為TERMINATED狀态
// 從 Stop 狀态修改為 TIDYING，在修改為 TERMINATED
tryTerminate();
return tasks;
}
// targetState = STOP
// 作用:将目前線程池的狀态修改為Stop
private void advanceRunState(int targetState) {
// 進來直接死循環
for (;;) {
// 拿到目前的ctl
int c = ctl.get();
// runStateAtLeast(c, targetState):目前的c是不是大于STOP(如果大于Stop的話，說明線程池狀态已經G了
// 基于CAS,将ctl從c修改為Stop狀态，不修改工作線程個數，僅僅将狀态修改為Stop
// 如果可以修改成功,直接退出即可
if (runStateAtLeast(c, targetState) || ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
break;
}
}
// 将線程池中的線程全部中斷
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>();
// 清空阻塞隊列并将資料放入taskList中
q.drainTo(taskList);
// 校驗目前的資料是夠真的清空
if (!q.isEmpty()) {
// 如果确實有遺漏的,畢竟這哥們也沒上鎖
// 手動的将線程從workQueue删除掉并且放到taskList中
for (Runnable r : q.toArray(new Runnable[0])) {
if (q.remove(r))
taskList.add(r);
}
}
// 最終傳回即可
return taskList;
}
final void tryTerminate() {
for (;;) {
// 拿到ctl
int c = ctl.get();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// CAS将目前的ctl設定成TIDYING
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
// 該方法是一個鈎子函數，我們自己定義，線上程池銷毀之前做最後的處理
terminated();
} finally {
// 将ctl設定成TERMINATED标志着線程池的正式結束
ctl.set(ctlOf(TERMINATED, 0));
// 線程池提供了一個方法，主線程在送出任務到線程池後，是可以繼續做其他操作的。
// 咱們也可以讓主線程送出任務後，等待線程池處理完畢，再做後續操作
// 這裡線程池涼涼後，要喚醒哪些調用了awaitTermination方法的線程
// 簡單來說，當時等待線程池傳回的主線程，由于線程池已經銷毀了，他們也必須要喚醒
termination.signalAll();
}
return;
}
} finally {
// 解鎖
mainLock.unlock();
}
}
}
複制代碼 8.2 shutdown 方法
public void shutdown() {
// 加鎖
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
// 将線程池狀态修改為SHUTDOWN
advanceRunState(SHUTDOWN);
// 将線程池中的線程全部中斷
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
// 檢視目前線程池是否可以變為TERMINATED狀态
// 從 SHUTDOWN 狀态修改為 TIDYING，在修改為 TERMINATED
tryTerminate();
}
複制代碼 四、流程圖
原文連結：https://juejin.cn/post/7230381710599536700