java?ThreadPoolExecutor線程池內(nèi)部處理流程解析
ThreadPoolExecutor線程池內(nèi)部處理淺析
我們知道如果程序中并發(fā)的線程數(shù)量很多,并且每個線程都是執(zhí)行一個時間很短的任務就結(jié)束時,會因為頻繁創(chuàng)建線程而大大降低系統(tǒng)的效率,因此出現(xiàn)了線程池的使用方式,它可以提前創(chuàng)建好線程來執(zhí)行任務。本文主要通過java的ThreadPoolExecutor來查看線程池的內(nèi)部處理過程。
ThreadPoolExecutor
java.uitl.concurrent.ThreadPoolExecutor類是線程池中最核心的一個類,下面我們來看一下ThreadPoolExecutor類的部分實現(xiàn)源碼。
構(gòu)造方法
ThreadPoolExecutor類提供了如下4個構(gòu)造方法
// 設置線程池時指定核心線程數(shù)、最大線程數(shù)、線程存活時間及等待隊列。
// 線程創(chuàng)建工廠和拒絕策略使用默認的(AbortPolicy)
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
// 設置線程池時指定核心線程數(shù)、最大線程數(shù)、線程存活時間、等待隊列及線程創(chuàng)建工廠
// 拒絕策略使用默認的(AbortPolicy)
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
threadFactory, defaultHandler);
}
// 設置線程池時指定核心線程數(shù)、最大線程數(shù)、線程存活時間、等待隊列及拒絕策略
// 線程創(chuàng)建工廠使用默認的
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), handler);
}
// 設置線程池時指定核心線程數(shù)、最大線程數(shù)、線程存活時間、等待隊列、線程創(chuàng)建工廠及拒絕策略
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;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}通過觀察上述每個構(gòu)造器的源碼實現(xiàn),我們可以發(fā)現(xiàn)前面三個構(gòu)造器都是調(diào)用的第四個構(gòu)造器進行的初始化工作。
下面解釋一下構(gòu)造器中各個參數(shù)的含義:
- corePoolSize:核心池的線程個數(shù)上線,在創(chuàng)建了線程池后,默認情況下,線程池中并沒有任何線程,而是等待有任務到來才創(chuàng)建線程去執(zhí)行任務。默認情況下,在創(chuàng)建了線程池后,線程池中的線程數(shù)為0,當有任務來之后,就會創(chuàng)建一個線程去執(zhí)行任務,當線程池中的線程數(shù)目達到corePoolSize后,就會把到達的任務放到緩存隊列當中。
- maximumPoolSize:線程池最大線程數(shù),這個參數(shù)也是一個非常重要的參數(shù),它表示在線程池中最多能創(chuàng)建多少個線程。
- keepAliveTime:表示線程沒有任務執(zhí)行時最多保持多久時間會終止。默認情況下,只有當線程池中的線程數(shù)大于corePoolSize時,keepAliveTime才會起作用,直到線程池中的線程數(shù)不大于corePoolSize,即當線程池中的線程數(shù)大于corePoolSize時,如果一個線程空閑的時間達到keepAliveTime,則會終止,直到線程池中的線程數(shù)不超過corePoolSize。但是如果調(diào)用了allowCoreThreadTimeOut(boolean)方法,在線程池中的線程數(shù)不大于corePoolSize時,keepAliveTime參數(shù)也會起作用,直到線程池中的線程數(shù)為0。
- unit:參數(shù)keepAliveTime的時間單位。
- workQueue:一個阻塞隊列,用來存儲等待執(zhí)行的任務,這個參數(shù)的選擇也很重要,會對線程池的運行過程產(chǎn)生重大影響;
- threadFactory:線程工廠,主要用來創(chuàng)建線程;
- handler:表示當拒絕處理任務時的策略。有以下四種取值:ThreadPoolExecutor.AbortPolicy:丟棄任務并拋出RejectedExecutionException異常。 ThreadPoolExecutor.DiscardPolicy:也是丟棄任務,但是不拋出異常。 ThreadPoolExecutor.DiscardOldestPolicy:丟棄隊列最前面的任務,然后重新嘗試執(zhí)行任務(重復此過程)ThreadPoolExecutor.CallerRunsPolicy:由調(diào)用線程處理該任務。
核心方法
在ThreadPoolExecutor類中,最核心的任務提交方法是execute()方法,雖然通過submit也可以提交任務,但是實際上submit方法里面最終調(diào)用的還是execute()方法。
public void execute(Runnable command) {
// 判斷提交的任務command是否為null,若是null,則拋出空指針異常;
if (command == null)
throw new NullPointerException();
// 獲取線程池中當前線程數(shù)
int c = ctl.get();
// 如果線程池中當前線程數(shù)小于核心池大小,進入if語句塊
if (workerCountOf(c) < corePoolSize) {
// 如果以給定的命令啟動一個核心線程執(zhí)行任務成功,直接返回
if (addWorker(command, true))
return;
c = ctl.get();
}
// 如果當前線程池處于RUNNING狀態(tài),則將任務放入任務緩存隊列
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 如果線程池不處于運行狀態(tài)并且移除剛加入的任務成功則執(zhí)行拒絕策略
if (! isRunning(recheck) && remove(command))
reject(command);
// 如果當前線程數(shù)為0,則在線程池里增加一個線程,保證隊列里的任務不會沒有線程執(zhí)行
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 嘗試啟動核心線程之外的線程,如果不滿足,則執(zhí)行對應的拒絕策略
else if (!addWorker(command, false))
reject(command);
}主要方法addWorker。
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 如果線程池狀態(tài)大于SHUTDOWN或者線程池狀態(tài)等于SHUTDOWN,firstTask不等于null
// 或者線程池狀態(tài)等于SHUTDOWN,任務隊列等于空時,直接返回false結(jié)束。
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
// 如果線程數(shù)量大于等于最大數(shù)量或者大于等于上限
//(入?yún)ore傳true,取核心線程數(shù),否則取最大線程數(shù)),直接返回false結(jié)束。
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false
// CAS操作給工作線程數(shù)加1,成功則跳到retry處,不再進入循環(huán)。
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
// 如果線程池狀態(tài)與剛進入時不一致,則跳到retry處,再次進入循環(huán)
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 {
// 新建一個線程
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int rs = runStateOf(ctl.get());
// 如果線程池狀態(tài)在SHUTDOWN之前或者
// 線程池狀態(tài)等于SHUTDOWN并且firstTask等于null時,進入處理。
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
// 如果要執(zhí)行的線程正在運行,則拋異常
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
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;
}任務執(zhí)行run方法
在上述addWorker中,當調(diào)用線程的start方法啟動線程后,會執(zhí)行其中的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 {
// 如果任務不為空或者新獲取到的任務不為空
while (task != null || (task = getTask()) != null) {
w.lock();
// 當線程池狀態(tài),大于等于 STOP 時,保證工作線程都有中斷標志。
// 當線程池狀態(tài),小于STOP時,保證工作線程都沒有中斷標志。
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
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);
}
}整體處理過程
通過上述源碼分析,我們可以得出線程池處理任務的過程如下:

