溫馨提示×
您好，登錄后才能下訂單哦！
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶服務(wù)條款》
您好，登錄后才能下訂單哦！
/**
* 使用給定的參數(shù)創(chuàng)建ThreadPoolExecutor.
*
* @param corePoolSize 核心線程池中的最大線程數(shù)
* @param maximumPoolSize 總線程池中的最大線程數(shù)
* @param keepAliveTime 空閑線程的存活時(shí)間
* @param unit keepAliveTime的單位
* @param workQueue 任務(wù)隊(duì)列, 保存已經(jīng)提交但尚未被執(zhí)行的任務(wù)
* @param threadFactory 線程工廠(用于指定如果創(chuàng)建一個(gè)線程)
* @param handler 拒絕策略 (當(dāng)任務(wù)太多導(dǎo)致工作隊(duì)列滿時(shí)的處理策略)
*/
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.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime); // 使用納秒保存存活時(shí)間
this.threadFactory = threadFactory;
this.handler = handler;
}1、ThreadPoolExecutor在邏輯上將自身管理的線程池劃分為兩部分：核心線程池（大小對(duì)應(yīng)為corePoolSize）、非核心線程池（大小對(duì)應(yīng)為maximumPoolSize-corePoolSize）。
2、ThreadPoolExecutor中只有一種類型的線程，名叫Worker，它是ThreadPoolExecutor內(nèi)部類，封裝著Runnable任務(wù)和執(zhí)行該任務(wù)的Thread對(duì)象，我們稱它為【工作線程】，它也是ThreadPoolExecutor唯一需要進(jìn)行維護(hù)的線程；
3、【核心線程池】【非核心線程池】都是邏輯上的概念，ThreadPoolExecutor在任務(wù)調(diào)度過程中會(huì)根據(jù)corePoolSize和maximumPoolSize的大小，判斷如何執(zhí)行任務(wù)。
1、低29位保存線程數(shù)
2、高3位保存線程池狀態(tài)
//保存線程池狀態(tài)和工作線程數(shù):低29位: 工作線程數(shù),高3位 : 線程池狀態(tài)
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
// 最大線程數(shù): 2^29-1
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// 線程池狀態(tài)
private static final int RUNNING = -1 << COUNT_BITS; //接受新任務(wù), 且處理已經(jīng)進(jìn)入阻塞隊(duì)列的任務(wù)
private static final int SHUTDOWN = 0 << COUNT_BITS; //不接受新任務(wù), 但處理已經(jīng)進(jìn)入阻塞隊(duì)列的任務(wù)
private static final int STOP = 1 << COUNT_BITS; //不接受新任務(wù), 且不處理已經(jīng)進(jìn)入阻塞隊(duì)列的任務(wù), 同時(shí)中斷正在運(yùn)行的任務(wù)
private static final int TIDYING = 2 << COUNT_BITS; //所有任務(wù)都已終止, 工作線程數(shù)為0, 線程轉(zhuǎn)化為TIDYING狀態(tài)并準(zhǔn)備調(diào)用terminated方法
private static final int TERMINATED = 3 << COUNT_BITS; //terminated方法已經(jīng)執(zhí)行完成
/**
* Worker表示線程池中的一個(gè)工作線程, 可以與任務(wù)相關(guān)聯(lián).
* 由于實(shí)現(xiàn)了AQS框架, 其同步狀態(tài)值的定義如下:
* -1: 初始狀態(tài)
* 0: 無鎖狀態(tài)
* 1: 加鎖狀態(tài)
*/
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
/**
* 與該Worker關(guān)聯(lián)的線程.
*/
final Thread thread;
/**
*初始化任務(wù)，可以為空，為空的時(shí)候則去任務(wù)隊(duì)列workQueue里獲取
*/
Runnable firstTask;
/**
* 當(dāng)前工作線程處理完成的任務(wù)數(shù)
*/
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // 初始的同步狀態(tài)值
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
/**
* 執(zhí)行任務(wù)
*/
public void run() {
runWorker(this);
}
/**
* 是否加鎖
*/
protected boolean isHeldExclusively() {
return getState() != 0;
}
/**
* 嘗試獲取鎖
*/
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
/**
* 嘗試釋放鎖
*/
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
public void lock() {
acquire(1);
}
public boolean tryLock() {
return tryAcquire(1);
}
public void unlock() {
release(1);
}
public boolean isLocked() {
return isHeldExclusively();
}
/**
* 中斷線程(僅任務(wù)非初始狀態(tài))
*/
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
//獲取線程池狀態(tài)和工作線程數(shù)量
int c = ctl.get();
//如果工作線程數(shù) < 核心線程數(shù)，則創(chuàng)建工作線程
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true)) //addWorker創(chuàng)建工作線程
return;
c = ctl.get();
}
//工作線程創(chuàng)建失敗，或者，工作線程 >= 核心線程數(shù)，任務(wù)插入任務(wù)隊(duì)列workQueue
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
//再次檢查線程池狀態(tài)，如果不是運(yùn)行狀態(tài)，移除任務(wù)，并拒絕
if (! isRunning(recheck) && remove(command))
reject(command);
//如果工作線程為0，則創(chuàng)建一個(gè)不帶任務(wù)的線程，線程自動(dòng)去任務(wù)隊(duì)列獲取任務(wù)執(zhí)行
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//任務(wù)插入隊(duì)列失敗，創(chuàng)建非核心工作線程，如果失敗，則說明工作線程 > 總線程數(shù)量，則執(zhí)行拒絕策略
else if (!addWorker(command, false))
reject(command);
}
說明：
1、如果工作線程數(shù)小于核心線程池上限（CorePoolSize），則直接新建一個(gè)工作線程并執(zhí)行任務(wù)；
2、如果工作線程數(shù)大于等于CorePoolSize，則嘗試將任務(wù)加入到隊(duì)列等待以后執(zhí)行。如果隊(duì)列已滿，則在總線程池未滿的情況下（CorePoolSize ≤ 工作線程數(shù) ＜ maximumPoolSize）新建一個(gè)工作線程立即執(zhí)行任務(wù)，否則執(zhí)行拒絕策略。
/**
* 添加工作線程并執(zhí)行任務(wù)
*
* @param firstTask 如果指定了該參數(shù), 表示將立即執(zhí)行該firstTask任務(wù); 否則從工作隊(duì)列中獲取任務(wù)并執(zhí)行
* @param core 執(zhí)行任務(wù)的工作線程歸屬于哪個(gè)線程池: true-核心線程池 false-非核心線程池
*/
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
/**
* 這個(gè)if主要是判斷哪些情況下, 線程池不再接受新任務(wù)執(zhí)行, 而是直接返回.總結(jié)下, 有以下幾種情況：
* 1. 線程池狀態(tài)為 STOP 或 TIDYING 或 TERMINATED: 線程池狀態(tài)為上述任一一種時(shí), 都不會(huì)再接受任務(wù)，所以直接返回
* 2. 線程池狀態(tài)≥ SHUTDOWN 且 firstTask != null: 因?yàn)楫?dāng)線程池狀態(tài)≥ SHUTDOWN時(shí), 不再接受新任務(wù)的提交，所以直接返回
* 3. 線程池狀態(tài)≥ SHUTDOWN 且 隊(duì)列為空: 隊(duì)列中已經(jīng)沒有任務(wù)了, 所以也就不需要執(zhí)行任何任務(wù)了，可以直接返回
*/
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
//獲取工作線程數(shù)
int wc = workerCountOf(c);
/**
* 這個(gè)if主要是判斷工作線程數(shù)是否超限, 以下任一情況屬于超限, 直接返回:
* 1. 工作線程數(shù)超過最大工作線程數(shù)(2^29-1)
* 2. 工作線程數(shù)超過核心線程池上限(入?yún)ore為true, 表示歸屬核心線程池)
* 3. 工作線程數(shù)超過總線程池上限(入?yún)ore為false, 表示歸屬非核心線程池)
*/
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//增加工作線程數(shù)
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get();
//CAS失敗，自旋重新操作
if (runStateOf(c) != rs)
continue retry;
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//將任務(wù)包裝成工作線程
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());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
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) {
//啟動(dòng)線程，這個(gè)很關(guān)鍵，因?yàn)閠在new的時(shí)候是用Work作為任務(wù)的，
// Work實(shí)現(xiàn)了Runnale接口，所以t.start就是執(zhí)行Work的run方法
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}特別注意：
啟動(dòng)線程，這個(gè)很關(guān)鍵，因?yàn)閠hread在new的時(shí)候是用Work（this）作為任務(wù)的, Work實(shí)現(xiàn)了Runnale接口，所以t.start就是執(zhí)行Work的run方法。
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
// 任務(wù), 如果是null則從隊(duì)列取任務(wù)
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // 允許執(zhí)行線程被中斷
boolean completedAbruptly = true;
try {
// 當(dāng)task==null時(shí)會(huì)通過getTask從隊(duì)列取任務(wù)
while (task != null || (task = getTask()) != null) {
w.lock();
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
// 鉤子方法，由子類自定義實(shí)現(xiàn)
beforeExecute(wt, task);
Throwable thrown = null;
try {
// 執(zhí)行任務(wù)
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 {
// 鉤子方法，由子類自定義實(shí)現(xiàn)
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++; // 完成任務(wù)數(shù)+1
w.unlock();
}
}
//說明該工作線程自身既沒有攜帶任務(wù), 也沒從任務(wù)隊(duì)列中獲取到任務(wù)
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}獲取任務(wù)方法getTask
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
/**
* 以下IF用于判斷哪些情況下不允許再從隊(duì)列獲取任務(wù):
* 1. 線程池進(jìn)入停止?fàn)顟B(tài)（STOP/TIDYING/TERMINATED）, 此時(shí)即使隊(duì)列中還有任務(wù)未執(zhí)行, 也不再執(zhí)行
* 2. 線程池非RUNNING狀態(tài), 且隊(duì)列為空
*/
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
/**
* timed變量用于判斷是否需要進(jìn)行超時(shí)控制:
* 對(duì)于核心線程池中的工作線程, 除非設(shè)置了allowCoreThreadTimeOut==true, 否則不會(huì)超時(shí)回收;
* 對(duì)于非核心線程池中的工作線程, 都需要超時(shí)控制
*/
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// 這里主要是當(dāng)外部通過setMaximumPoolSize方法重新設(shè)置了最大線程數(shù)時(shí)
// 需要回收多出的工作線程
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
//區(qū)分超時(shí)操作還是非超時(shí)獲取
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}所謂拒絕策略，就是在構(gòu)造ThreadPoolExecutor時(shí)，傳入的RejectedExecutionHandler對(duì)象，一共4種
1.AbortPolicy（默認(rèn)），拋出一個(gè)RejectedExecutionException異常
2、DiscardPolicy，什么都不做，等任務(wù)自己被回收
3、DiscardOldestPolicy，丟棄任務(wù)隊(duì)列中的最近一個(gè)任務(wù)，并執(zhí)行當(dāng)前任務(wù)
4、CallerRunsPolicy，以自身線程來執(zhí)行任務(wù),減緩新任務(wù)提交的速度
免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場，如果涉及侵權(quán)請(qǐng)聯(lián)系站長郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。
億速云公眾號(hào)
手機(jī)網(wǎng)站二維碼