Java線程池ThreadPoolExecutor源碼深入分析
更新時間：2022年08月26日 10:46:54 作者：niuyongzhi
ThreadPoolExecutor作為java.util.concurrent包對外提供基礎實現，以內部線程池的形式對外提供管理任務執(zhí)行，線程調度，線程池管理等等服務
1.線程池Executors的簡單使用
1)創(chuàng)建一個線程的線程池。
Executors.newSingleThreadExecutor();
//創(chuàng)建的源碼
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
2)創(chuàng)建固定大小的線程池，參數為int，是線程池核心線程和最大線程的數量
Executors.newFixedThreadPool(2);
//創(chuàng)建的源碼
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
3)創(chuàng)建一個線程數不設限的線程池，
//創(chuàng)建的源碼，核心線程是0，最大線程是Integer.MAX_VALUE
Executors.newCachedThreadPool()；
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}使用方法，使用同步代碼塊，保證線程池實例是唯一的。
使用方法：
private static ExecutorService sSingleThreadExecutor = null; // lazy, guarded by class
public static ExecutorService singleThreadExecutor() {
//當前的類對象為鎖
synchronized (ThreadPool.class) {
if (sSingleThreadExecutor == null) {
sSingleThreadExecutor = Executors.newSingleThreadExecutor();
}
return sSingleThreadExecutor;
}
}通過以上三種方式，可以創(chuàng)建一個簡單的線程池。
但是有弊端：
newSingleThreadExecutor和newFixedThreadPool，運行的請求隊列是長度為Integer.MAX_VALUE,可能會堆積大量的請求，從而造成oom。
而newCachedThreadPool允許的線程數量為最大值Integer.MAX_VALUE，也會造成oom。
2.通過ThreadPoolExecutor創(chuàng)建線程池
下面是OkHttp中Dispatcher.java線程池：
ExecutorService executorService;
public synchronized ExecutorService executorService() {
if (executorService == null) {
executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));
}
return executorService;
}OkHttp中ConnectionPool.java
private static final Executor executor = new ThreadPoolExecutor(0 ,
Integer.MAX_VALUE , 60L , TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp ConnectionPool", true));使用方式：
//call 實現 Runnable 接口。調用execute方法即可將入線程池，執(zhí)行run方法中的代碼。 executorService().execute(call);
3.ThreadPoolExecutor各個參數的含義
corePoolSize：核心線程數，即使是空閑線程也不會銷毀。這樣做的目的是為了降低執(zhí)行任務時創(chuàng)建線程的時間和性能開銷。
maximumPoolSize：最大線程數。當核心線程被用完時，會創(chuàng)建新的線程來執(zhí)行任務，但是創(chuàng)建的數量不能超過這個最大值。
keepAliveTime：線程的存活時間。除核心線程外，其他線程一旦執(zhí)行完任務，就會處于空閑狀態(tài)，超過這個時間就會被銷毀。
unit：keepAliveTime設置的時間單位。
workQueue：任務的阻塞隊列。線程數量有限，當任務過多來不及執(zhí)行時，就會加入到這個阻塞隊列中，等到有空閑進程，
就會從這個隊列取出任務去執(zhí)行。隊列都是先進先出的FIFO。
threadFactory：新線程產生的方式。
handler：拒絕策略，超過任務隊列設置的最大值時。再有新的任務進來，就會執(zhí)行這個拒絕策略。
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
}
線程池的阻塞隊列：
ArrayBlockingQueue：
是基于數組的任務隊列。里面用一個數組來存放任務。當我們new的時候，需要指定數組大小。
還有兩個int變量putIndex和takeIndex用來表示隊列的頭部和尾部在數組中的位置。
LinkedBlockingQueue:
是基于鏈表的，內部用一個單向鏈表來存放任務。創(chuàng)建時可以指定大小，如果不指定則是Integer.MAX_VALUE
PriorityBlockingQueue:
基于優(yōu)先級的阻塞隊列。
SynchronousQueue：
一種無緩沖的等待隊列。有新任務進來直接交給線程執(zhí)行。
OkHttp中使用的就是這種隊列，他的最大線程數為Integer.MAX_VALUE。保證有任務進來就能馬上執(zhí)行。
RejectedExecutionHandler拒絕策略，這是一個接口。不同的實現執(zhí)行不同的策略。
public interface RejectedExecutionHandler {
void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}
AbortPolicy:拒絕行為直接拋出異常 RejectedExecutionException
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
DiscardPolicy:保持靜默，什么也不做。
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
DiscardOldestPolicy：丟棄任務隊里中最老的任務，嘗試將新任務加入隊列
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
CallerRunsPolicy：直接由提交任務這執(zhí)行這個任務。
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
如果在創(chuàng)建線程池的時候，不知道具體的拒絕策略。那么ThreadPoolExecutor默認的策略是AbortPolicy。
private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();線程池可以執(zhí)行兩種類型的任務：Runable和Callable
class MyRunable implements Runnable{
@Override
public void run() {
}
}
class MyCallable implements Callable{
@Override
public Object call() throws Exception {
return null;
}
}
Runnable 沒有返回值，返回的是void，不允許拋出異常。
Callable 有返回值，返回的是Object，允許拋出異常。4.線程池的源碼分析
線程池的狀態(tài)：
//運行狀態(tài)，可以接受新任務，并且處理排隊任務。 private static final int RUNNING = -1 << COUNT_BITS; //關閉狀態(tài)，不再接受新任務，不過仍然會處理排隊任務。 private static final int SHUTDOWN = 0 << COUNT_BITS; //停止狀態(tài)，不再接受新任務，也不處理排隊任務，同時中斷處理中的任務 private static final int STOP = 1 << COUNT_BITS; //整理狀態(tài)，當前所有任務終止，workerCount計數為0，線程切換為TIDYING狀態(tài)，并且執(zhí)行terminal()方法 private static final int TIDYING = 2 << COUNT_BITS; //終止狀態(tài)，說明terminal()方法執(zhí)行完成。 private static final int TERMINATED = 3 << COUNT_BITS;
ctlof是得到新的ctl值。通過ctl可以計算線程池的狀態(tài)和數量
runStateOf 計算當前線程池的狀態(tài)。
workerCountOf計算線程池的數量。
// ctlOf計算ctl的新值，也就是線程池狀態(tài)和線程池中線程數量。
private static int ctlOf(int rs, int wc) { return rs | wc; }
//獲取ctl的高三位，也就是線程池的狀態(tài)。
private static int runStateOf(int c) { return c & ~CAPACITY; }
//獲取ctl的低29位，也就是線程池中的線程數。
private static int workerCountOf(int c) { return c & CAPACITY; }
其中runStateOf(int c)和workerCountOf(int c)的參數c就是通過ctlOf(int rs, int wc)獲得的ctl值。向線程池中添加一個任務：executorService().execute(call);
然后看看源碼中是如何執(zhí)行的，是如何添加任務的。
ctl 用來表示線程池的狀態(tài)和線程數量，
在ThreadPoolExcutor中使用32位二進制數來表示線程池的狀態(tài)和線程中線程數量。
其中前3位表示線程池的狀態(tài)，后29位表示線程池中的線程數。
public void execute(Runnable command) {
int c = ctl.get();
//如果工作線程數量小于核心線程數，
//提交的任務會通過addWorker(command, true)創(chuàng)建一個新的核心線程來執(zhí)行， 這個參數傳的是true，表示去新增核心線程。
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true)){
//添加成功則return
return;
}
//添加核心線程失敗則重新獲取線程池的狀態(tài)和數量
c = ctl.get();
}
//進入到下面說明當前工作線程大于或等于核心線程。
//如果線程池處于運行狀態(tài)，則加入隊列
if (isRunning(c) && workQueue.offer(command)) {
//如果入隊成功，則重新獲取線程池的狀態(tài)
int recheck = ctl.get();
//如果線程池不處于運行狀態(tài)，則從隊列中remove
if (!isRunning(recheck) && remove(command)){
//成功刪除，則執(zhí)行拒絕策略
reject(command);
}else if (workerCountOf(recheck) == 0){
//進入這個分支有兩種情況1.線程池處于運行狀態(tài) 2.線程從不處于運行狀態(tài)，但是remove失敗
則會判斷workerCountOf如果工作線程為0，則會創(chuàng)建非核心線程去執(zhí)行任務。
addWorker為null，和false。false表示非核心線程。null說明創(chuàng)建的線程去執(zhí)行隊列里的任務。
addWorker(null, false);
}
//進入到這個分支有兩種情況1.線程池處于非運行狀態(tài)2.運行狀態(tài)但是入隊失敗了。
這時候創(chuàng)建非核心線程去執(zhí)行任務
}else if (!addWorker(command, false)){
如果創(chuàng)建非核心線程失敗了，則執(zhí)行拒絕策略。
reject(command);
}
}通過以上源碼分析，線程池的運行原理可以總結為一下幾點：
1.通過execute方法提交任務時，運行線程小于corePoolSize時，則會創(chuàng)建新的核心線程來執(zhí)行這個任務。
2.通過excute方法提交任務時，運行線程大于等于corePoolSize時，則會加入到隊列中，等待線程調度執(zhí)行。
3.通過excuete方法提交任務時，運行線程大于等于corePoolSize時，并且加入隊列失敗(隊列滿了)，新提交的任務將會通過創(chuàng)建新的線程執(zhí)行。
4.通過excute方法提交任務時，運行線程大于maximumPoolSize時，隊列也滿了，則會執(zhí)行拒絕策略。
5.當線程池中的線程執(zhí)行完任務處于空閑狀態(tài)時，則會嘗試從任務隊列中取頭結點任務執(zhí)行。
接下來看addWorker如何添加任務。
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 如果線程池處于非運行狀態(tài)，則不會創(chuàng)建線程。
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty())){
return false;
}
//如果線程池處于運行狀態(tài)，則直接走下面的創(chuàng)建添加邏輯。
for (;;) {
//獲取工作線程數量
int wc = workerCountOf(c);
//wc >= CAPACITY 工作線程大于最大容量
// wc >= (core ? corePoolSize : maximumPoolSize) 如果工作線程大于了核心線程或最大線程，
//只要這兩個條件有一個成立則return。
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize)){
return false;
}
//創(chuàng)建線程數量+1，這里用到了CAS。關于CAS后面再寫文章分析。
if (compareAndIncrementWorkerCount(c)){
break retry;
}
//如果CAS操作失敗，線程數量沒有加1，則重新獲取線程的狀態(tài)。
c = ctl.get(); // Re-read ctl
//判斷當前狀態(tài)和之前狀態(tài)，如果不同，說明線程池狀態(tài)發(fā)生了變化。重新跳到retry的外層循環(huán)。
//如果相同，則說明線程池沒有變化，繼續(xù)進行內層循環(huán)。
if (runStateOf(c) != rs){
continue retry;
}
// else CAS failed due to workerCount change; retry inner loop
}
}
//執(zhí)行到這說明線程數量已經完成+1，接下來進行線程的創(chuàng)建。
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//這個創(chuàng)建一個worker對象。在worker構造方法中，會利用ThreadPoolExecutor中傳遞過了的ThreadFactory創(chuàng)建一個Thread
//默認是通過Executors.defaultThreadFactory()，創(chuàng)建一個線程。
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
//拿到一個重入鎖對象。
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//拿到線程池的狀態(tài)
int rs = runStateOf(ctl.get());
//如果線程池處于運行狀態(tài)或者處于關閉狀態(tài)并且firstTask == null
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) {
throw new IllegalThreadStateException();
}
//添加到work集合
workers.add(w);
int s = workers.size();
if (s > largestPoolSize){ //更新一下最大線程數
largestPoolSize = s;
}
//標志位，添加成功
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
//添加成功則啟動線程
t.start();
//啟動成功
workerStarted = true;
}
}
} finally {
//如果沒有啟動成功則從線程池中移除。
if (! workerStarted){
addWorkerFailed(w);
}
}
return workerStarted;
}關鍵代碼看看 w = new Worker(firstTask);做了啥
Worker(Runnable firstTask) {
setState(-1);
//將傳進來的任務賦值給成員變量
this.firstTask = firstTask;
//創(chuàng)建一個線程，并把Worker本身當做Runnable傳進了Thread中去。
this.thread = getThreadFactory().newThread(this);
}
public interface ThreadFactory {
Thread newThread(Runnable r);
}注意newThread(this)。Worker把自己當做Runnable傳到了線程中去。當調用t.start()方法時會調用Worker的run方法。
public void run() {
runWorker(this);
}
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//如果task不為null，則先執(zhí)行當前任務
//如果task傳進來是null則從隊列中取任務，執(zhí)行隊列里的任務。
//getTask()就是從任務隊列中提取在等待的隊伍。
while (task != null || (task = getTask()) != null) {
w.lock();
//(runStateAtLeast(ctl.get(), STOP) 線程池處于STOP，TIDYING，TERMINATED狀態(tài)
處于這些狀態(tài)的線程池是無法執(zhí)行任務的。
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted()){
//中斷線程
wt.interrupt();
}
//執(zhí)行到下面說明線程池處于RUNNING或SHUTDOWN狀態(tài)
//由此也可以看出SHUTDOWN狀態(tài)的線程池，是可以執(zhí)行隊列里的任務的，但是隊列不在接收新的任務添加
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
//執(zhí)行任務的
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);
}
}getTask()從任務隊列中，提取任務。
private Runnable getTask() {
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
try {
//從任務隊列中取出任務
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}通過以上源碼分析，可以總結一下幾點。
addWorker(Runnable firstTask, boolean core)
1.如果firstTask為null，則會創(chuàng)建線程去執(zhí)行隊列里的任務。
2.如果不為null，則會去執(zhí)行當前任務，然后再執(zhí)行隊列里的任務。
3.core 如果為true，則會創(chuàng)建核心線程，如果為false，則會創(chuàng)建非核心線程。
4.addWorker 會創(chuàng)建線程，啟動線程，執(zhí)行任務。
在創(chuàng)建線程之前會判斷線程池的狀態(tài)、以及核心線程或最大線程數。
如果創(chuàng)建成功啟動線程的start方法，然后調用worker的runWorker()方法。
到此這篇關于Java線程池ThreadPoolExecutor源碼深入分析的文章就介紹到這了,更多相關Java ThreadPoolExecutor內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:
- Java線程池?ThreadPoolExecutor?詳解
- Java多線程ThreadPoolExecutor詳解
- java高并發(fā)ThreadPoolExecutor類解析線程池執(zhí)行流程
- java高并發(fā)ScheduledThreadPoolExecutor與Timer區(qū)別
- 徹底搞懂java并發(fā)ThreadPoolExecutor使用
- Java多線程編程基石ThreadPoolExecutor示例詳解
- 源碼分析Java中ThreadPoolExecutor的底層原理
- 一文搞懂Java的ThreadPoolExecutor原理
- 一文弄懂Java中ThreadPoolExecutor
相關文章
spring boot加入攔截器Interceptor過程解析
這篇文章主要介紹了spring boot加入攔截器Interceptor過程解析,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2019-10-10
Trie又被稱為前綴樹、字典樹。Trie利用字符串的公共前綴來高效地存儲和檢索字符串數據集中的關鍵詞，最大限度地減少無謂的字符串比較，其核心思想是用空間換時間。本文主要介紹了Trie的原理及實現，感興趣的可以了解一下2022-11-11
這篇文章主要介紹了如何用Java來進行文件切割和簡單的內容過濾的實現，具有一定的參考價值，感興趣的小伙伴們可以參考一下。2017-01-01
Spring是一個非常受歡迎的Java框架，它用于構建web和企業(yè)應用。本文介紹將各種Spring的配置方式，幫助您了解配置Spring應用的復雜性2017-08-08
Java SpringBoot在RequestBody中高效的使用枚舉參數原理案例詳解
這篇文章主要介紹了Java SpringBoot在RequestBody中高效的使用枚舉參數原理案例詳解,本篇文章通過簡要的案例,講解了該項技術的了解與使用,以下就是詳細內容,需要的朋友可以參考下2021-09-09
這篇文章主要介紹了JAVA 注解詳解及簡單實例的相關資料,需要的朋友可以參考下2017-05-05
這篇文章主要介紹了java包本質及命名規(guī)范,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-03-03
這篇文章主要介紹了Linux部署springboot項目彩色日志打印方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-04-04
Java矩陣連乘問題(動態(tài)規(guī)劃)算法實例分析
這篇文章主要介紹了Java矩陣連乘問題(動態(tài)規(guī)劃)算法,結合實例形式分析了java實現矩陣連乘的算法原理與相關實現技巧,需要的朋友可以參考下2017-11-11
虛擬線程是JDK19中引入的,JDK21正式發(fā)布,本文主要介紹了JDK21中虛擬線程,具有一定的參考價值,感興趣的可以了解一下2023-10-10