總結(jié)
本文從源碼層面主要分析了線程池的創(chuàng)建、運行過程,通過上述的分析,可以看出當線程池中的線程數(shù)量超過核心線程數(shù)后,會先將任務放入等待隊列,隊列放滿后當最大線程數(shù)大于核心線程數(shù)時,才會創(chuàng)建新的線程執(zhí)行。
以上就是java ThreadPoolExecutor線程池內(nèi)部處理流程解析的詳細內(nèi)容,更多關(guān)于java ThreadPoolExecutor線程池的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
springboot整合sentinel接口熔斷的實現(xiàn)示例
為了防止慢接口導致的服務阻塞,可以通過添加熔斷處理來避免應用的大量工作線程陷入阻塞,保證其他接口的正常運行,本文介紹了如何使用Spring Boot與Sentinel進行接口熔斷的配置與實現(xiàn),感興趣的可以了解一下2024-09-09
關(guān)于Idea卡在Resolving Maven dependencies的解決方案
本文詳細介紹了關(guān)于Idea卡在Resolving Maven dependencies的解決方案,文中通過圖文結(jié)合的形式給大家介紹的非常詳細,對大家解決問題有一定的幫助,需要的朋友可以參考下2024-02-02
Springboot mybatis-plus配置及用法詳解
這篇文章主要介紹了Springboot mybatis-plus配置及用法詳解,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-09-09
Intellij IDEA 斷點不可用報錯 No executable 
這篇文章主要介紹了Intellij IDEA 斷點不可用報錯 No executable code found問題及解決方案,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-10-10

