目錄
- 1 帶著問題去閱讀
- 1.1 執行緒池的執行緒復用原理
- 1.2 執行緒池如何管理執行緒
- 1.3 執行緒池配置的重要引數
- 1.4 shutdown()和shutdownNow()區別
- 1.5 執行緒池中的兩個鎖
- 2 原始碼分析程序中的困惑及解惑
- 3 原始碼分析
- 3.1 類繼承關系
- 3.2 類的常量/成員變數
- 3.3 成員變數訪問方法
- 3.4 建構式
- 3.5 靜態內部類Worker
- 3.5.1 Worker繼承關系
- 3.5.2 Worker原始碼分析
- 4 重要方法詳解
- 4.1 execute()方法
- 4.2 addWorker()方法
- 4.3 runWorker()方法
- 4.4 getTask()方法
- 4.5 processWorkerExit()方法
- 4.6 tryTerminate()方法
- 4.7 interruptIdleWorker()方法
- 4.8 shutdown()方法
- 4.9 shutdownNow()方法
- 4.10 isShutdown()方法
- 4.11 prestartCoreThread()方法
- 4.12 prestartAllCoreThreads()方法
1 帶著問題去閱讀
1.1 執行緒池的執行緒復用原理
用戶每次呼叫execute()來提交一個任務，然后任務包裝成Worker物件，并且啟動一個worker執行緒來執行任務（任務可能會被先加入佇列），只要任務佇列不為空且worker執行緒沒有被中斷，執行緒的run()方法通過一個while回圈，不斷去佇列獲取任務并執行，而不會進入到run()方法底部，while回圈是執行緒復用的關鍵
1.2 執行緒池如何管理執行緒
首先定義兩個說明：
-
關于獲取任務超時，會依賴以下條件:
--1、開啟核心執行緒超時設定 或 執行緒池執行緒數大于核心執行緒數
--2、符合1，且從workqueue獲取任務超時，（如果不符合1，則以阻塞方式獲取任務，不會超時） -
執行緒池最小保留執行緒數：
--1、如果沒有開啟核心執行緒超時配置，則至少保留corePoolSize個執行緒
--2、如果開啟核心執行緒超時并且當前佇列里面還有任務，只需保留1個執行緒
將執行緒池的生命周期分為三個階段：創建階段、運行期間、終止階段，
一、創建階段
- 當執行緒池執行緒數（ctl低位）少于核心執行緒數（corePoolSize），創建新執行緒執行任務
- 當執行緒池執行緒數大于等于核心執行緒數，且任務佇列未滿時，將新任務放入到任務佇列中，不創建執行緒
- 當執行緒池執行緒數大于等于核心執行緒數（maximumPoolSize），且任務佇列已滿
--如果作業執行緒數少于最大執行緒數，則創建新執行緒執行任務
--如果作業執行緒數等于最大執行緒數，則拋出例外，拒絕新任務進入
二、運行期間
1、執行緒啟動后，將一直回圈獲取任務并執行，只有當獲取任務超時，或者執行緒池被終止，才會結束，
2、如果獲取任務超時，那么Worker執行緒自然結束，此時執行緒池減少了1個執行緒，
3、在執行緒結束后，執行緒池會檢查：1、執行緒池執行緒數<最少保留執行緒數 2、任務執行例外結束，如果符合，執行緒池會自動補充1個Worker
三、終止階段
呼叫shutdown()和shutdownNow()都導致執行緒池執行緒數減少，
1、shutdown()方式終止執行緒池：
--停止提交新的任務，已在佇列的任務會繼續執行，并且中斷空閑的Worker執行緒(Work.state從0->1成功)，執行緒池狀態變為SHUTDOWN
2、shutdownNow()方式終止執行緒池：
--關閉執行緒池，不再接受新的任務，中斷已經啟動的Worker執行緒（Work.state>0），執行緒池狀態改為STOP
執行緒池創建執行緒及處理任務程序：
梳理一下大概流程：
- 用戶執行緒呼叫execute()提交Runnable任務
- execute()呼叫addWork()將任務提交給執行緒池處理：如果有可用的核心執行緒，則提交給核心執行緒處理，反則，將任務先添加到任務佇列（workQueen）中，
- addWorker()方法將啟動一個worker執行緒，呼叫runWorker()來處理任務，
- runWorker()方法將回圈獲取任務，并運行任務的run()方法來執行真正的業務，如果是以核心執行緒提交任務，則優先處理該任務，否則，回圈呼叫getTask()來獲取任務
- getTask()方法，從任務佇列（workQueen）取出任務，并回傳，
- getTask()沒有拿到任務，則執行執行緒結束processWorkerExit()
執行緒池創建階段：
1.3 執行緒池配置的重要引數
- ctl：存盤執行緒池狀態以及執行緒數
- corePoolSize、maximumPoolSize、keepAliveTime、workQueue 參照下面的原始碼分析說明
- allowCoreThreadTimeOut：是否開啟核心執行緒超時，默認false，不在建構式設定，需要呼叫方法設定
- HashSet
workers：執行緒池終止時會從該集合找執行緒來中斷，原始碼分析有說明
1.4 shutdown()和shutdownNow()區別
- shutdown() ：關閉執行緒池，不再接受新的任務，已提交執行的任務繼續執行；中斷所有空閑執行緒；將執行緒池狀態改為SHUTDOWN
- ShutDownNow()：關閉執行緒池，不再接受新的任務，中斷已經啟動的Worker執行緒；將執行緒池狀態改為STOP；回傳未完成的任務佇列
1.5 執行緒池中的兩個鎖
- mainLock主鎖是可重入的鎖，用來同步更新的成員變數
- Worker內部實作了一個鎖，它是不可重入的，在shutdown()場景中，通過tryLock確保不會中斷還沒有開始執行或者還在執行中的worker執行緒，
2 原始碼分析程序中的困惑及解惑
---什么情況任務會提交失敗？
同時符合以下條件，任務才會被提交：
- 執行緒池狀態等于RUNNING狀態；
- 如果任務佇列已經滿了，并且執行緒池執行緒數 少于 配置的執行緒池最大執行緒數（maximumPoolSize） 且小于執行緒池的最大支持執行緒數(CAPACITY)時，（如果佇列沒滿，任務將會先加入到佇列中）
特別說明：特殊情況會創建任務為空的Worker執行緒來幫助佇列中的任務跑完
---核心執行緒數的意義？從測驗結果看，他決定了作業執行緒最大并發數，但未代碼驗證
- 核心執行緒數決定提交任務什么時候會被放入到佇列中：即執行緒池執行緒數>=核心執行緒數時，
- 核心執行緒數大小跟并發執行執行緒（任務）無關，也就是，它不決定作業執行緒最大并發數
- 核心執行緒數可以動態修改，（如果增大了，可能會馬上創建新的Worker執行緒）
---執行緒池狀態不是RUNNING,或者往workQueue添加worker失敗，這是為什么還要提交任務
以下情況會創建任務為空的Worker執行緒來執行佇列中的任務
- 當前執行緒池狀態為shutdown，但是任務佇列不為空，這時創建Worker執行緒來幫助執行佇列的任務
- 當前執行緒池狀態為running， 任務添加到佇列后，接著執行緒池被關閉，并且從佇列移除該任務失敗，并且執行緒池執行緒數為0，這時創建Worker執行緒來確保剛提交的任務有機會執行，
---為什么runWorker()方法在執行任務前后加鎖，但是執行緒依然能夠并發？
- worker執行緒是通過創建Worker物件來創建的，在addWorke()的while回圈創建了多個Worker物件，每個Worker物件都有自己的鎖，Worker執行緒通過runWorker()訪問的是當前物件的鎖，因此Worker執行緒能夠并發；
- 鎖的意義是限制不能中斷執行中的任務，因為主執行緒呼叫shutdown()和shutdownNow()方法時，會遍歷WorkerSet的Worker物件，呼叫tryLock()，這時主執行緒和Worker執行緒競爭同一個鎖，
3 原始碼分析
3.1 類繼承關系
- Executo介面：專門提交任務，只有一個execute()方法，Executor 提供了一種將任務的提交和任務的執行兩個操作進行解耦的思路：客戶端無需關注執行任務的執行緒是如何創建、運行和回收的，只需要將任務的執行邏輯包裝為一個 Runnable 物件傳遞進來即可，由 Executor 的實作類自己來完成最復雜的執行邏輯
- ExecutorService介面：繼承了Executor，擴展執行任務的能力，例如：獲取任務的執行結果、取消任務等功能；提供了關閉執行緒池、停止執行緒池，以及阻塞等待執行緒池完全終止的方法,需要ThreadPoolExecutor實作
- AbstractExecutorServic類：實作了 ExecutorService ，是上層的抽象類，負責將任務的執行流程串聯起來，從而使得下層的實作類 ThreadPoolExecutor只需要實作一個執行任務的方法即可
- ThreadPoolExecutor：可以看做是基于生產者-消費者模式的一種服務，內部維護的多個執行緒相當于消費者，提交的任務相當于產品，提交任務的外部就相當于生產者
3.2 類的常量/成員變數
//--------------------------常量部分------------------------
// 常量29，用在移位計算Integer.SIZE=32）
private static final int COUNT_BITS = Integer.SIZE - 3; //29
// 最大支持執行緒數 2^29-1：000 11111111111111111...
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// 以下為執行緒池的四個狀態，用32位中的前三位表示
// 011 terminated() 方法執行完成后,執行緒池的狀態會轉為TERMINATED.
private static final int TERMINATED = 3 << COUNT_BITS;
// 010 所有任務都銷毀了,workCount=0的時候,執行緒池的裝填在轉換為TIDYING是,會執行鉤子方法terminated()
private static final int TIDYING = 2 << COUNT_BITS; //翻譯為整理
// 001 拒絕新的任務提交,清空在佇列中的任務
private static final int STOP = 1 << COUNT_BITS;
// 000 拒絕新的任務提交,會將佇列中的任務執行完,正在執行的任務繼續執行.
private static final int SHUTDOWN = 0 << COUNT_BITS;
// 111 00000 00000000 00000000 00000000 執行緒運行中 【running狀態值為負數最小】
private static final int RUNNING = -1 << COUNT_BITS; //執行緒池的默認狀態
//------------------------變數部分------------------------
// ctl存盤執行緒池狀態和執行緒池大小，那么用前3位表示執行緒池狀態，后29位表示：執行緒池大小，即執行緒池執行緒數
//執行緒池狀態初始值為RUNNING
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//任務佇列
//保存不能馬上執行的Runnable任務，
//執行shutdownNow()時，會回傳還在佇列的任務
private final BlockingQueue<Runnable> workQueue;
// 主鎖，對workers、largestPoolSize、completedTaskCount的訪問都必須先獲取該鎖
private final ReentrantLock mainLock = new ReentrantLock();
// 包含池中的所有作業執行緒的集合，持有mainLock訪問
// 創建Worker時，添加到集合
// 執行緒結束時，從集合移除
// 呼叫shutdown()時，從該集合中找到空閑執行緒并中斷
// 呼叫shutdownNow()時，從該集合中找到已啟動的執行緒并中斷
private final HashSet<Worker> workers = new HashSet<Worker>();
// 執行緒通信手段, 用于支持awaitTermination方法：等待所有任務完成,并支持設定超時時間,回傳值代表是不是超時.
private final Condition termination = mainLock.newCondition();
// 記錄workers歷史以來的最大值，持有mainLock訪問
// 每次增加worker的時候,都會判斷當前workers.size()是否大于最大值，大于則更新
// 用于執行緒池監控的，作為重要指標
private int largestPoolSize;
// 計數所有已完成任務,持有mainLock訪問
// 每個worker都有一個自己的成員變數 completedTasks 來記錄當前 worker 執行的任務次數, 當前線worker作業執行緒終止的時候, 才會將worker中的completedTasks的數量加入到 completedTaskCount 指標中.
private long completedTaskCount;
// 執行緒工廠
private volatile ThreadFactory threadFactory;
// 拒絕策略,默認四種AbortPolicy、CallerRunsPolicy、DiscardPolicy、DiscardOldestPolicy,建議自己實作,增加監控指標
private volatile RejectedExecutionHandler handler;
// keepAliveTime和allowCoreThreadTimeOut 是關于執行緒空閑是否會被銷毀的配置
// 關于空閑的說明：
// 1、執行緒池在沒有關閉之前，會一直向任務佇列（workqueue）獲取任務執行，如果任務佇列是空的，在新任務提交上來之前，就會產生一個等待時間，期間，執行緒處于空閑狀態
// 2、向任務佇列獲取任務用：workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)，表示阻塞式獲取元素，等待超時，則終止等待并回傳false，通過判斷poll()方法是true/falle來判定執行緒是否超時
// 獲取任務的等待時間 ，以下兩種情況會使用到該值
//1、如果啟用allowCoreThreadTimeOut，那表示核心執行緒的空閑時間
// 2、當執行緒池內執行緒數超過corePoolSize，表示執行緒獲取任務的等待時間
private volatile long keepAliveTime;
// 核心執行緒是否開啟超時
// false：表示核心執行緒一旦啟動，會一直運行，直至關閉執行緒池，默認該值
// true：表示核心執行緒處于空閑且時間超過keepAliveTime，核心執行緒結束后，將不再創建新執行緒
// （默認的建構式沒有設定這個屬性，需要手工呼叫allowCoreThreadTimeOut()方法來設定)
private volatile boolean allowCoreThreadTimeOut;
//核心執行緒數量
//核心執行緒是指：執行緒會一直存活在執行緒池中，不會被主動銷毀【如果核心執行緒開啟超時，有可能被被銷毀】，
private volatile int corePoolSize;
// 配置的執行緒池最大執行緒數
private volatile int maximumPoolSize;
// 默認拒絕策略 AbortPolicy
private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
// 安全控制訪問（主要用于shutdown和 shutdownNow方法
private static final RuntimePermission shutdownPerm = new RuntimePermission("modifyThread");
// 在threadPoolExecutor初始化的時候賦值,acc物件是指當前呼叫背景關系的快照，其中包括當前執行緒繼承的AccessControlContext和任何有限的特權范圍，使得可以在稍后的某個時間點(可能在另一個執行緒中)檢查此背景關系，
private final AccessControlContext acc;
3.3 成員變數訪問方法
// 獲取當前執行緒池的狀態(前3位)
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 獲取當前執行緒池中執行緒數(后29位)
private static int workerCountOf(int c){ return c & CAPACITY; }
// 更新狀態和數量
private static int ctlOf(int rs, int wc) { return rs | wc; }
// 小于判斷C是不是小于S,比如runStateLessThan(var,STOP),那var就只有可能是(RUNNING,SHUTDOWN)
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
// 是不是C >= S
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
// 判斷狀態是不是RUNNING
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
關于-1<<29說明：
-1 << COUNT_BITS
這里是-1往左移29位，稍微有點不一樣，-1的話需要我們自己算出補碼來
-1的原碼
10000000 00000000 00000000 00000001
-1的反碼，負數的反碼是將原碼除符號位以外全部取反
11111111 11111111 11111111 11111110
-1的補碼，負數的補碼就是將反碼+1
11111111 11111111 11111111 11111111
關鍵了，往左移29位，所以高3位全是1就是RUNNING狀態
111 00000 00000000 00000000 00000000
3.4 建構式
//corePoolSize、maximumPoolSize、keepAliveTime、unit、workQueue 這五個引數必須指定
//最多參建構式
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> 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;//默認使用SynchronousQueue<Runnable>
this.keepAliveTime = unit.toNanos(keepAliveTime); //默認60S
this.threadFactory = threadFactory; //默認使用DefaultThreadFactory
this.handler = handler;
}
建構式總結：
初始化：corePoolSize（核心執行緒池大小）、maximumPoolSize（執行緒池容納最大執行緒數）、workQueue（任務佇列）、threadFactory（執行緒工廠）、keepAliveTime（空閑執行緒存活時長）、handler（拒絕策略）AccessControlContext
3.5 靜態內部類Worker
3.5.1 Worker繼承關系
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
}
- --Worker繼承于AbstractQueuedSynchronizer
Worker繼承于AQS 為的就是自定義實作不可重入的特性（所以沒有使用 synchronized 或者 ReentrantLock）來輔助判斷執行緒是否處于執行任務的狀態：在開始執行任務前進行加鎖，在任務執行結束后解鎖，以便在后續通過判斷 Worker 是否處于鎖定狀態來得知其是否處于執行階段
- -- Worker實作Runnable介面
Worker實作Runnable介面，執行緒是通過getThreadFactory().newThread(this) 來創建的，即將 Worker 本身作為構造引數傳給 Thread 進行初始化，所以在 thread 啟動的時候 Worker 的 run() 方法就會被執行，
關于ThreadFactory說明：
public interface ThreadFactory {
Thread newThread(Runnable r);
}
3.5.2 Worker原始碼分析
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
private static final long serialVersionUID = 6138294804551838833L;
//執行緒型別的屬性：thread，執行緒池啟動作業執行緒，就是啟動這個thread，
// 1、通過this.thread=getThreadFactory().newThread(this)，初始化了屬性thread，this就是指Worker物件
//2、因為Worker類實作了Runnable介面，所以thread啟動后，會運行Worker的run()方法，然后就去執行runWorker(this)方法
final Thread thread;
//執行緒要執行的第1個任務（可能為 null) 它表示這個任務立即執行，不需要放到任務佇列，在作業執行緒數<核心執行緒數時，這種場景會出現
Runnable firstTask;
//保存Worker執行緒池執行過的任務數，在runWorker()的finally中累加更新，任務執行成功與否都會更新
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // AQS父類的state，設為-1
this.firstTask = firstTask; //firstTask賦初值
this.thread = getThreadFactory().newThread(this); //屬性thread賦值
}
//Runnable run方法實作
public void run() {
runWorker(this); //呼叫runWorkder方法：將Worker物件傳遞給呼叫者，這樣就可以訪問firstTask、thread等屬性以及lock()相關方法
}
// state 的值說明
// -1：worker初始化； 1 ：鎖被獨占； 0：鎖空閑
//是否持有鎖 AQS父類方法的實作
protected boolean isHeldExclusively() {
return getState() != 0;
}
//以獨占方式獲取鎖，將state設為1 AQS父類方法的實作
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false; //假如state=1，那么cas失敗，回傳false，執行緒就會進入AQS佇列等待
}
//釋放鎖，state設為0 AQS父類方法的實作
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(); }
//向執行緒發起中斷請求
// 符合：1、運行中的；2、沒有處于中斷 才能中斷
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
Worker類總結：
- 所謂的執行緒池，其實就是正在運行的多個Worker執行緒，
- Worker作為執行緒啟動后，它實際執行的是通過execute()提交的Runnable任務（實際業務），worker執行緒通過一個while回圈來不斷獲取并任務，從而達到執行緒復用的效果
- firstTask：執行緒要執行的第1個任務（可能為 null) 它表示這個任務立即執行，不需要放到任務佇列，在 1、執行緒數<核心執行緒數 2、佇列已滿且執行緒池不在運行狀態 這兩個場景下，
4 重要方法詳解
4.1 execute()方法
execute()用來提交要運行的任務
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get(); // 計算當前執行緒池的狀態及執行緒數
// 1、執行緒池執行緒數小于配置的核心執行緒數
if (workerCountOf(c) < corePoolSize) {
// 將任務提交給核心執行緒處理
if (addWorker(command, true))
return;
//失敗的情況：1、執行緒池已經被關閉、2、執行緒池執行緒數大于等于核心執行緒數 （不能以true的方式提交了 ）
c = ctl.get(); // 重新獲取執行緒池狀態
}
// 2、無空閑核心執行緒，將任務加入佇列
// 再次確認執行緒池為RUNNING狀態，將任務加入佇列【非阻塞式，佇列滿了會立即回傳false】
if (isRunning(c) && workQueue.offer(command)) {
//任務加入佇列成功
int recheck = ctl.get() ;//再次獲取當前執行緒池狀態(執行緒池可能被其它執行緒關閉了)
//判斷當前執行緒池狀態是不是RUNNING狀態,不是就從workQueue中洗掉command任務
if (! isRunning(recheck) && remove(command))
reject(command);//執行拒絕策略
//如果當前執行緒數是0（那證明還沒有其他作業執行緒去處理這個任務）,那么剛剛的任務肯定在阻塞佇列里面了,這
else if (workerCountOf(recheck) == 0)
addWorker(null, false);//開啟一個沒有任務的Worker執行緒去執行佇列的任務
}
// 3 workQueue添加worker失敗，即佇列滿了
//創建非核心執行緒并執行任務
else if (!addWorker(command, false)) //如果執行緒創建失敗，說明要么是執行緒池當前狀態!=RUNNING，或者是任務佇列已滿且執行緒總數達到最大執行緒數了
reject(command);//執行拒絕策略.
}
execute()總結
- 進行三次addWorker的嘗試：
- addWorker(command, true)：創建任務并以核心執行緒執行
- 核心執行緒數達到上限， 創建任務添加到任務佇列，不創建執行緒
- addWorker(null, false) ：任務添加到佇列后，接著執行緒池被關閉，并且從佇列移除該任務失敗，并且執行緒池執行緒數為0，這時創建任務并以非核心執行緒執行
- addWorker(command, false) ：任務佇列已滿，創建非核心執行緒并執行
- 任務提交失敗情況：執行緒池非RUNNING狀態 并且 任務佇列已滿并且執行緒池執行緒數達到最大執行緒數（maximumPoolSize）
4.2 addWorker()方法
//TERMINATED >TIDYING > STOP > SHUTDOWN > RUNNING
//創建新的執行緒執行當前任務
//firstTask: 指定新增執行緒執行的第一個任務或者不執行任務
private boolean addWorker(Runnable firstTask, boolean core) {
//外回圈：
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 如果執行緒池狀態是SHUTDOWN、STOP、TIDYING、TERMINATED就不允許提交，
// && 后面的特殊情況，執行緒池的狀態是SHUTDOWN并且要要執行的任務為Null并且佇列不是空，這種情況下是允許增加一個執行緒來幫助佇列中的任務跑完的，因為shutdown狀態下，允許執行完成阻塞隊里中的任務
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null && //execute()有addWorkder(null,false)的場景
! workQueue.isEmpty()))
return false;
//內回圈：cas修改作業執行緒數，同時判斷能否添加work
for (;;) {
int wc = workerCountOf(c);
//添加任務前，執行緒池執行緒數已達到上限，此時不允許添加，上限分這三種情況：
// 1、最大支持執行緒數
// 2、以core=true提交時，配置的核心執行緒數，（回傳false后，會以core=false再提交一次）
// 3、以core=false提交時，配置的執行緒池可容納最大執行緒數，
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize)) //使用core則上限為核心執行緒數，否則最大執行緒數
return false;
//沒超過上限，通過CAS的方式增加worker的數量（+1），增加成功就跳出外層回圈
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); //獲取最新的執行緒池狀態，與剛開始的狀態比較
// - 變了，就從外層回圈重新執行，重新進行狀態的檢查，
// - 沒變，從當前回圈重新執行，重新執行CAS操作，
if (runStateOf(c) != rs)
continue retry;
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//創建Worker，并給firstTask賦初值
w = new Worker(firstTask);
final Thread t = w.thread; //拿到屬性thread
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock(); //此處加鎖：因為涉及屬性：workers、largestPoolSize（可能） 更新
try {
int rs = runStateOf(ctl.get()); //獲取執行緒池最新狀態
if (rs < SHUTDOWN || //如果當前狀態是<SHUTDOWN也就是RUNNING狀態
(rs == SHUTDOWN && firstTask == null)) { //或者狀態是SHUTDOWN并且當前任務是空的（比如前面說的場景：阻塞隊里里面還有，但當前已經是不允許提交的狀態了）
if (t.isAlive()) // 檢查Worker執行緒已經開始跑了，（thread.start()變為alive）
throw new IllegalThreadStateException();
workers.add(w); //增加worker
int s = workers.size(); //獲取最新worker的總數，比較并更新largestPoolSize
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true; //表示添加worker成功
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
//啟動worker執行緒，該執行緒會一直回圈執行getTask()，直至回傳null，執行緒才結束
t.start(); //執行runWorker()
workerStarted = true; //表示執行緒已經跑起來了
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);//worker執行緒沒成功啟動，進入失敗處理邏輯
}
return workerStarted;//;回傳當前worker是否啟動成功，
}
addWorker()總結：
- 檢查執行緒池狀態以確定能否提交任務
- 校驗能否以核心執行緒的方式提交任務
- 執行緒池的狀態是SHUTDOWN并且任務佇列不是空，允許增加一個執行緒來幫助佇列中的任務跑完，但不會提交任務
- 更新執行緒池執行緒數
- 超過執行緒池執行緒數峰值則更新峰值（largestPoolSize）
- 加鎖（mainLock）來更新
- 啟動worker執行緒
4.3 runWorker()方法
//執行任務
final void runWorker(Worker w) {
Thread wt = Thread.currentThread(); //runWorker()是由Worker.run()呼叫，因此wt就是worker執行緒
Runnable task = w.firstTask; //拿到firstTask并賦值給區域變數task
w.firstTask = null; //firstTask置空
w.unlock(); // 將state設定為0，因為建構式設成-1，在執行任務前置為0，
boolean completedAbruptly = true;//標識任務是不是立刻就完成了，
try {
//回圈：先執行firstTask（不為空），后續通過getTask()獲取任務，
while (task != null || (task = getTask()) != null) {
//任務執行前加鎖，任務完成后解鎖，
//任何地方可通過判斷鎖狀態來確認worker是否執行中
w.lock(); //加鎖，防止任務在執行程序中被中斷，
//判斷目的：確保執行緒池當狀態值大于等于 STOP 時有向執行緒發起過中斷請求【呼叫了shutdownNow()】
// 兩種情況:
//1)如果當前執行緒池的狀態是>=Stop的，并且當前執行緒沒有被中斷，那么就要執行中斷，
//2)或者當前執行緒目前是已中斷的狀態并且執行緒池的狀態也是>=Stop的（注意Thread.interrupted是會擦除中斷識別符號的），那么因為中斷識別符號已經被擦除了，那么!wt.isInterrupted()一定回傳true，這個時候還是要將當前執行緒中斷，第二次執行runStateAtLeast(ctl.get(), STOP)相當于一個二次檢查
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();//中斷worker執行緒 ，因為執行緒池將要終止了，所以這里沒有從workerSet移除當前執行緒
try {
beforeExecute(wt, task);//前置操作，空方法，可以業務自己實作
Throwable thrown = null;
try {
//執行任務：就是執行通過execute()提交的Runnable
task.run();//第一個是firstTask，后面的是通過getTask()拿到的任務
} 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;//最后將task置為null，觸發while回圈的條件getTask()
w.completedTasks++; //已完成的任務計數器+1
w.unlock();//釋放當前執行緒的獨占鎖
}
}
completedAbruptly = false; //當第一個try的代碼塊有例外， completedAbruptly = false 不生效，最后completedAbruptly為true表示發生未知例外了
} finally {
//getTask回傳null時，執行任務退出
processWorkerExit(w, completedAbruptly);//completedAbruptly=true表示是突然退出的
}
}
runWorker()總結：
- 執行任務前先判斷執行緒池是否是STOPING狀態，是則中斷worker執行緒，
- 執行任務：先執行firstTask，再從任務佇列獲取執行
- 如果沒有任務，呼叫processWorkerExit()來執行執行緒退出的作業，
- 只要還有任務，worker執行緒就一直執行任務，并重繪completedTasks
4.4 getTask()方法
private Runnable getTask() {
boolean timedOut = false;
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
//1、先判斷能否獲取到任務
// 1)如果執行緒池的狀態是>=STOP狀態,這個時候不再處理佇列中的任務,并且減少worker記錄數量,回傳的任務為null,這個時候在runRWorker方法中會執行processWorkerExit進行worker的退出操作.
// 2)如果執行緒池的狀態是>=SHUTDOWN并且workQueue為空,就說明處于SHOTdown以上的狀態下,且沒有任務在等待,那么也屬于獲取不到任務,getTask回傳null.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();//扣減執行緒池執行緒數，在processWorkerExit()處理執行緒退出
return null;
}
int wc = workerCountOf(c);//獲取當前wokrer的數量
//以下涉及空閑執行緒是否會被執行緒池銷毀的處理邏輯
// 執行緒超時處理前置條件：開啟核心執行緒超時 或 執行緒池執行緒數大于核心執行緒數
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
//執行緒超時處理的進一步判斷：
// 執行緒池執行緒數超過maximumPoolSize 或者 執行緒設定允許超時且當前worker取任務超時
//并且
// 執行緒池大小不是零或阻塞佇列是空的),這種就回傳null,并減少執行緒池執行緒計數
// 1、 (wc>maximumPoolSize) && (wc>1) 一般情況，執行緒池執行緒數會少于配置的最大執行緒數，但在addWork中 狀態=shutdown且佇列不為空時，會創建一個Worker，此時可能導致wc>maximumPoolSize，這里同時限定wc>1，因此執行緒池減少1個執行緒也不影響任務的執行【processWorkerExit()會保證還有任務就至少留有1個worker執行緒】，
// 2、 (wc>maximumPoolSize) && (workQueue.isEmpty()) 沒有任務了，扣減更不影響
// 3 、(timed && timedOut) && (wc > 1) 超時了，先扣減再說
// 4 、(timed && timedOut) && (workQueue.isEmpty()) 超時了&佇列沒有任務，必須要扣減
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
//這里為啥不用decrementWorkerCount()呢,上面使用decrementWorkerCount()是因為確定不管是什么情況下,數量都要減,多減一次也沒事,因為這個時候就是要關閉執行緒池釋放資源
//這里不一樣,執行緒池的狀態可能是RUNNING狀態,多減一次,可能導致獲取不到worker去跑
if (compareAndDecrementWorkerCount(c))
return null; //扣減執行緒池執行緒數，在processWorkerExit()處理執行緒退出
continue;//扣減失敗， 跳出本次回圈重新檢查
}
//從佇列中獲取任務
//符合【執行緒超時處理前置條件】時用poll設定超時時間，不符合就使用take（阻塞直至有回傳）
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r; //task不為空，此處回傳task
timedOut = true; // 此處，r == null，肯定是poll操作超時了（注意，不代表佇列空了），繼續for回圈，回到if ((wc > maximumPoolSize || (timed && timedOut)) 這個地方退出回圈
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
private void decrementWorkerCount() {
do {} while (! compareAndDecrementWorkerCount(ctl.get()));
}
getTask()總結：
- 從workQueue中獲取一個任務并回傳
- 沒有獲取到任務就扣減執行緒池執行緒數，獲取不到任務的四種情況：
- 執行緒池的狀態是>=STOP
- 執行緒池的狀態是SHUTDOWN并且任務佇列為空
- 獲取任務超時
- 執行緒池執行緒數大于maximumPoolSize并且佇列為空
4.5 processWorkerExit()方法
//worker執行緒沒有拿到任務，成為空閑執行緒，該方法對空閑執行緒進一步處理
private void processWorkerExit(Worker w, boolean completedAbruptly) {
//如果completedAbruptly為true，則說明執行緒執行時出現例外，需要將workerCount數量減一
//如果completedAbruptly為false，說明在getTask方法中已經對workerCount進行減一，這里不用再減
if (completedAbruptly)
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//更新已完成任務的數量的統計項
completedTaskCount += w.completedTasks;
//從worker集合中移除該worker
workers.remove(w);
} finally {
mainLock.unlock();
}
//嘗試關閉執行緒池,但如果是正常運行狀態,就不會關閉
tryTerminate();
int c = ctl.get();
if (runStateLessThan(c, STOP)) {//1、執行緒池是SHUTDOWN或RUNNING（如果不是這兩個狀態,說明執行緒已經停止了，不做任何操作）
if (!completedAbruptly) {//2、執行緒正常結束
// 如果沒有開啟核心執行緒超時配置，則至少保留corePoolSize個執行緒；
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && !workQueue.isEmpty())//如果允許核心執行緒超時并且當前佇列里面還有任務沒跑，必須留1個執行緒,不能全死掉.
min = 1;
// 如果執行緒池數量>=最少預留執行緒數
if (workerCountOf(c) >= min)
return; // 執行緒自然結束了，不用補充worker
}
// 1、執行任務例外結束的，補充worker
// 2、如果執行緒池數量<最少預留執行緒數，補充worker
addWorker(null, false);//例外結束 增加worker
//注: 別問我為啥上面要洗掉worker,還要再加,不刪是不是不用加了. 明確下那個任務已經退出getTask那塊的死回圈了,永遠回不去了,只能新增worker.
}
}
processWorkerExit()方法總結！！！！！：
- 當Worker執行緒結束前，完成以下作業：扣減執行緒池執行緒數（ctl）、更新已完成任務數（completedTaskCount）、Worker集合中移除一個Worker（workers）、嘗試終止執行緒池、計算執行緒池的最少保留執行緒數、根據最少保留執行緒數來確定是否補充一個Worker，
- 關于最少保留執行緒數：如果沒有開啟核心執行緒超時配置，則至少保留corePoolSize個執行緒；如果開啟核心執行緒超時并且當前佇列里面還有任務，只需保留1個執行緒；
- 需要補充worker的兩種情況：1、執行緒池執行緒數<最少保留執行緒數 2、任務執行例外結束
4.6 tryTerminate()方法
//嘗試終止執行緒池
final void tryTerminate() {
for (;;) { //cas自旋 確保更新成功
int c = ctl.get();
//RUNNING狀態,不能終止執行緒池
//執行緒池狀態是TIDYING或TERMINATED說明執行緒池已經處于正在終止的路上,不用再終止了.
//狀態為SHUTDOWN，但是任務佇列不為空,也不能終止執行緒池
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
//呼叫shutdown()或者shutdownNow()方法時，執行以下處理
//作業執行緒數量不等于0，中斷一個空閑的作業執行緒并回傳
//這個時候執行緒池一定是 1、STOP的狀態或者 2、SHUTDOW且佇列為空 這兩種情況中斷一個空閑worker
if (workerCountOf(c) != 0) {
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 設定執行緒池狀態為TIDYING，如果設定成功，則呼叫terminated()
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
terminated(); //鉤子方法，子類實作，默認什么都不做
} finally {
// 設定狀態為TERMINATED
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll(); //喚醒阻塞等待的執行緒 （future的場景）
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
tryTerminate()總結
- 嘗試終止執行緒池
- 不能終止執行緒池：
- 狀態是RUNNING，不能直接終止（如果是呼叫shutdown()，shutdownNow()，會先將狀態改為SHUTDOWN）
- 狀態是TIDYING或者TERMINATED，不能終止（因為已經處于終止程序中）
- 狀態是SHUTDOWN并且任務佇列不為空，不能終止（因為還有任務要處理）
- 可以終止執行緒池：
- 狀態是SHUTDOWN并且任務佇列為空
- 狀態是STOP
- 符合可以終止執行緒池的條件下，如果執行緒池執行緒數不等于0，那就中斷1個Worker執行緒，不修改執行緒池狀態
- 符合可以終止執行緒池的條件下，并且執行緒池執行緒數等于0，那就將執行緒池狀態改為TIDYING，執行完鉤子方法terminated()后狀態再改為TERMINATED
interruptIdleWorkers(ONLY_ONE); 是否好奇為啥這里只中斷一個worker呢, 這里就涉及到了執行緒池的優雅退出了.
當執行到 interruptIdleWorkers(ONLY_ONE) 前面的時候, 執行緒池只能處于兩種狀態:
1) STOP 狀態 , 這個時候 workQueue 可能是有值的 , workQueue 在清空的程序中了.
2) SHUTDOWN 狀態并且 workQueue 是空的 .
這兩種狀態都是說明, 執行緒池即將關閉, 或者說空閑的執行緒此時已經沒用了,這個時候隨手關一個, 反正要關,早關晚關而已.
4.7 interruptIdleWorker()方法
//中斷一個或多個執行緒
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//遍歷worker，根據onlyOne判斷，如果為ture只中斷一個執行緒
for (Worker w : workers) {
Thread t = w.thread;
//執行緒沒有被中斷并且執行緒是空閑狀態
//通過tryLock實作：不能中斷還沒有開始執行或者還在執行中的worker執行緒，
//執行緒未啟動：-1 ，執行緒正在執行：1 ，trylock：0->1 ;
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt(); //中斷操作，之后該執行緒就結束了
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
interruptIdleWorker()總結：
- 從worker集合中遍歷并中斷worker執行緒
- 只有worker執行緒狀態是0的，才能夠中斷（不能中斷未啟動或者還在執行中的Worker執行緒）
4.8 shutdown()方法
//初始化一個有序的關閉，之前提交的任務都會被執行，但是新提交的任務則不會被允許放入任務佇列中，如果之前被呼叫過了的話，那么再次呼叫也沒什么用
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock(); //mainLock是全域變數，加鎖確保不會并發關閉執行緒池
try {
checkShutdownAccess();//安全策略判斷，方法檢查每一個執行緒池的執行緒是否有可以ShutDown的權限，
advanceRunState(SHUTDOWN); //CAS自旋把ctl中的狀態從RUNNING變為SHUTDOWN
interruptIdleWorkers();//中斷所有空閑執行緒
onShutdown(); // 方法告知子類，執行緒池要處于ShutDown狀態了 ，ScheduledThreadPoolExecutor預留的鉤子
} finally {
mainLock.unlock();
}
tryTerminate();//嘗試終止執行緒池
}
shutdown()方法總結
- 執行shutdown()方法：關閉執行緒池，不再接受新的任務，已提交執行的任務繼續執行，
- 呼叫interruptIdleWorkers()先中斷所有空閑執行緒
- 呼叫tryTerminate()嘗試終止執行緒池
- shutdown()將執行緒池狀態改為SHUTDOWN但不是STOP
4.9 shutdownNow()方法
//關閉執行緒池，不再接受新的任務，正在執行的任務嘗試終止
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP);//執行緒池的狀態置為STOP
interruptWorkers();
tasks = drainQueue(); //將剩余任務回傳
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
private void interruptWorkers() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) //回圈所有的worker
w.interruptIfStarted();//已經啟動的執行緒直接執行中斷
} finally {
mainLock.unlock();
}
}
void interruptIfStarted() {
Thread t;
//只有剛剛構建的worker的時候,狀態state值是-1(這里也能體現剛構建的worker無法被中斷),其他情況都是>=0的
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
ShutDownNow()方法總結
- 關閉執行緒池，不再接受新的任務，中斷已經啟動的Worker執行緒
- 將執行緒池狀態改為STOP
- 回傳未完成的任務佇列
4.10 isShutdown()方法
確認執行緒池是否關閉，判斷狀態是不是RUNNING.
public boolean isShutdown() {
return ! isRunning(ctl.get());
}
4.11 prestartCoreThread()方法
public boolean prestartCoreThread() {
return workerCountOf(ctl.get()) < corePoolSize &&
addWorker(null, true);
}
- 啟動一個空閑的執行緒作為核心執行緒
- 如果核心執行緒數已到閾值, 會加入失敗, 回傳false, 如果執行緒池處于SHUTDOWN以上的狀態也回傳false
- 只有真正這個執行緒呼叫start方法跑起來, 才會回傳true
4.12 prestartAllCoreThreads()方法
啟動所有核心執行緒，使他們等待獲取任務
public int prestartAllCoreThreads() {
int n = 0;
while (addWorker(null, true))//null代表空閑執行緒,true代表是增加的是核心執行緒
++n;//死回圈增加空閑 worker 而已
return n;
}
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/503520.html
標籤：Java