Java線程池ThreadPoolExecutor源碼深入分析
更新時間：2022年08月26日 10:46:54 作者：niuyongzhi
ThreadPoolExecutor作為java.util.concurrent包對外提供基礎實現(xiàn)，以內(nèi)部線程池的形式對外提供管理任務執(zhí)行，線程調(diào)度，線程池管理等等服務
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)建固定大小的線程池，參數(shù)為int，是線程池核心線程和最大線程的數(shù)量
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)建一個線程數(shù)不設限的線程池，
//創(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允許的線程數(shù)量為最大值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 實現(xiàn) Runnable 接口。調(diào)用execute方法即可將入線程池，執(zhí)行run方法中的代碼。 executorService().execute(call);
3.ThreadPoolExecutor各個參數(shù)的含義
corePoolSize：核心線程數(shù)，即使是空閑線程也不會銷毀。這樣做的目的是為了降低執(zhí)行任務時創(chuàng)建線程的時間和性能開銷。
maximumPoolSize：最大線程數(shù)。當核心線程被用完時，會創(chuàng)建新的線程來執(zhí)行任務，但是創(chuàng)建的數(shù)量不能超過這個最大值。
keepAliveTime：線程的存活時間。除核心線程外，其他線程一旦執(zhí)行完任務，就會處于空閑狀態(tài)，超過這個時間就會被銷毀。
unit：keepAliveTime設置的時間單位。
workQueue：任務的阻塞隊列。線程數(shù)量有限，當任務過多來不及執(zhí)行時，就會加入到這個阻塞隊列中，等到有空閑進程，
就會從這個隊列取出任務去執(zhí)行。隊列都是先進先出的FIFO。
threadFactory：新線程產(chǎn)生的方式。
handler：拒絕策略，超過任務隊列設置的最大值時。再有新的任務進來，就會執(zhí)行這個拒絕策略。
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
}
線程池的阻塞隊列：
ArrayBlockingQueue：
是基于數(shù)組的任務隊列。里面用一個數(shù)組來存放任務。當我們new的時候，需要指定數(shù)組大小。
還有兩個int變量putIndex和takeIndex用來表示隊列的頭部和尾部在數(shù)組中的位置。
LinkedBlockingQueue:
是基于鏈表的，內(nèi)部用一個單向鏈表來存放任務。創(chuàng)建時可以指定大小，如果不指定則是Integer.MAX_VALUE
PriorityBlockingQueue:
基于優(yōu)先級的阻塞隊列。
SynchronousQueue：
一種無緩沖的等待隊列。有新任務進來直接交給線程執(zhí)行。
OkHttp中使用的就是這種隊列，他的最大線程數(shù)為Integer.MAX_VALUE。保證有任務進來就能馬上執(zhí)行。
RejectedExecutionHandler拒絕策略，這是一個接口。不同的實現(xiàn)執(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計數(shù)為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)和數(shù)量
runStateOf 計算當前線程池的狀態(tài)。
workerCountOf計算線程池的數(shù)量。
// ctlOf計算ctl的新值，也就是線程池狀態(tài)和線程池中線程數(shù)量。
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位，也就是線程池中的線程數(shù)。
private static int workerCountOf(int c) { return c & CAPACITY; }
其中runStateOf(int c)和workerCountOf(int c)的參數(shù)c就是通過ctlOf(int rs, int wc)獲得的ctl值。向線程池中添加一個任務：executorService().execute(call);
然后看看源碼中是如何執(zhí)行的，是如何添加任務的。
ctl 用來表示線程池的狀態(tài)和線程數(shù)量，
在ThreadPoolExcutor中使用32位二進制數(shù)來表示線程池的狀態(tài)和線程中線程數(shù)量。
其中前3位表示線程池的狀態(tài)，后29位表示線程池中的線程數(shù)。
public void execute(Runnable command) {
int c = ctl.get();
//如果工作線程數(shù)量小于核心線程數(shù)，
//提交的任務會通過addWorker(command, true)創(chuàng)建一個新的核心線程來執(zhí)行， 這個參數(shù)傳的是true，表示去新增核心線程。
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true)){
//添加成功則return
return;
}
//添加核心線程失敗則重新獲取線程池的狀態(tài)和數(shù)量
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);
}
}通過以上源碼分析，線程池的運行原理可以總結(jié)為一下幾點：
1.通過execute方法提交任務時，運行線程小于corePoolSize時，則會創(chuàng)建新的核心線程來執(zhí)行這個任務。
2.通過excute方法提交任務時，運行線程大于等于corePoolSize時，則會加入到隊列中，等待線程調(diào)度執(zhí)行。
3.通過excuete方法提交任務時，運行線程大于等于corePoolSize時，并且加入隊列失敗(隊列滿了)，新提交的任務將會通過創(chuàng)建新的線程執(zhí)行。
4.通過excute方法提交任務時，運行線程大于maximumPoolSize時，隊列也滿了，則會執(zhí)行拒絕策略。
5.當線程池中的線程執(zhí)行完任務處于空閑狀態(tài)時，則會嘗試從任務隊列中取頭結(jié)點任務執(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 (;;) {
//獲取工作線程數(shù)量
int wc = workerCountOf(c);
//wc >= CAPACITY 工作線程大于最大容量
// wc >= (core ? corePoolSize : maximumPoolSize) 如果工作線程大于了核心線程或最大線程，
//只要這兩個條件有一個成立則return。
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize)){
return false;
}
//創(chuàng)建線程數(shù)量+1，這里用到了CAS。關于CAS后面再寫文章分析。
if (compareAndIncrementWorkerCount(c)){
break retry;
}
//如果CAS操作失敗，線程數(shù)量沒有加1，則重新獲取線程的狀態(tài)。
c = ctl.get(); // Re-read ctl
//判斷當前狀態(tài)和之前狀態(tài)，如果不同，說明線程池狀態(tài)發(fā)生了變化。重新跳到retry的外層循環(huán)。
//如果相同，則說明線程池沒有變化，繼續(xù)進行內(nèi)層循環(huán)。
if (runStateOf(c) != rs){
continue retry;
}
// else CAS failed due to workerCount change; retry inner loop
}
}
//執(zhí)行到這說明線程數(shù)量已經(jīng)完成+1，接下來進行線程的創(chuàng)建。
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//這個創(chuàng)建一個worker對象。在worker構(gòu)造方法中，會利用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){ //更新一下最大線程數(shù)
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傳到了線程中去。當調(diào)用t.start()方法時會調(diào)用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;
}
}通過以上源碼分析，可以總結(jié)一下幾點。
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)、以及核心線程或最大線程數(shù)。
如果創(chuàng)建成功啟動線程的start方法，然后調(diào)用worker的runWorker()方法。
到此這篇關于Java線程池ThreadPoolExecutor源碼深入分析的文章就介紹到這了,更多相關Java ThreadPoolExecutor內(nèi)容請搜索腳本之家以前的文章或繼續(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
相關文章
這篇文章主要介紹了JAVA設計模式之裝飾模式的的相關資料，文中示例代碼非常詳細，供大家參考和學習，感興趣的朋友可以了解下2021-11-11
在設計接口時我們通常需要對接口中的非法參數(shù)做校驗,以降低在程序運行時因為一些非法參數(shù)而導致程序發(fā)生異常的風險,這篇文章給大家介紹Springboot接口參數(shù)校驗的方法,感興趣的朋友一起看看吧2024-03-03
Springboot實現(xiàn)多服務器session共享
這篇文章主要為大家詳細介紹了Springboot實現(xiàn)多服務器session共享，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-05-05
本文主要介紹Java中幾種分配內(nèi)存的方法。我們會看到如何使用sun.misc.Unsafe來統(tǒng)一操作任意類型的內(nèi)存。以前用C語言開發(fā)的同學通常都希望能在Java中通過較底層的接口來操作內(nèi)存，他們一定會對本文中要講的內(nèi)容感興趣2014-03-03
這篇文章主要介紹了Java中泛型的相關資料，幫助大家更好的理解和學習Java，感興趣的朋友可以了解下2020-08-08
Optional是Java實現(xiàn)函數(shù)式編程的強勁一步,并且?guī)椭诜妒街袑崿F(xiàn),但是Optional的意義顯然不止于此,下面這篇文章主要給大家介紹了關于Jackson處理Optional時遇到問題的解決與分析的相關資料,需要的朋友可以參考下2022-02-02
教你1秒將本地SpringBoot項目jar包部署到Linux環(huán)境(超詳細!)
spring Boot簡化了Spring應用的開發(fā)過程,遵循約定優(yōu)先配置的原則提供了各類開箱即用（out-of-the-box）的框架配置,下面這篇文章主要給大家介紹了關于1秒將本地SpringBoot項目jar包部署到Linux環(huán)境的相關資料,超級詳細,需要的朋友可以參考下2023-04-04
mybatis-plus 表名添加前綴的實現(xiàn)方法
這篇文章主要介紹了mybatis-plus 表名添加前綴的實現(xiàn)方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-08-08
SpringBoot項目實現(xiàn)統(tǒng)一異常處理的最佳方案
在前后端分離的項目開發(fā)過程中,我們通常會對數(shù)據(jù)返回格式進行統(tǒng)一的處理,這樣可以方便前端人員取數(shù)據(jù),后端發(fā)生異常時同樣會使用此格式將異常信息返回給前端,本文介紹了如何在SpringBoot項目中實現(xiàn)統(tǒng)一異常處理,如有錯誤,還望批評指正2024-02-02
Javaweb監(jiān)聽器實例之統(tǒng)計在線人數(shù)
這篇文章主要為大家詳細介紹了Javaweb監(jiān)聽器實例之統(tǒng)計在線人數(shù)，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-11-11