Java線程池ThreadPoolExecutor實現原理剖析 #28
原文連接: github.com/aCoder2013/…java
引言
在Java中，使用線程池來異步執行一些耗時任務是很是常見的操做。最初咱們通常都是直接使用new Thread().start的方式，但咱們知道，線程的建立和銷燬都會耗費大量的資源，關於線程能夠參考以前的一片博客Java線程那點事兒, 所以咱們須要重用線程資源。git
固然也有其餘待解決方案，好比說coroutine, 目前Kotlin已經支持了，JDK也已經有了相關的提案:Project Loom, 目前的實現方式和Kotlin有點相似，都是基於ForkJoinPool,固然目前還有不少限制，以及問題沒解決，好比synchronized仍是鎖住當前線程等。github
繼承結構
void execute(Runnable command)方法，而後是ExecutorService，ExecutorService提供了一些管理相關的方法，例如關閉、判斷當前線程池的狀態等，另外不一樣於
Executor#execute，ExecutorService提供了一系列方法，能夠將任務包裝成一個Future，從而使得任務提交方能夠跟蹤任務的狀態。而父類AbstractExecutorService則提供了一些默認的實現。
構造器
ThreadPoolExecutor的構造器提供了很是多的參數，每個參數都很是的重要，一不當心就容易踩坑，所以設置的時候，你必需要知道本身在幹什麼。安全
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.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
複製代碼
-
corePoolSize、 maximumPoolSize。線程池會自動根據corePoolSize和maximumPoolSize去調整當前線程池的大小。當你經過submit或者execute方法提交任務的時候，若是當前線程池的線程數小於corePoolSize,那麼線程池就會建立一個新的線程處理任務, 即便其餘的core線程是空閒的。若是當前線程數大於corePoolSize而且小於maximumPoolSize，那麼只有在隊列"滿"的時候纔會建立新的線程。所以這裏會有不少的坑，好比你的core和max線程數設置的不同，但願請求積壓在隊列的時候可以實時的擴容，但若是制定了一個無界隊列，那麼就不會擴容了，由於隊列不存在滿的概念。多線程
-
keepAliveTime。若是當前線程池中的線程數超過了corePoolSize，那麼若是在keepAliveTime時間內都沒有新的任務須要處理，那麼超過corePoolSize的這部分線程就會被銷燬。默認狀況下是不會回收core線程的，能夠經過設置
allowCoreThreadTimeOut改變這一行爲。併發 -
workQueue。即實際用於存儲任務的隊列，這個能夠說是最核心的一個參數了，直接決定了線程池的行爲，好比說傳入一個有界隊列，那麼隊列滿的時候，線程池就會根據core和max參數的設置狀況決定是否須要擴容，若是傳入了一個SynchronousQueue,這個隊列只有在另外一個線程在同步remove的時候才能夠put成功，對應到線程池中，簡單來講就是若是有線程池任務處理完了，調用poll或者take方法獲取新的任務的時候，新提交的任務纔會put成功，不然若是當前的線程都在忙着處理任務，那麼就會put失敗，也就會走擴容的邏輯，若是傳入了一個DelayedWorkQueue，顧名思義，任務就會根據過時時間來決定何時彈出，即爲ScheduledThreadPoolExecutor的機制。框架
-
threadFactory。建立線程都是經過ThreadFactory來實現的，若是沒指定的話，默認會使用
Executors.defaultThreadFactory()，通常來講，咱們會在這裏對線程設置名稱、異常處理器等。異步 -
handler。即當任務提交失敗的時候，會調用這個處理器，ThreadPoolExecutor內置了多個實現，好比拋異常、直接拋棄等。這裏也須要根據業務場景進行設置，好比說當隊列積壓的時候，針對性的對線程池擴容或者發送告警等策略。函數
看完這幾個參數的含義，咱們看一下Executors提供的一些工具方法，只要是爲了方便使用，可是我建議最好少用這個類，而是直接用ThreadPoolExecutor的構造函數，多瞭解一下這幾個參數究竟是什麼意思，本身的業務場景是什麼樣的，好比線程池需不須要擴容、用不用回收空閒的線程等。工具
public class Executors {
/* * 提供一個固定大小的線程池，而且線程不會回收，因爲傳入的是一個無界隊列，至關於隊列永遠不會滿 * 也就不會擴容，所以須要特別注意任務積壓在隊列中致使內存爆掉的問題 */
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
/* * 這個線程池會一直擴容，因爲SynchronousQueue的特性，若是當前全部的線程都在處理任務，那麼 * 新的請求過來，就會致使建立一個新的線程處理任務。若是線程一分鐘沒有新任務處理，就會被回 * 收掉。特別注意，若是每個任務都比較耗時，併發又比較高，那麼可能每次任務過來都會建立一個線 * 程 */
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
}
複製代碼
源碼分析
既然是個線程池，那就必然有其生命週期:運行中、關閉、中止等。ThreadPoolExecutor是用一個AtomicInteger去的前三位表示這個狀態的，另外又重用了低29位用於表示線程數，能夠支持最大大概5億多，絕逼夠用了，若是之後硬件真的發展到可以啓動這麼多線程，改爲AtomicLong就能夠了。 狀態這裏主要分爲下面幾種:
- RUNNING: 表示當前線程池正在運行中，能夠接受新任務以及處理隊列中的任務
- SHUTDOWN: 再也不接受新的任務，但會繼續處理隊列中的任務
- STOP: 再也不接受新的任務，也不處理隊列中的任務了，而且會中斷正在進行中的任務
- TIDYING: 全部任務都已經處理完畢，線程數爲0，轉爲爲TIDYING狀態以後，會調用
terminated()回調 - TERMINATED:
terminated()已經執行完畢
同時咱們能夠看到全部的狀態都是用二進制位表示的，而且依次遞增，從而方便進行比較，好比想獲取當前狀態是否至少爲SHUTDOWN等，同時狀態以前有幾種轉換:
- RUNNING -> SHUTDOWN。調用了
shutdown()以後，或者執行了finalize() - (RUNNING 或者 SHUTDOWN) -> STOP。調用了
shutdownNow()以後會轉換這個狀態 - SHUTDOWN -> TIDYING。當線程池和隊列都爲空的時候
- STOP -> TIDYING。當線程池爲空的時候
- IDYING -> TERMINATED。執行完
terminated()回調以後會轉換爲這個狀態
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
//因爲前三位表示狀態，所以將CAPACITY取反，和進行與操做便可
private static int runStateOf(int c) { return c & ~CAPACITY; }
private static int workerCountOf(int c) { return c & CAPACITY; }
//高三位+第三位進行或操做便可
private static int ctlOf(int rs, int wc) { return rs | wc; }
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
//下面三個方法，經過CAS修改worker的數目
private boolean compareAndIncrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect + 1);
}
//只嘗試一次，失敗了則返回，是否重試由調用方決定
private boolean compareAndDecrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect - 1);
}
//跟上一個不同，會一直重試
private void decrementWorkerCount() {
do {} while (! compareAndDecrementWorkerCount(ctl.get()));
}
複製代碼
下面是比較核心的字段,這裏workers採用的是非線程安全的HashSet, 而不是線程安全的版本，主要是由於這裏有些複合的操做，好比說將worker添加到workers後，咱們還須要判斷是否須要更新largestPoolSize等，workers只在獲取到mainLock的狀況下才會進行讀寫，另外這裏的mainLock也用於在中斷線程的時候串行執行，不然若是不加鎖的話，可能會形成併發去中斷線程，引發沒必要要的中斷風暴。
private final ReentrantLock mainLock = new ReentrantLock();
private final HashSet<Worker> workers = new HashSet<Worker>();
private final Condition termination = mainLock.newCondition();
private int largestPoolSize;
private long completedTaskCount;
複製代碼
核心方法
拿到一個線程池以後，咱們就能夠開始提交任務，讓它去執行了，那麼咱們看一下submit方法是如何實現的。
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
複製代碼
這兩個方法都很簡單，首先將提交過來的任務(有兩種形式:Callable、Runnable )都包裝成統一的 RunnableFuture，而後調用execute方法，execute能夠說是線程池最核心的一個方法。
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
/* 獲取當前worker的數目，若是小於corePoolSize那麼就擴容， 這裏不會判斷是否已經有core線程，而是隻要小於corePoolSize就會直接增長worker */
if (workerCountOf(c) < corePoolSize) {
/* 調用addWorker(Runnable firstTask, boolean core)方法擴容 firstTask表示爲該worker啓動以後要執行的第一個任務，core表示要增長的爲core線程 */
if (addWorker(command, true))
return;
//若是增長失敗了那麼從新獲取ctl的快照,好比可能線程池在這期間關閉了
c = ctl.get();
}
/* 若是當前線程池正在運行中，而且將任務丟到隊列中成功了， 那麼就會進行一次double check,看下在這期間線程池是否關閉了， 若是關閉了，好比處於SHUTDOWN狀態，如上文所講的，SHUTDOWN狀態的時候， 再也不接受新任務，remove成功後調用拒絕處理器。而若是仍然處於運行中的狀態， 那麼這裏就double check下當前的worker數，若是爲0，有可能在上述邏輯的執行 過程當中，有worker銷燬了，好比說任務拋出了未捕獲異常等，那麼就會進行一次擴容， 但不一樣於擴容core線程，這裏因爲任務已經丟到隊列中去了，所以就不須要再傳遞firstTask了， 同時要注意，這裏擴容的是非core線程 */
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))
/* 若是在上一步中，將任務丟到隊列中失敗了，那麼就進行一次擴容， 這裏會將任務傳遞到firstTask參數中，而且擴容的是非core線程， 若是擴容失敗了，那麼就執行拒絕策略。 */
reject(command);
}
複製代碼
這裏要特別注意下防止隊列失敗的邏輯，不一樣的隊列丟任務的邏輯也不同，例如說無界隊列，那麼就永遠不會put失敗，也就是說擴容也永遠不會執行，若是是有界隊列，那麼當隊列滿的時候，會擴容非core線程，若是是SynchronousQueue,這個隊列比較特殊，當有另一個線程正在同步獲取任務的時候，你才能put成功，所以若是當前線程池中全部的worker都忙着處理任務的時候，那麼後續的每次新任務都會致使擴容, 固然若是worker沒有任務處理了，阻塞在獲取任務這一步的時候，新任務的提交就會直接丟到隊列中去，而不會擴容。 上文中屢次提到了擴容，那麼咱們下面看一下線程池具體是如何進行擴容的:
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
//獲取當前線程池的狀態
int rs = runStateOf(c);
/* 若是狀態爲大於SHUTDOWN, 好比說STOP，STOP上文說過隊列中的任務不處理了，也不接受新任務， 所以能夠直接返回false不擴容了，若是狀態爲SHUTDOWN而且firstTask爲null，同時隊列非空， 那麼就能夠擴容 */
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
/* 若worker的數目大於CAPACITY則直接返回， 而後根據要擴容的是core線程仍是非core線程，進行判斷worker數目 是否超過設置的值，超過則返回 */
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
/* 經過CAS的方式自增worker的數目，成功了則直接跳出循環 */
if (compareAndIncrementWorkerCount(c))
break retry;
//從新讀取狀態變量，若是狀態改變了，好比線程池關閉了，那麼就跳到最外層的for循環，
//注意這裏跳出的是retry。
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//建立Worker
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
/* 獲取鎖，並判斷線程池是否已經關閉 */
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // 若線程已經啓動了，好比說已經調用了start()方法，那麼就拋異常，
throw new IllegalThreadStateException();
//添加到workers中
workers.add(w);
int s = workers.size();
if (s > largestPoolSize) //更新largestPoolSize
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
//若Worker建立成功，則啓動線程，這麼時候worker就會開始執行任務了
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
//添加失敗
addWorkerFailed(w);
}
return workerStarted;
}
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (w != null)
workers.remove(w);
decrementWorkerCount();
//每次減小worker或者從隊列中移除任務的時候都須要調用這個方法
tryTerminate();
} finally {
mainLock.unlock();
}
}
複製代碼
這裏有個貌似不太起眼的方法tryTerminate，這個方法會在全部可能致使線程池終結的地方調用，好比說減小worker的數目等，若是知足條件的話，那麼將線程池轉換爲TERMINATED狀態。另外這個方法沒有用private修飾，由於ScheduledThreadPoolExecutor繼承自ThreadPoolExecutor，而ScheduledThreadPoolExecutor也會調用這個方法。
final void tryTerminate() {
for (;;) {
int c = ctl.get();
/* 若是當前線程處於運行中、TIDYING、TERMINATED狀態則直接返回，運行中的沒 什麼好說的，後面兩種狀態能夠說線程池已經正在終結了，另外若是處於SHUTDOWN狀態， 而且workQueue非空，代表還有任務須要處理，也直接返回 */
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
//能夠退出，可是線程數非0，那麼就中斷一個線程，從而使得關閉的信號可以傳遞下去,
//中斷worker後，worker捕獲異常後，會嘗試退出，並在這裏繼續執行tryTerminate()方法，
//從而使得信號傳遞下去
if (workerCountOf(c) != 0) {
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//嘗試轉換成TIDYING狀態，執行完terminated回調以後
//會轉換爲TERMINATED狀態，這個時候線程池已經完整關閉了，
//經過signalAll方法，喚醒全部阻塞在awaitTermination上的線程
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
terminated();
} finally {
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
/** * 中斷空閒的線程 * @param onlyOne */
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
//遍歷全部worker，若以前沒有被中斷過，
//而且獲取鎖成功，那麼就嘗試中斷。
//鎖可以獲取成功，那麼代表當前worker沒有在執行任務，而是在
//獲取任務，所以也就達到了只中斷空閒線程的目的。
Thread t = w.thread;
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
複製代碼
Worker
下面看一下Worker類，也就是這個類實際負責執行任務，Worker類繼承自AbstractQueuedSynchronizer，AQS能夠理解爲一個同步框架，提供了一些通用的機制，利用模板方法模式，讓你可以原子的管理同步狀態、blocking和unblocking線程、以及隊列，具體的內容以後有時間會再寫，仍是比較複雜的。這裏Worker對AQS的使用相對比較簡單，使用了狀態變量state表示是否得到鎖，0表示解鎖、1表示已得到鎖，同時經過exclusiveOwnerThread存儲當前持有鎖的線程。另外再簡單提一下，好比說CountDownLatch, 也是基於AQS框架實現的，countdown方法遞減state，await阻塞等待state爲0。
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
/** Thread this worker is running in. Null if factory fails. */
final Thread thread;
/** Initial task to run. Possibly null. */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
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(); }
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
複製代碼
注意這裏Worker初始化的時候，會經過setState(-1)將state設置爲-1，並在runWorker()方法中置爲0，上文說過Worker是利用state這個變量來表示鎖的狀態，那麼加鎖的操做就是經過CAS將state從0改爲1，那麼初始化的時候改爲-1，也就是表示在Worker啓動以前，都不容許加鎖操做，咱們再看interruptIfStarted()以及interruptIdleWorkers()方法，這兩個方法在嘗試中斷Worker以前，都會先加鎖或者判斷state是否大於0，所以這裏的將state設置爲-1，就是爲了禁止中斷操做，並在runWorker中置爲0，也就是說只能在Worker啓動以後纔可以中斷Worker。 另外線程啓動以後，其實就是調用了runWorker方法，下面咱們看一下具體是如何實現的。
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // 調用unlock()方法，將state置爲0，表示其餘操做能夠得到鎖或者中斷worker
boolean completedAbruptly = true;
try {
/* 首先嚐試執行firstTask，若沒有的話，則調用getTask()從隊列中獲取任務 */
while (task != null || (task = getTask()) != null) {
w.lock();
/* 若是線程池正在關閉，那麼中斷線程。 */
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
//執行beforeExecute回調
beforeExecute(wt, task);
Throwable thrown = null;
try {
//實際開始執行任務
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回調
afterExecute(task, thrown);
}
} finally {
task = null;
//這裏加了鎖，所以沒有線程安全的問題，volatile修飾保證其餘線程的可見性
w.completedTasks++;
w.unlock();//解鎖
}
}
completedAbruptly = false;
} finally {
//拋異常了，或者當前隊列中已沒有任務須要處理等
processWorkerExit(w, completedAbruptly);
}
}
private void processWorkerExit(Worker w, boolean completedAbruptly) {
//若是是異常終止的，那麼減小worker的數目
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//將當前worker中workers中刪除掉，並累加當前worker已執行的任務到completedTaskCount中
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
//上文說過，減小worker的操做都須要調用這個方法
tryTerminate();
/* 若是當前線程池仍然是運行中的狀態，那麼就看一下是否須要新增另一個worker替換此worker */
int c = ctl.get();
if (runStateLessThan(c, STOP)) {
/* 若是是異常結束的則直接擴容，不然的話則爲正常退出，好比當前隊列中已經沒有任務須要處理， 若是容許core線程超時的話，那麼看一下當前隊列是否爲空，空的話則不用擴容。不然話看一下 是否少於corePoolSize個worker在運行。 */
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
addWorker(null, false);
}
}
private Runnable getTask() {
boolean timedOut = false; // 上一次poll()是否超時了
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 若線程池關閉了(狀態大於STOP)
// 或者線程池處於SHUTDOWN狀態，可是隊列爲空，那麼返回null
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
/* 若是容許core線程超時 或者 不容許core線程超時但當前worker的數目大於core線程數， 那麼下面的poll()則超時調用 */
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
/* 獲取任務超時了而且(當前線程池中還有不止一個worker 或者 隊列中已經沒有任務了)，那麼就嘗試 減小worker的數目，若失敗了則重試 */
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
//從隊列中抓取任務
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
//走到這裏代表，poll調用超時了
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
複製代碼
關閉線程池
關閉線程池通常有兩種形式，shutdown()和shutdownNow()
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
//經過CAS將狀態更改成SHUTDOWN，這個時候線程池不接受新任務，但會繼續處理隊列中的任務
advanceRunState(SHUTDOWN);
//中斷全部空閒的worker,也就是說除了正在處理任務的worker，其餘阻塞在getTask()上的worker
//都會被中斷
interruptIdleWorkers();
//執行回調
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
//這個方法不會等待全部的任務處理完成才返回
}
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
/* 不一樣於shutdown()，會轉換爲STOP狀態，再也不處理新任務，隊列中的任務也不處理， 並且會中斷全部的worker，而不僅是空閒的worker */
advanceRunState(STOP);
interruptWorkers();
tasks = drainQueue();//將全部的任務從隊列中彈出
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
private List<Runnable> drainQueue() {
BlockingQueue<Runnable> q = workQueue;
ArrayList<Runnable> taskList = new ArrayList<Runnable>();
/* 將隊列中全部的任務remove掉，並添加到taskList中， 可是有些隊列比較特殊，好比說DelayQueue，若是第一個任務還沒到過時時間，則不會彈出， 所以這裏經過調用toArray方法，而後再一個一個的remove掉 */
q.drainTo(taskList);
if (!q.isEmpty()) {
for (Runnable r : q.toArray(new Runnable[0])) {
if (q.remove(r))
taskList.add(r);
}
}
return taskList;
}
複製代碼
從上文中能夠看到，調用了shutdown()方法後，不會等待全部的任務處理完畢才返回，所以須要調用awaitTermination()來實現
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (;;) {
//線程池若已經終結了，那麼就返回
if (runStateAtLeast(ctl.get(), TERMINATED))
return true;
//若超時了，也返回掉
if (nanos <= 0)
return false;
//阻塞在信號量上，等待線程池終結,可是要注意這個方法可能會由於一些未知緣由隨時喚醒當前線程，
//所以須要重試，在tryTerminate()方法中，執行完terminated()回調後，代表線程池已經終結了,
//而後會經過termination.signalAll()喚醒當前線程
nanos = termination.awaitNanos(nanos);
}
} finally {
mainLock.unlock();
}
}
複製代碼
一些統計相關的方法
public int getPoolSize() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//若線程已終結則直接返回0，不然計算works中的數目
//想一下爲何不用workerCount呢?
return runStateAtLeast(ctl.get(), TIDYING) ? 0
: workers.size();
} finally {
mainLock.unlock();
}
}
public int getActiveCount() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int n = 0;
for (Worker w : workers)
if (w.isLocked())//上鎖的代表worker當前正在處理任務，也就是活躍的worker
++n;
return n;
} finally {
mainLock.unlock();
}
}
public int getLargestPoolSize() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
return largestPoolSize;
} finally {
mainLock.unlock();
}
}
//獲取任務的總數，這個方法慎用，如果個無解隊列，或者隊列擠壓比較嚴重，會很蛋疼
public long getTaskCount() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
long n = completedTaskCount;//好比有些worker被銷燬後，其處理完成的任務就會疊加到這裏
for (Worker w : workers) {
n += w.completedTasks;//疊加歷史處理完成的任務
if (w.isLocked())//上鎖代表正在處理任務,也算一個
++n;
}
return n + workQueue.size();//獲取隊列中的數目
} finally {
mainLock.unlock();
}
}
public long getCompletedTaskCount() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
long n = completedTaskCount;
for (Worker w : workers)
n += w.completedTasks;
return n;
} finally {
mainLock.unlock();
}
}
複製代碼
總結
這篇博客基本上覆蓋了線程池的方方面面，但仍然有很是多的細節能夠深究，好比說異常的處理，能夠參照以前的一篇博客:深度解析Java線程池的異常處理機制 ,另外還有AQS、unsafe等能夠以後再單獨總結。
相關文章
- 1. Java----線程池ThreadPoolExecutor實現原理剖析
- 2. ThreadPoolExecutor 原理-- java 線程池
- 3. Java線程池ThreadPoolExecutor原理解析
- 4. 線程池ThreadPoolExecutor實現原理
- 5. Java線程池核心原理剖析
- 6. JAVA線程池原理剖析
- 7. ThreadPoolExecutor線程池原理
- 8. Java中的線程池——ThreadPoolExecutor的原理
- 9. Java中線程池ThreadPoolExecutor原理探究
- 10. Java線程池-ThreadPoolExecutor原理分析與實戰
- 更多相關文章...
- • C# 多線程 - C#教程
- • MyBatis的工作原理 - MyBatis教程
- • Java Agent入門實戰（三）-JVM Attach原理與使用
- • ☆基於Java Instrument的Agent實現
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. eclipse設置粘貼字符串自動轉義
- 2. android客戶端學習-啓動模擬器異常Emulator: failed to initialize HAX: Invalid argument
- 3. android.view.InflateException: class com.jpardogo.listbuddies.lib.views.ListBuddiesLayout問題
- 4. MYSQL8.0數據庫恢復 MYSQL8.0ibd數據恢復 MYSQL8.0恢復數據庫
- 5. 你本是一個肉體，是什麼驅使你前行【1】
- 6. 2018.04.30
- 7. 2018.04.30
- 8. 你本是一個肉體，是什麼驅使你前行【3】
- 9. 你本是一個肉體，是什麼驅使你前行【2】
- 10. 【資訊】LocalBitcoins達到每週交易比特幣的7年低點
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. Java----線程池ThreadPoolExecutor實現原理剖析
- 2. ThreadPoolExecutor 原理-- java 線程池
- 3. Java線程池ThreadPoolExecutor原理解析
- 4. 線程池ThreadPoolExecutor實現原理
- 5. Java線程池核心原理剖析
- 6. JAVA線程池原理剖析
- 7. ThreadPoolExecutor線程池原理
- 8. Java中的線程池——ThreadPoolExecutor的原理
- 9. Java中線程池ThreadPoolExecutor原理探究
- 10. Java線程池-ThreadPoolExecutor原理分析與實戰