Java線程池框架核心代碼分析
前言
多線程編程中,為每個(gè)任務(wù)分配一個(gè)線程是不現(xiàn)實(shí)的,線程創(chuàng)建的開(kāi)銷(xiāo)和資源消耗都是很高的。線程池應(yīng)運(yùn)而生,成為我們管理線程的利器。Java 通過(guò)Executor
接口,提供了一種標(biāo)準(zhǔn)的方法將任務(wù)的提交過(guò)程和執(zhí)行過(guò)程解耦開(kāi)來(lái),并用Runnable
表示任務(wù)。
下面,我們來(lái)分析一下 Java 線程池框架的實(shí)現(xiàn)ThreadPoolExecutor
。
下面的分析基于JDK1.7
生命周期
ThreadPoolExecutor
中,使用CAPACITY
的高3位來(lái)表示運(yùn)行狀態(tài),分別是:
- RUNNING:接收新任務(wù),并且處理任務(wù)隊(duì)列中的任務(wù)
- SHUTDOWN:不接收新任務(wù),但是處理任務(wù)隊(duì)列的任務(wù)
- STOP:不接收新任務(wù),不出來(lái)任務(wù)隊(duì)列,同時(shí)中斷所有進(jìn)行中的任務(wù)
- TIDYING:所有任務(wù)已經(jīng)被終止,工作線程數(shù)量為 0,到達(dá)該狀態(tài)會(huì)執(zhí)行
terminated()
- TERMINATED:
terminated()
執(zhí)行完畢
狀態(tài)轉(zhuǎn)換圖
ThreadPoolExecutor
中用原子類(lèi)來(lái)表示狀態(tài)位
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
線程池模型
核心參數(shù)
corePoolSize
:最小存活的工作線程數(shù)量(如果設(shè)置allowCoreThreadTimeOut
,那么該值為 0)maximumPoolSize
:最大的線程數(shù)量,受限于CAPACITY
keepAliveTime
:對(duì)應(yīng)線程的存活時(shí)間,時(shí)間單位由TimeUnit指定workQueue
:工作隊(duì)列,存儲(chǔ)待執(zhí)行的任務(wù)RejectExecutionHandler
:拒絕策略,線程池滿(mǎn)后會(huì)觸發(fā)
線程池的最大容量:CAPACITY
中的前三位用作標(biāo)志位,也就是說(shuō)工作線程的最大容量為(2^29)-1
四種模型
CachedThreadPool
:一個(gè)可緩存的線程池,如果線程池的當(dāng)前規(guī)模超過(guò)了處理需求時(shí),那么將回收空閑的線程,當(dāng)需求增加時(shí),則可以添加新的線程,線程池的規(guī)模不存在任何的限制。FixedThreadPool
:一個(gè)固定大小的線程池,提交一個(gè)任務(wù)時(shí)就創(chuàng)建一個(gè)線程,直到達(dá)到線程池的最大數(shù)量,這時(shí)線程池的大小將不再變化。SingleThreadPool
:一個(gè)單線程的線程池,它只有一個(gè)工作線程來(lái)執(zhí)行任務(wù),可以確保按照任務(wù)在隊(duì)列中的順序來(lái)串行執(zhí)行,如果這個(gè)線程異常結(jié)束將創(chuàng)建一個(gè)新的線程來(lái)執(zhí)行任務(wù)。ScheduledThreadPool
:一個(gè)固定大小的線程池,并且以延遲或者定時(shí)的方式來(lái)執(zhí)行任務(wù),類(lèi)似于Timer。
執(zhí)行任務(wù) execute
核心邏輯:
- 當(dāng)前線程數(shù)量 <
corePoolSize
,直接開(kāi)啟新的核心線程執(zhí)行任務(wù)addWorker(command, true)
- 當(dāng)前線程數(shù)量 >=
corePoolSize
,且任務(wù)加入工作隊(duì)列成功- 檢查線程池當(dāng)前狀態(tài)是否處于
RUNNING
- 如果否,則拒絕該任務(wù)
- 如果是,判斷當(dāng)前線程數(shù)量是否為 0,如果為 0,就增加一個(gè)工作線程。
- 檢查線程池當(dāng)前狀態(tài)是否處于
- 開(kāi)啟普通線程執(zhí)行任務(wù)
addWorker(command, false)
,開(kāi)啟失敗就拒絕該任務(wù)
從上面的分析可以總結(jié)出線程池運(yùn)行的四個(gè)階段:
poolSize < corePoolSize
且隊(duì)列為空,此時(shí)會(huì)新建線程來(lái)處理提交的任務(wù)poolSize == corePoolSize
,此時(shí)提交的任務(wù)進(jìn)入工作隊(duì)列,工作線程從隊(duì)列中獲取任務(wù)執(zhí)行,此時(shí)隊(duì)列不為空且未滿(mǎn)。poolSize == corePoolSize
,并且隊(duì)列已滿(mǎn),此時(shí)也會(huì)新建線程來(lái)處理提交的任務(wù),但是poolSize < maxPoolSize
poolSize == maxPoolSize
,并且隊(duì)列已滿(mǎn),此時(shí)會(huì)觸發(fā)拒絕策略
拒絕策略
前面我們提到任務(wù)無(wú)法執(zhí)行會(huì)被拒絕,RejectedExecutionHandler
是處理被拒絕任務(wù)的接口。下面是四種拒絕策略。
AbortPolicy
:默認(rèn)策略,終止任務(wù),拋出RejectedExceptionCallerRunsPolicy
:在調(diào)用者線程執(zhí)行當(dāng)前任務(wù),不拋異常DiscardPolicy
: 拋棄策略,直接丟棄任務(wù),不拋異常DiscardOldersPolicy
:拋棄最老的任務(wù),執(zhí)行當(dāng)前任務(wù),不拋異常
線程池中的 Worker
Worker
繼承了AbstractQueuedSynchronizer
和Runnable
,前者給Worker
提供鎖的功能,后者執(zhí)行工作線程的主要方法runWorker(Worker w)
(從任務(wù)隊(duì)列撈任務(wù)執(zhí)行)。Worker 引用存在workers
集合里面,用mainLock
守護(hù)。
private final ReentrantLock mainLock = new ReentrantLock();
private final HashSet<Worker> workers = new HashSet<Worker>();
核心函數(shù) runWorker
下面是簡(jiǎn)化的邏輯,注意:每個(gè)工作線程的run
都執(zhí)行下面的函數(shù)
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
while (task != null || (task = getTask()) != null) {
w.lock();
beforeExecute(wt, task);
task.run();
afterExecute(task, thrown);
w.unlock();
}
processWorkerExit(w, completedAbruptly);
}
- 從
getTask()
中獲取任務(wù) - 鎖住 worker
- 執(zhí)行
beforeExecute(wt, task)
,這是ThreadPoolExecutor
提供給子類(lèi)的擴(kuò)展方法 - 運(yùn)行任務(wù),如果該worker有配置了首次任務(wù),則先執(zhí)行首次任務(wù)且只執(zhí)行一次。
- 執(zhí)行
afterExecute(task, thrown);
- 解鎖 worker
- 如果獲取到的任務(wù)為 null,關(guān)閉 worker
獲取任務(wù) getTask
線程池內(nèi)部的任務(wù)隊(duì)列是一個(gè)阻塞隊(duì)列,具體實(shí)現(xiàn)在構(gòu)造時(shí)傳入。
private final BlockingQueue<Runnable> workQueue;
getTask()
從任務(wù)隊(duì)列中獲取任務(wù),支持阻塞和超時(shí)等待任務(wù),四種情況會(huì)導(dǎo)致返回null
,讓worker
關(guān)閉。
- 現(xiàn)有的線程數(shù)量超過(guò)最大線程數(shù)量
- 線程池處于
STOP
狀態(tài) - 線程池處于
SHUTDOWN
狀態(tài)且工作隊(duì)列為空 - 線程等待任務(wù)超時(shí),且線程數(shù)量超過(guò)保留線程數(shù)量
核心邏輯:根據(jù)timed
在阻塞隊(duì)列上超時(shí)等待或者阻塞等待任務(wù),等待任務(wù)超時(shí)會(huì)導(dǎo)致工作線程被關(guān)閉。
timed = allowCoreThreadTimeOut || wc > corePoolSize;
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
在以下兩種情況下等待任務(wù)會(huì)超時(shí):
- 允許核心線程等待超時(shí),即
allowCoreThreadTimeOut(true)
- 當(dāng)前線程是普通線程,此時(shí)
wc > corePoolSize
工作隊(duì)列使用的是BlockingQueue
,這里就不展開(kāi)了,后面再寫(xiě)一篇詳細(xì)的分析。
總結(jié)
ThreadPoolExecutor
基于生產(chǎn)者-消費(fèi)者模式,提交任務(wù)的操作相當(dāng)于生產(chǎn)者,執(zhí)行任務(wù)的線程相當(dāng)于消費(fèi)者。Executors
提供了四種基于ThreadPoolExecutor
構(gòu)造線程池模型的方法,除此之外,我們還可以直接繼承ThreadPoolExecutor
,重寫(xiě)beforeExecute
和afterExecute
方法來(lái)定制線程池任務(wù)執(zhí)行過(guò)程。- 使用有界隊(duì)列還是無(wú)界隊(duì)列需要根據(jù)具體情況考慮,工作隊(duì)列的大小和線程的數(shù)量也是需要好好考慮的。
- 拒絕策略推薦使用
CallerRunsPolicy
,該策略不會(huì)拋棄任務(wù),也不會(huì)拋出異常,而是將任務(wù)回退到調(diào)用者線程中執(zhí)行。