Java线程池框架核心代码分析
前言
多線程編程中,為每個(gè)任務(wù)分配一個(gè)線程是不現(xiàn)實(shí)的,線程創(chuàng)建的開銷和資源消耗都是很高的。線程池應(yīng)運(yùn)而生,成為我們管理線程的利器。Java 通過Executor接口,提供了一種標(biāo)準(zhǔn)的方法將任務(wù)的提交過程和執(zhí)行過程解耦開來,并用Runnable表示任務(wù)。
下面,我們來分析一下 Java 線程池框架的實(shí)現(xiàn)ThreadPoolExecutor。
下面的分析基于JDK1.7
生命周期
ThreadPoolExecutor中,使用CAPACITY的高3位來表示運(yùn)行狀態(tài),分別是:
狀態(tài)轉(zhuǎn)換圖
ThreadPoolExecutor中用原子類來表示狀態(tài)位
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));線程池模型
核心參數(shù)
- corePoolSize:最小存活的工作線程數(shù)量(如果設(shè)置allowCoreThreadTimeOut,那么該值為 0)
- maximumPoolSize:最大的線程數(shù)量,受限于CAPACITY
- keepAliveTime:對應(yīng)線程的存活時(shí)間,時(shí)間單位由TimeUnit指定
- workQueue:工作隊(duì)列,存儲待執(zhí)行的任務(wù)
- RejectExecutionHandler:拒絕策略,線程池滿后會觸發(fā)
線程池的最大容量:CAPACITY中的前三位用作標(biāo)志位,也就是說工作線程的最大容量為(2^29)-1
四種模型
- CachedThreadPool:一個(gè)可緩存的線程池,如果線程池的當(dāng)前規(guī)模超過了處理需求時(shí),那么將回收空閑的線程,當(dāng)需求增加時(shí),則可以添加新的線程,線程池的規(guī)模不存在任何的限制。
- FixedThreadPool:一個(gè)固定大小的線程池,提交一個(gè)任務(wù)時(shí)就創(chuàng)建一個(gè)線程,直到達(dá)到線程池的最大數(shù)量,這時(shí)線程池的大小將不再變化。
- SingleThreadPool:一個(gè)單線程的線程池,它只有一個(gè)工作線程來執(zhí)行任務(wù),可以確保按照任務(wù)在隊(duì)列中的順序來串行執(zhí)行,如果這個(gè)線程異常結(jié)束將創(chuàng)建一個(gè)新的線程來執(zhí)行任務(wù)。
- ScheduledThreadPool:一個(gè)固定大小的線程池,并且以延遲或者定時(shí)的方式來執(zhí)行任務(wù),類似于Timer。
執(zhí)行任務(wù) execute
核心邏輯:
從上面的分析可以總結(jié)出線程池運(yùn)行的四個(gè)階段:
拒絕策略
前面我們提到任務(wù)無法執(zhí)行會被拒絕,RejectedExecutionHandler是處理被拒絕任務(wù)的接口。下面是四種拒絕策略。
- AbortPolicy:默認(rèn)策略,終止任務(wù),拋出RejectedException
- CallerRunsPolicy:在調(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
下面是簡化的邏輯,注意:每個(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); }獲取任務(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ù),四種情況會導(dǎo)致返回null,讓worker關(guān)閉。
核心邏輯:根據(jù)timed在阻塞隊(duì)列上超時(shí)等待或者阻塞等待任務(wù),等待任務(wù)超時(shí)會導(dǎo)致工作線程被關(guān)閉。
timed = allowCoreThreadTimeOut || wc > corePoolSize; Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();在以下兩種情況下等待任務(wù)會超時(shí):
工作隊(duì)列使用的是BlockingQueue,這里就不展開了,后面再寫一篇詳細(xì)的分析。
總結(jié)
- ThreadPoolExecutor基于生產(chǎn)者-消費(fèi)者模式,提交任務(wù)的操作相當(dāng)于生產(chǎn)者,執(zhí)行任務(wù)的線程相當(dāng)于消費(fèi)者。
- Executors提供了四種基于ThreadPoolExecutor構(gòu)造線程池模型的方法,除此之外,我們還可以直接繼承ThreadPoolExecutor,重寫beforeExecute和afterExecute方法來定制線程池任務(wù)執(zhí)行過程。
- 使用有界隊(duì)列還是無界隊(duì)列需要根據(jù)具體情況考慮,工作隊(duì)列的大小和線程的數(shù)量也是需要好好考慮的。
- 拒絕策略推薦使用CallerRunsPolicy,該策略不會拋棄任務(wù),也不會拋出異常,而是將任務(wù)回退到調(diào)用者線程中執(zhí)行。
總結(jié)
以上是生活随笔為你收集整理的Java线程池框架核心代码分析的全部內(nèi)容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: 《Head first设计模式》学习笔记
- 下一篇: Javadoc注释规范