溫馨提示×
您好，登錄后才能下訂單哦！
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶服務(wù)條款》
您好，登錄后才能下訂單哦！
本篇內(nèi)容主要講解“Java的線程池的工作原理”，感興趣的朋友不妨來(lái)看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓小編來(lái)帶大家學(xué)習(xí)“Java的線程池的工作原理”吧!
相信大家都用過(guò)線程池，對(duì)該類ThreadPoolExecutor應(yīng)該一點(diǎn)都不陌生了；
我們之所以要用到線程池，線程池主要用來(lái)解決線程生命周期開(kāi)銷問(wèn)題和資源不足問(wèn)題；
我們通過(guò)對(duì)多個(gè)任務(wù)重用線程以及控制線程池的數(shù)目可以有效防止資源不足的情況；
本章節(jié)就著重和大家分享分析一下JDK8的ThreadPoolExecutor核心類，看看線程池是如何工作的；
線程池在內(nèi)部實(shí)際上構(gòu)建了一個(gè)生產(chǎn)者消費(fèi)者模型，將任務(wù)的提交和任務(wù)執(zhí)行兩者解耦，并不直接關(guān)聯(lián)，從而良好的緩沖任務(wù)，復(fù)用線程。
線程池的運(yùn)行主要分成兩部分：任務(wù)管理、線程管理。
任務(wù)管理充當(dāng)生產(chǎn)者的角色。
當(dāng)任務(wù)提交后，線程池會(huì)判斷該任務(wù)后續(xù)的流轉(zhuǎn)：
直接申請(qǐng)線程執(zhí)行該任務(wù)；
緩沖到隊(duì)列中等待線程執(zhí)行；
拒絕該任務(wù)。
線程管理是消費(fèi)者的角色
它們被統(tǒng)一維護(hù)在線程池內(nèi)，根據(jù)任務(wù)請(qǐng)求進(jìn)行線程的分配
當(dāng)線程執(zhí)行完任務(wù)后則會(huì)繼續(xù)獲取新的任務(wù)去執(zhí)行，
最終當(dāng)線程獲取不到任務(wù)的時(shí)候，線程就會(huì)被回收。
接下來(lái)，我們會(huì)按照以下三個(gè)部分去詳細(xì)講解線程池運(yùn)行機(jī)制：
線程池如何維護(hù)自身狀態(tài)。
線程池如何管理任務(wù)。
線程池如何管理線程。
// 構(gòu)造器一
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
// 構(gòu)造器二
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
threadFactory, defaultHandler);
}
// 構(gòu)造器三
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), handler);
}
// 構(gòu)造器四
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);
this.threadFactory = threadFactory;
this.handler = handler;
}通過(guò)仔細(xì)查看構(gòu)造器代碼，發(fā)現(xiàn)最終都是調(diào)用構(gòu)造器四，緊接著賦值了一堆的字段，接下來(lái)我們先看看這些字段是什么含義；
corePoolSize：核心運(yùn)行的線程池?cái)?shù)量大小，當(dāng)線程數(shù)量超過(guò)該值時(shí)，就需要將超過(guò)該數(shù)量值的線程放到等待隊(duì)列中；
maximumPoolSize：線程池最大能容納的線程數(shù)(該數(shù)量已經(jīng)包含了corePoolSize數(shù)量)，當(dāng)線程數(shù)量超過(guò)該值時(shí)，則會(huì)拒絕執(zhí)行處理策略；
workQueue：等待隊(duì)列，當(dāng)達(dá)到corePoolSize的時(shí)候，就將新加入的線程追加到workQueue該等待隊(duì)列中；當(dāng)然BlockingQueue類也是一個(gè)抽象類，也有很多子類來(lái)實(shí)現(xiàn)不同的隊(duì)列等待， 一般來(lái)說(shuō)，阻塞隊(duì)列有一下幾種：
ArrayBlockingQueue
LinkedBlockingQueue
SynchronousQueue
DelayedWorkQueue
ArrayBlockingQueue，PriorityBlockingQueue使用較少，一般使用LinkedBlockingQueue和Synchronous。
keepAliveTime：表示線程沒(méi)有任務(wù)執(zhí)行時(shí)最多保持多久存活時(shí)間。
默認(rèn)情況下當(dāng)線程數(shù)量大于corePoolSize后keepAliveTime才會(huì)起作用，并生效，一旦線程池的數(shù)量小于corePoolSize后keepAliveTime又不起作用了；
但是如果調(diào)用了 allowCoreThreadTimeOut(boolean) 方法，在線程池中的線程數(shù)不大于corePoolSize時(shí)，keepAliveTime參數(shù)也會(huì)起作用，直到線程池中的線程數(shù)為0；
threadFactory：新創(chuàng)建線程出生的地方；
handler：拒絕執(zhí)行處理抽象類，就是說(shuō)當(dāng)線程池在一些場(chǎng)景中，不能處理新加入的線程任務(wù)時(shí)，會(huì)通過(guò)該對(duì)象處理拒絕策略；
策略一( CallerRunsPolicy )：只要線程池沒(méi)關(guān)閉，就直接用調(diào)用者所在線程來(lái)運(yùn)行任務(wù)；
策略二( AbortPolicy )：默認(rèn)策略，直接拋出RejectedExecutionException異常；
策略三( DiscardPolicy )：執(zhí)行空操作，什么也不干，拒絕任務(wù)后也不做任何回應(yīng)；
策略四( DiscardOldestPolicy )：將隊(duì)列中存活最久的那個(gè)未執(zhí)行的任務(wù)拋棄掉，然后將當(dāng)前新的線程放進(jìn)去（LRU）；
該對(duì)象RejectedExecutionHandler有四個(gè)實(shí)現(xiàn)類，即四種策略，讓我們有選擇性的在什么場(chǎng)景下該怎么使用拒絕策略；
largestPoolSize：變量記錄了線程池在整個(gè)生命周期中曾經(jīng)出現(xiàn)的最大線程個(gè)數(shù)；
allowCoreThreadTimeOut：當(dāng)為true時(shí)，核心線程也有超時(shí)退出的概念一說(shuō)；
線程池運(yùn)行的狀態(tài)，并不是用戶顯式設(shè)置的，而是伴隨著線程池的運(yùn)行，由內(nèi)部來(lái)維護(hù)。
線程池內(nèi)部使用一個(gè)變量維護(hù)兩個(gè)值：運(yùn)行狀態(tài)(runState)和線程數(shù)量 (workerCount)，兩個(gè)關(guān)鍵參數(shù)的維護(hù)放在了一起。
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
ctl 這個(gè)AtomicInteger類型，是對(duì)線程池的運(yùn)行狀態(tài)和線程池中有效線程的數(shù)量進(jìn)行控制的一個(gè)字段。
原子變量值，是一個(gè)復(fù)核類型的成員變量，是一個(gè)原子整數(shù)，借助高低位包裝了兩個(gè)概念，它同時(shí)包含兩部分的信息：線程池的運(yùn)行狀態(tài) (runState) 和線程池內(nèi)有效線程的數(shù)量 (workerCount)，高3位保存runState，低29位保存workerCount，兩個(gè)變量之間互不干擾。
用一個(gè)變量去存儲(chǔ)兩個(gè)值，可避免在做相關(guān)決策時(shí)，出現(xiàn)不一致的情況，不必為了維護(hù)兩者的一致，而占用鎖資源。
線程池也提供了若干方法去供用戶獲得線程池當(dāng)前的運(yùn)行狀態(tài)、線程個(gè)數(shù)。這里都使用的是位運(yùn)算的方式，相比于基本運(yùn)算，速度也會(huì)快很多。
關(guān)于內(nèi)部封裝的獲取生命周期狀態(tài)、獲取線程池線程數(shù)量的計(jì)算方法如以下代碼所示：
// 偏移位數(shù)，常量值29，之所以偏移29，目的是將32位的原子變量值ctl的高3位設(shè)置為
// 線程池的狀態(tài)，低29位作為線程池大小數(shù)量值；32-3
private static final int COUNT_BITS = Integer.SIZE - 3;
//低29位都為1，高位都為0
// 線程池的最大容量值
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// runState is stored in the high-order bits
// 接受新任務(wù)，并處理隊(duì)列任務(wù)
private static final int RUNNING = -1 << COUNT_BITS;//111
// 不接受新任務(wù)，但會(huì)處理隊(duì)列任務(wù)
private static final int SHUTDOWN = 0 << COUNT_BITS;//000
// 不接受新任務(wù)，不會(huì)處理隊(duì)列任務(wù)，而且會(huì)中斷正在處理過(guò)程中的任務(wù)
private static final int STOP = 1 << COUNT_BITS;//001
// 所有的任務(wù)已結(jié)束，workerCount為0，線程過(guò)渡到TIDYING狀態(tài)，將會(huì)執(zhí)行terminated()鉤子方法
private static final int TIDYING = 2 << COUNT_BITS;//010
// terminated()方法已經(jīng)完成
private static final int TERMINATED = 3 << COUNT_BITS;//011
// Packing and unpacking ctl線程池的狀態(tài)，原子變量值ctl的高三位
//計(jì)算當(dāng)前運(yùn)行狀態(tài)，取高三位
private static int runStateOf(int c) { return c & ~CAPACITY; }
//計(jì)算當(dāng)前線程數(shù)量，取低29位
private static int workerCountOf(int c) { return c & CAPACITY; }
//通過(guò)狀態(tài)和線程數(shù)生成ctl
private static int ctlOf(int rs, int wc) { return rs | wc; }
HashSet<Worker> workers = new HashSet<Worker>();
// 存放工作線程的線程池；ThreadPoolExecutor的運(yùn)行狀態(tài)有5種，分別為：
成員方法
// 提交任務(wù)，添加Runnable對(duì)象到線程池，由線程池調(diào)度執(zhí)行
public void execute(Runnable command)
// c & 高3位為0，低29位為1的CAPACITY，用于獲取低29位的線程數(shù)量
private static int workerCountOf(int c) { return c & CAPACITY; }
// 添加worker工作線程，根據(jù)邊界值來(lái)決定是否創(chuàng)建新的線程
private boolean addWorker(Runnable firstTask, boolean core)
// c通常一般為ctl，ctl值小于0，則處于可以接受新任務(wù)狀態(tài)
private static boolean isRunning(int c)
// 拒絕執(zhí)行任務(wù)方法，當(dāng)線程池在一些場(chǎng)景中，不能處理新加入的線程時(shí)，會(huì)通過(guò)該對(duì)象處理拒絕策略；
final void reject(Runnable command)
// 該方法被Worker工作線程的run方法調(diào)用，真正核心處理Runnable任務(wù)的方法
final void runWorker(Worker w)
// c & 高3位為1，低29位為0的~CAPACITY，用于獲取高3位保存的線程池狀態(tài)
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 不會(huì)立即終止線程池，而是要等所有任務(wù)緩存隊(duì)列中的任務(wù)都執(zhí)行完后才終止，但再也不會(huì)接受新的任務(wù) void shutdown()
public void shutdown()
// 立即終止線程池，并嘗試打斷正在執(zhí)行的任務(wù)，并且清空任務(wù)緩存隊(duì)列，返回尚未執(zhí)行的任務(wù)
public List<Runnable> shutdownNow()
// worker線程退出
private void processWorkerExit(Worker w, boolean completedAbruptly)首先，所有任務(wù)的調(diào)度都是由execute方法完成的，這部分完成的工作是：檢查現(xiàn)在線程池的運(yùn)行狀態(tài)、運(yùn)行線程數(shù)、運(yùn)行策略，決定接下來(lái)執(zhí)行的流程，是直接申請(qǐng)線程執(zhí)行，或是緩沖到隊(duì)列中執(zhí)行，亦或是直接拒絕該任務(wù)。其執(zhí)行過(guò)程如下：
/**
* Executes the given task sometime in the future. The task
* may execute in a new thread or in an existing pooled thread.
* If the task cannot be submitted for execution, either because this
* executor has been shutdown or because its capacity has been reached,
* the task is handled by the current {@code RejectedExecutionHandler}.
* @param command the task to execute
* @throws RejectedExecutionException at discretion of
* {@code RejectedExecutionHandler}, if the task
* cannot be accepted for execution
* @throws NullPointerException if {@code command} is null
*/
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
int c = ctl.get(); // 獲取原子計(jì)數(shù)值最新值
if (workerCountOf(c) < corePoolSize) {
// 判斷當(dāng)前線程池?cái)?shù)量是否小于核心線程數(shù)量
if (addWorker(command, true))
// 嘗試添加command任務(wù)到核心線程
return;
c = ctl.get();
// 重新獲取當(dāng)前線程池狀態(tài)值，為后面的檢查做準(zhǔn)備。
}
// 執(zhí)行到此，說(shuō)明核心線程任務(wù)數(shù)量已滿，新添加的線程入等待隊(duì)列，
這個(gè)是大于corePoolSize且小于maximumPoolSize
if (isRunning(c) && workQueue.offer(command)) {
// 如果線程池處于可接受任務(wù)狀態(tài)，嘗試添加到等待隊(duì)列
int recheck = ctl.get(); // 雙重校驗(yàn)
if (! isRunning(recheck) && remove(command))
// 如果線程池突然不可接受任務(wù)，則嘗試移除該command任務(wù)
reject(command);
// 不可接受任務(wù)且成功從等待隊(duì)列移除任務(wù)，則執(zhí)行拒絕策略操作，
// 通過(guò)策略告訴調(diào)用方任務(wù)入隊(duì)情況
else if (workerCountOf(recheck) == 0)
// 如果此刻線程數(shù)量為0的話將沒(méi)有Worker執(zhí)行新的task，所以增加一個(gè)Worker
addWorker(null, false);
// 添加一個(gè)Worker
}
// 執(zhí)行到此，說(shuō)明添加任務(wù)等待隊(duì)列已滿，所以嘗試添加一個(gè)Worker
else if (!addWorker(command, false))
// 如果添加失敗的話，那么拒絕此線程任務(wù)添加
// 拒絕此線程任務(wù)添加
reject(command);
}首先檢測(cè)線程池運(yùn)行狀態(tài)，如果不是RUNNING，則直接拒絕，線程池要保證在RUNNING的狀態(tài)下執(zhí)行任務(wù)。
如果workerCount < corePoolSize，則創(chuàng)建并啟動(dòng)一個(gè)線程來(lái)執(zhí)行新提交的任務(wù)。
如果workerCount >= corePoolSize，且線程池內(nèi)的阻塞隊(duì)列未滿，則將任務(wù)添加到該阻塞隊(duì)列中。
如果workerCount >= corePoolSize && workerCount < maximumPoolSize，且線程池內(nèi)的阻塞隊(duì)列已滿，則創(chuàng)建并啟動(dòng)一個(gè)線程來(lái)執(zhí)行新提交的任務(wù)。
如果workerCount >= maximumPoolSize，并且線程池內(nèi)的阻塞隊(duì)列已滿, 則根據(jù)拒絕策略來(lái)處理該任務(wù), 默認(rèn)的處理方式是直接拋異常。就用RejectedExecutionHandler來(lái)執(zhí)行拒絕策略；
/**
* Checks if a new worker can be added with respect to current
* pool state and the given bound (either core or maximum). If so,
* the worker count is adjusted accordingly, and, if possible, a
* new worker is created and started, running firstTask as its
* first task. This method returns false if the pool is stopped or
* eligible to shut down. It also returns false if the thread
* factory fails to create a thread when asked. If the thread
* creation fails, either due to the thread factory returning
* null, or due to an exception (typically OutOfMemoryError in
* Thread.start()), we roll back cleanly.
*
* @param firstTask the task the new thread should run first (or
* null if none). Workers are created with an initial first task
* (in method execute()) to bypass queuing when there are fewer
* than corePoolSize threads (in which case we always start one),
* or when the queue is full (in which case we must bypass queue).
* Initially idle threads are usually created via
* prestartCoreThread or to replace other dying workers.
*
* @param core if true use corePoolSize as bound, else
* maximumPoolSize. (A boolean indicator is used here rather than a
* value to ensure reads of fresh values after checking other pool
* state).
* @return true if successful
*/
private boolean addWorker(Runnable firstTask, boolean core) {
// 外層循環(huán)，負(fù)責(zé)判斷線程池狀態(tài)，處理線程池狀態(tài)變量加1操作
retry:
for (;;) {
// 狀態(tài)總體相關(guān)值：運(yùn)行狀態(tài) + 執(zhí)行線程任務(wù)數(shù)量
int c = ctl.get();
// 讀取狀態(tài)值 - 運(yùn)行狀態(tài)
int rs = runStateOf(c);
// Check if queue empty only if necessary.
// 滿足下面兩大條件的，說(shuō)明線程池不能接受任務(wù)了，直接返回false處理
// 主要目的就是想說(shuō)，只有線程池的狀態(tài)為 RUNNING 狀態(tài)時(shí)，線程池才會(huì)接收
// 新的任務(wù)，增加新的Worker工作線程
// 線程池的狀態(tài)已經(jīng)至少已經(jīng)處于不能接收任務(wù)的狀態(tài)了
if (rs >= SHUTDOWN &&
//目的是檢查線 程池是否處于關(guān)閉狀態(tài)
! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()))
return false;
// 內(nèi)層循環(huán)，負(fù)責(zé)worker數(shù)量加1操作
for (;;) {
// 獲取當(dāng)前worker線程數(shù)量
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
// 如果線程池?cái)?shù)量達(dá)到最大上限值CAPACITY
// core為true時(shí)判斷是否大于corePoolSize核心線程數(shù)量
// core為false時(shí)判斷是否大于maximumPoolSize最大設(shè)置的線程數(shù)量
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 調(diào)用CAS原子操作，目的是worker線程數(shù)量加1
if (compareAndIncrementWorkerCount(c)) //
break retry;
c = ctl.get(); // Re-read ctl
// CAS原子操作失敗的話，則再次讀取ctl值
if (runStateOf(c) != rs)
// 如果剛剛讀取的c狀態(tài)不等于先前讀取的rs狀態(tài)，則繼續(xù)外層循環(huán)判斷
continue retry;
// else CAS failed due to workerCount change; retry inner loop
// 之所以會(huì)CAS操作失敗，主要是由于多線程并發(fā)操作，導(dǎo)致workerCount
// 工作線程數(shù)量改變而導(dǎo)致的，因此繼續(xù)內(nèi)層循環(huán)嘗試操作
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
// 創(chuàng)建一個(gè)Worker工作線程對(duì)象，將任務(wù)firstTask，
// 新創(chuàng)建的線程thread都封裝到了Worker對(duì)象里面
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
// 由于對(duì)工作線程集合workers的添加或者刪除，
// 涉及到線程安全問(wèn)題，所以才加上鎖且該鎖為非公平鎖
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
// 獲取鎖成功后，執(zhí)行臨界區(qū)代碼，首先檢查獲取當(dāng)前線程池的狀態(tài)rs
int rs = runStateOf(ctl.get());
// 當(dāng)線程池處于可接收任務(wù)狀態(tài)
// 或者是不可接收任務(wù)狀態(tài)，但是有可能該任務(wù)等待隊(duì)列中的任務(wù)
// 滿足這兩種條件時(shí)，都可以添加新的工作線程
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
workers.add(w);
// 添加新的工作線程到工作線程集合workers，workers是set集合
int s = workers.size();
if (s > largestPoolSize)
// 變量記錄了線程池在整個(gè)生命周期中曾經(jīng)出現(xiàn)的最大線程個(gè)數(shù)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
// 往workers工作線程集合中添加成功后，則立馬調(diào)用線程start方法啟動(dòng)起來(lái)
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
// 如果啟動(dòng)線程失敗的話，還得將剛剛添加成功的線程共集合中移除并且做線 // 程數(shù)量做減1操作
addWorkerFailed(w);
}
return workerStarted;
}該方法是任務(wù)提交的一個(gè)核心方法，主要完成狀態(tài)的檢查，工作線程的創(chuàng)建并添加到線程集合切最后順利的話將創(chuàng)建的線程啟動(dòng)；
addWorker(command, true)：當(dāng)線程數(shù)小于corePoolSize時(shí)，添加一個(gè)需要處理的任務(wù)command進(jìn)線程集合，如果workers數(shù)量超過(guò)corePoolSize時(shí)，則返回false不需要添加工作線程；
addWorker(command, false)：當(dāng)?shù)却?duì)列已滿時(shí)，將新來(lái)的任務(wù)command添加到workers線程集合中去，若線程集合大小超過(guò)maximumPoolSize時(shí)，則返回false不需要添加工作線程；
addWorker(null, false)：放一個(gè)空的任務(wù)進(jìn)線程集合，當(dāng)這個(gè)空任務(wù)的線程執(zhí)行時(shí)，會(huì)從等待任務(wù)隊(duì)列中通過(guò)getTask獲取任務(wù)再執(zhí)行，創(chuàng)建新線程且沒(méi)有任務(wù)分配，當(dāng)執(zhí)行時(shí)才去取任務(wù)；
addWorker(null, true)：創(chuàng)建空任務(wù)的工作線程到workers集合中去，在setCorePoolSize方法調(diào)用時(shí)目的是初始化核心工作線程實(shí)例；
任務(wù)調(diào)度是線程池的主要入口，當(dāng)用戶提交了一個(gè)任務(wù)，接下來(lái)這個(gè)任務(wù)將如何執(zhí)行都是由這個(gè)階段決定的。了解這部分就相當(dāng)于了解了線程池的核心運(yùn)行機(jī)制。
/**
* Main worker run loop. Repeatedly gets tasks from queue and
* executes them, while coping with a number of issues:
*
* 1. We may start out with an initial task, in which case we
* don't need to get the first one. Otherwise, as long as pool is
* running, we get tasks from getTask. If it returns null then the
* worker exits due to changed pool state or configuration
* parameters. Other exits result from exception throws in
* external code, in which case completedAbruptly holds, which
* usually leads processWorkerExit to replace this thread.
*
* 2. Before running any task, the lock is acquired to prevent
* other pool interrupts while the task is executing, and then we
* ensure that unless pool is stopping, this thread does not have
* its interrupt set.
*
* 3. Each task run is preceded by a call to beforeExecute, which
* might throw an exception, in which case we cause thread to die
* (breaking loop with completedAbruptly true) without processing
* the task.
*
* 4. Assuming beforeExecute completes normally, we run the task,
* gathering any of its thrown exceptions to send to afterExecute.
* We separately handle RuntimeException, Error (both of which the
* specs guarantee that we trap) and arbitrary Throwables.
* Because we cannot rethrow Throwables within Runnable.run, we
* wrap them within Errors on the way out (to the thread's
* UncaughtExceptionHandler). Any thrown exception also
* conservatively causes thread to die.
*
* 5. After task.run completes, we call afterExecute, which may
* also throw an exception, which will also cause thread to
* die. According to JLS Sec 14.20, this exception is the one that
* will be in effect even if task.run throws.
*
* The net effect of the exception mechanics is that afterExecute
* and the thread's UncaughtExceptionHandler have as accurate
* information as we can provide about any problems encountered by
* user code.
*
* @param w the worker
*/
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
// allow interrupts 允許中斷
w.unlock();
boolean completedAbruptly = true;
try {
// 不斷從等待隊(duì)列blockingQueue中獲取任務(wù)
// 之前addWorker(null, false)這樣的線程執(zhí)行時(shí)，
// 會(huì)通過(guò)getTask中再次獲取任務(wù)并執(zhí)行
while (task != null || (task = getTask()) != null) {
w.lock();
// 上鎖，并不是防止并發(fā)執(zhí)行任務(wù)，
// 而是為了防止shutdown()被調(diào)用時(shí)不終止正在運(yùn)行的worker線程
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
// task.run()執(zhí)行前，由子類實(shí)現(xiàn)
beforeExecute(wt, task);
Throwable thrown = null;
try {
task.run(); // 執(zhí)行線程Runable的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 {
// task.run()執(zhí)行后，由子類實(shí)現(xiàn)
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}addWorker通過(guò)調(diào)用t.start()啟動(dòng)了線程，線程池的真正核心執(zhí)行任務(wù)的地方就在此runWorker中；
不斷的執(zhí)行我們提交任務(wù)的run方法，可能是剛剛提交的任務(wù)，可能是隊(duì)列中等待的隊(duì)列，原因在于Worker工作線程類繼承了AQS類；
Worker重寫(xiě)了AQS的tryAcquire方法，不管先來(lái)后到，一種非公平的競(jìng)爭(zhēng)機(jī)制，通過(guò)CAS獲取鎖，獲取到了就執(zhí)行代碼塊，沒(méi)獲取到的話則添加到CLH隊(duì)列中通過(guò)利用LockSuporrt的park/unpark阻塞任務(wù)等待；
addWorker通過(guò)調(diào)用t.start()啟動(dòng)了線程，線程池的真正核心執(zhí)行任務(wù)的地方就在此runWorker中；
/**
* Performs cleanup and bookkeeping for a dying worker. Called
* only from worker threads. Unless completedAbruptly is set,
* assumes that workerCount has already been adjusted to account
* for exit. This method removes thread from worker set, and
* possibly terminates the pool or replaces the worker if either
* it exited due to user task exception or if fewer than
* corePoolSize workers are running or queue is non-empty but
* there are no workers.
*
* @param w the worker
* @param completedAbruptly if the worker died due to user exception
*/
private void processWorkerExit(Worker w, boolean completedAbruptly) {
// 如果突然中止，說(shuō)明runWorker中遇到什么異常了，
// 那么正在工作的線程自然就需要減1操作了
if (completedAbruptly)
// If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
// 執(zhí)行到此，說(shuō)明runWorker正常執(zhí)行完了，
// 需要正常退出工作線程，上鎖正常操作移除線程
mainLock.lock();
try {
completedTaskCount += w.completedTasks; // 增加線程池完成任務(wù)數(shù)
workers.remove(w); // 從workers線程集合中移除已經(jīng)工作完的線程
} finally {
mainLock.unlock();
}
// 在對(duì)線程池有負(fù)效益的操作時(shí)，都需要“嘗試終止”線程池，主要是判斷線程池是否滿足終止的狀態(tài)；
// 如果狀態(tài)滿足，但還有線程池還有線程，嘗試對(duì)其發(fā)出中斷響應(yīng)，使其能進(jìn)入退出流程；
// 沒(méi)有線程了，更新?tīng)顟B(tài)為tidying->terminated；
tryTerminate();
int c = ctl.get();
// 如果狀態(tài)是running、shutdown，即tryTerminate()沒(méi)有成功終止線程池，嘗試再添加一個(gè)worker
if (runStateLessThan(c, STOP)) {
// 不是突然完成的，即沒(méi)有task任務(wù)可以獲取而完成的，計(jì)算min，并根據(jù)當(dāng)前worker數(shù)量判斷是否需要addWorker()
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
// 如果min為0，且workQueue不為空，至少保持一個(gè)線程
if (min == 0 && ! workQueue.isEmpty())
min = 1;
// 如果線程數(shù)量大于最少數(shù)量，直接返回，否則下面至少要addWorker一個(gè)
if (workerCountOf(c) >= min)
return; // replacement not needed
}
// 只要worker是completedAbruptly突然終止的，或者線程數(shù)量小于要維護(hù)的數(shù)量，就新添一個(gè)worker線程，即使是shutdown狀態(tài)
addWorker(null, false);
}
}異常中止情況worker數(shù)量減1，正常情況就上鎖從workers中移除；
tryTerminate()：在對(duì)線程池有負(fù)效益的操作時(shí)，都需要“嘗試終止”線程池;
是否需要增加worker線程，如果線程池還沒(méi)有完全終止，仍需要保持一定數(shù)量的線程;
到此，相信大家對(duì)“Java的線程池的工作原理”有了更深的了解，不妨來(lái)實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！
免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。
億速云公眾號(hào)
手機(jī)網(wǎng)站二維碼