總覽
下圖是 java 線程池幾個(gè)相關(guān)類的繼承結(jié)構(gòu)：
先簡單說說這個(gè)繼承結(jié)構(gòu)，Executor 位于最頂層，也是最簡單的，就一個(gè) execute(Runnable runnable) 接口方法定義。
ExecutorService 也是接口，在 Executor 接口的基礎(chǔ)上添加了很多的接口方法，所以一般來說我們會使用這個(gè)接口。
然后再下來一層是 AbstractExecutorService，從名字我們就知道，這是抽象類，這里實(shí)現(xiàn)了非常有用的一些方法供子類直接使用，之后我們再細(xì)說。
然后才到我們的重點(diǎn)部分 ThreadPoolExecutor 類，這個(gè)類提供了關(guān)于線程池所需的非常豐富的功能。
同在并發(fā)包中的 Executors 類，類名中帶字母 s，我們猜到這個(gè)是工具類，里面的方法都是靜態(tài)方法，如以下我們最常用的用于生成 ThreadPoolExecutor 的實(shí)例的一些方法,也就是四大線程池：
- newCachedThreadPool的cpu核心線程數(shù)為 0，最大線程數(shù)為 Integer.MAX_VALUE，keepAliveTime 為 60 秒，任務(wù)隊(duì)列采用 SynchronousQueue
這種線程池對于任務(wù)可以比較快速地完成的情況有比較好的性能。如果線程空閑了 60 秒都沒有任務(wù)，那么將關(guān)閉此線程并從線程池中移除。所以如果線程池空閑了很長時(shí)間也不會有問題，因?yàn)殡S著所有的線程都會被關(guān)閉，整個(gè)線程池不會占用任何的系統(tǒng)資源。
- newFixedThreadPool最大線程數(shù)設(shè)置為與核心線程數(shù)相等，此時(shí) keepAliveTime 設(shè)置為 0（因?yàn)檫@里它是沒用的，即使不為 0，線程池默認(rèn)也不會回收 corePoolSize 內(nèi)的線程），任務(wù)隊(duì)列采用 LinkedBlockingQueue，無界隊(duì)列。
- newSingleThreadExecutor生成只有一個(gè)線程的固定線程池，這個(gè)更簡單，和上面的一樣，只要設(shè)置線程數(shù)為 1 就可以了
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
SynchronousQueue 是一個(gè)比較特殊的 BlockingQueue，其本身不儲存任何元素，
它有一個(gè)虛擬隊(duì)列（或虛擬棧），不管讀操作還是寫操作，
如果當(dāng)前隊(duì)列中存儲的是與當(dāng)前操作相同模式的線程，那么當(dāng)前操作也進(jìn)入隊(duì)列中等待；
如果是相反模式，則配對成功，從當(dāng)前隊(duì)列中取隊(duì)頭節(jié)點(diǎn)。具體的信息，
可以看我的另一篇關(guān)于 BlockingQueue 的文章。
另外，由于線程池支持獲取線程執(zhí)行的結(jié)果，所以，引入了 Future 接口，RunnableFuture 繼承自此接口，然后我們最需要關(guān)心的就是它的實(shí)現(xiàn)類 FutureTask。到這里，記住這個(gè)概念，在線程池的使用過程中，我們是往線程池提交任務(wù)（task），使用過線程池的都知道，我們提交的每個(gè)任務(wù)是實(shí)現(xiàn)了 Runnable 接口的，其實(shí)就是先將 Runnable 的任務(wù)包裝成 FutureTask，然后再提交到線程池。這樣，讀者才能比較容易記住 FutureTask 這個(gè)類名：它首先是一個(gè)任務(wù)（Task），然后具有 Future 接口的語義，即可以在將來（Future）得到執(zhí)行的結(jié)果。
當(dāng)然，線程池中的 BlockingQueue 也是非常重要的概念，如果線程數(shù)達(dá)到 corePoolSize，我們的每個(gè)任務(wù)會提交到等待隊(duì)列中，等待線程池中的線程來取任務(wù)并執(zhí)行。這里的 BlockingQueue 通常我們使用其實(shí)現(xiàn)類 LinkedBlockingQueue、ArrayBlockingQueue 和 SynchronousQueue，每個(gè)實(shí)現(xiàn)類都有不同的特征，使用場景之后會慢慢分析。想要詳細(xì)了解各個(gè) BlockingQueue 的讀者，可以參考我的前面的一篇對 BlockingQueue 的各個(gè)實(shí)現(xiàn)類進(jìn)行詳細(xì)分析的文章。
Executor 接口
/*
* @since 1.5
* @author Doug Lea
*/
public interface Executor {
void execute(Runnable command);
}
我們可以看到 Executor 接口非常簡單，就一個(gè) void execute(Runnable command) 方法，代表提交一個(gè)任務(wù)。
比如我們想知道執(zhí)行結(jié)果、我們想知道當(dāng)前線程池有多少個(gè)線程活著、已經(jīng)完成了多少任務(wù)等等，這些都是這個(gè)接口的不足的地方。接下來我們要介紹的是繼承自 Executor 接口的 ExecutorService 接口，這個(gè)接口提供了比較豐富的功能，也是我們最常使用到的接口。
ExecutorService
一般我們定義一個(gè)線程池的時(shí)候，往往都是使用這個(gè)接口：
ExecutorService executor = Executors.newFixedThreadPool(args...);
ExecutorService executor = Executors.newCachedThreadPool(args...);
因?yàn)檫@個(gè)接口中定義的一系列方法大部分情況下已經(jīng)可以滿足我們的需要了。
那么我們簡單初略地來看一下這個(gè)接口中都有哪些方法：
public interface ExecutorService extends Executor {
// 關(guān)閉線程池，已提交的任務(wù)繼續(xù)執(zhí)行，不接受繼續(xù)提交新任務(wù)
void shutdown();
// 關(guān)閉線程池，嘗試停止正在執(zhí)行的所有任務(wù)，不接受繼續(xù)提交新任務(wù)
// 它和前面的方法相比，加了一個(gè)單詞“now”，區(qū)別在于它會去停止當(dāng)前正在進(jìn)行的任務(wù)
List<Runnable> shutdownNow();
// 線程池是否已關(guān)閉
boolean isShutdown();
// 如果調(diào)用了 shutdown() 或 shutdownNow() 方法后，所有任務(wù)結(jié)束了，那么返回true
// 這個(gè)方法必須在調(diào)用shutdown或shutdownNow方法之后調(diào)用才會返回true
boolean isTerminated();
// 等待所有任務(wù)完成，并設(shè)置超時(shí)時(shí)間
// 我們這么理解，實(shí)際應(yīng)用中是，先調(diào)用 shutdown 或 shutdownNow，
// 然后再調(diào)這個(gè)方法等待所有的線程真正地完成，返回值意味著有沒有超時(shí)
boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException;
// 提交一個(gè) Callable 任務(wù)
<T> Future<T> submit(Callable<T> task);
// 提交一個(gè) Runnable 任務(wù)，第二個(gè)參數(shù)將會放到 Future 中，作為返回值，
// 因?yàn)?Runnable 的 run 方法本身并不返回任何東西
<T> Future<T> submit(Runnable task, T result);
// 提交一個(gè) Runnable 任務(wù)
Future<?> submit(Runnable task);
// 執(zhí)行所有任務(wù)，返回 Future 類型的一個(gè) list
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException;
// 也是執(zhí)行所有任務(wù)，但是這里設(shè)置了超時(shí)時(shí)間
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException;
// 只有其中的一個(gè)任務(wù)結(jié)束了，就可以返回，返回執(zhí)行完的那個(gè)任務(wù)的結(jié)果
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException;
// 同上一個(gè)方法，只有其中的一個(gè)任務(wù)結(jié)束了，就可以返回，返回執(zhí)行完的那個(gè)任務(wù)的結(jié)果，
// 不過這個(gè)帶超時(shí)，超過指定的時(shí)間，拋出 TimeoutException 異常
<T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
這些方法都很好理解，一個(gè)簡單的線程池主要就是這些功能，能提交任務(wù)，能獲取結(jié)果，能關(guān)閉線程池，這也是為什么我們經(jīng)常用這個(gè)接口的原因。
FutureTask
在繼續(xù)往下層介紹 ExecutorService 的實(shí)現(xiàn)類之前，我們先來說說相關(guān)的類 FutureTask。
Future Runnable
\ /
\ /
RunnableFuture
|
|
FutureTask
FutureTask 通過 RunnableFuture 間接實(shí)現(xiàn)了 Runnable 接口，
所以每個(gè) Runnable 通常都先包裝成 FutureTask，
然后調(diào)用 executor.execute(Runnable command) 將其提交給線程池
我們知道，Runnable 的 void run() 方法是沒有返回值的，所以，通常，如果我們需要的話，會在 submit 中指定第二個(gè)參數(shù)作為返回值：
<T> Future<T> submit(Runnable task, T result);
其實(shí)到時(shí)候會通過這兩個(gè)參數(shù)，將其包裝成 Callable。它和 Runnable 的區(qū)別在于 run() 沒有返回值，而 Callable 的 call() 方法有返回值，同時(shí)，如果運(yùn)行出現(xiàn)異常，call() 方法會拋出異常。
public interface Callable<V> {
V call() throws Exception;
}
在這里，就不展開說 FutureTask 類了，因?yàn)楸疚钠緛砭蛪虼罅?，這里我們需要知道怎么用就行了。
下面，我們來看看 ExecutorService 的抽象實(shí)現(xiàn) AbstractExecutorService 。
AbstractExecutorService
AbstractExecutorService 抽象類派生自 ExecutorService 接口，然后在其基礎(chǔ)上實(shí)現(xiàn)了幾個(gè)實(shí)用的方法，這些方法提供給子類進(jìn)行調(diào)用。
這個(gè)抽象類實(shí)現(xiàn)了 invokeAny 方法和 invokeAll 方法，這里的兩個(gè) newTaskFor 方法也比較有用，用于將任務(wù)包裝成 FutureTask。定義于最上層接口 Executor中的 void execute(Runnable command) 由于不需要獲取結(jié)果，不會進(jìn)行 FutureTask 的包裝。
需要獲取結(jié)果（FutureTask），用 submit 方法，不需要獲取結(jié)果，可以用 execute 方法。
下面，我將一行一行源碼地來分析這個(gè)類，跟著源碼來看看其實(shí)現(xiàn)吧：
Tips: invokeAny 和 invokeAll 方法占了這整個(gè)類的絕大多數(shù)篇幅，這里選擇適當(dāng)跳過，
因?yàn)樗鼈兛赡茉谀愕膶?shí)踐中使用的頻次比較低，而且它們不帶有承前啟后的作用，不用擔(dān)心會漏掉什么導(dǎo)致看不懂后面的代碼。
public abstract class AbstractExecutorService implements ExecutorService {
// RunnableFuture 是用于獲取執(zhí)行結(jié)果的，我們常用它的子類 FutureTask
// 下面兩個(gè) newTaskFor 方法用于將我們的任務(wù)包裝成 FutureTask 提交到線程池中執(zhí)行
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
// 提交任務(wù)
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
// 1. 將任務(wù)包裝成 FutureTask
RunnableFuture<Void> ftask = newTaskFor(task, null);
// 2. 交給執(zhí)行器執(zhí)行，execute 方法由具體的子類來實(shí)現(xiàn)
// 前面也說了，F(xiàn)utureTask 間接實(shí)現(xiàn)了Runnable 接口。
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
// 1. 將任務(wù)包裝成 FutureTask
RunnableFuture<T> ftask = newTaskFor(task, result);
// 2. 交給執(zhí)行器執(zhí)行
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
// 1. 將任務(wù)包裝成 FutureTask
RunnableFuture<T> ftask = newTaskFor(task);
// 2. 交給執(zhí)行器執(zhí)行
execute(ftask);
return ftask;
}
}
到這里，我們發(fā)現(xiàn)，這個(gè)抽象類包裝了一些基本的方法，它們都沒有真正開啟線程來執(zhí)行任務(wù)，它們都只是在方法內(nèi)部調(diào)用了 execute 方法，所以最重要的 execute(Runnable runnable) 方法還沒出現(xiàn)，需要等具體執(zhí)行器來實(shí)現(xiàn)這個(gè)最重要的部分，這里我們要說的就是 ThreadPoolExecutor 類了。
ThreadPoolExecutor
ThreadPoolExecutor 是 JDK 中的線程池實(shí)現(xiàn)，這個(gè)類實(shí)現(xiàn)了一個(gè)線程池需要的各個(gè)方法，它實(shí)現(xiàn)了任務(wù)提交、線程管理、監(jiān)控等等方法。
我們可以基于它來進(jìn)行業(yè)務(wù)上的擴(kuò)展，以實(shí)現(xiàn)我們需要的其他功能，比如實(shí)現(xiàn)定時(shí)任務(wù)的類 ScheduledThreadPoolExecutor 就繼承自 ThreadPoolExecutor。當(dāng)然，這不是本文關(guān)注的重點(diǎn)，下面，還是趕緊進(jìn)行源碼分析吧。
首先看看類的結(jié)構(gòu)：
public class ThreadPoolExecutor extends AbstractExecutorService {
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//Integer.SIZE=32，32-3意味著前三位用于存放線程狀態(tài)，后29位用于存放線程數(shù)
//很多初學(xué)者很喜歡在自己的代碼中寫很多 29 這種數(shù)字，或者某個(gè)特殊的字符串，然后分布在各個(gè)地方，這是非常糟糕的
private static final int COUNT_BITS = Integer.SIZE - 3;
//1 << COUNT_BITS = 0010 0000 0000 0000 0000 0000 0000 0000
//(1 << COUNT_BITS) - 1 = 0001 1111 1111 1111 1111 1111 1111 1111
//得到29個(gè)1，也就是說線程池的最大線程數(shù)是 2^29-1=536870911
// 以我們現(xiàn)在計(jì)算機(jī)的實(shí)際情況，這個(gè)數(shù)量還是夠用的
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// 1=0000 0000 0000 0000 0000 0000 0000 0001
//1反碼=1111 1111 1111 1111 1111 1111 1111 1110
//-1=1補(bǔ)碼=反碼+1=1111 1111 1111 1111 1111 1111 1111 1111
//-1 << COUNT_BITS= -1左移29次=1110 0000 0000 0000 0000 0000 0000 0000
//也就是32的高3位為線程池狀態(tài)
private static final int RUNNING = -1 << COUNT_BITS;
// 0000 0000 0000 0000 0000 0000 0000 0000
private static final int SHUTDOWN = 0 << COUNT_BITS;
//0010 0000 0000 0000 0000 0000 0000 0000
private static final int STOP = 1 << COUNT_BITS;
//0100 0000 0000 0000 0000 0000 0000 0000
private static final int TIDYING = 2 << COUNT_BITS;
//0110 0000 0000 0000 0000 0000 0000 0000
private static final int TERMINATED = 3 << COUNT_BITS;
// 將整數(shù)c的低29位改成0，得到線程池的狀態(tài)
private static int runStateOf(int c) {
return c & ~CAPACITY;
}
//將整數(shù)c的高3位改為0，得到線程池中的線程數(shù)
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;
}
//c < SHUTDOWN也只有RUNNING，這個(gè)方法是判斷線程池是否運(yùn)行中
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
//cas比較設(shè)置增加ctl值，ctl值錢3位為狀態(tài)，后29位為線程數(shù)
private boolean compareAndIncrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect + 1);
}
//cas比較設(shè)置減少ctl值，ctl值錢3位為狀態(tài)，后29位為線程數(shù)
private boolean compareAndDecrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect - 1);
}
//循環(huán)設(shè)置ctl減少，直到更新成功
private void decrementWorkerCount() {
do {
} while (!compareAndDecrementWorkerCount(ctl.get()));
}
//等待隊(duì)列
private final BlockingQueue<Runnable> workQueue;
//全局鎖
private final ReentrantLock mainLock = new ReentrantLock();
//工作線程所在集合
private final HashSet<Worker> workers = new HashSet<Worker>();
//這個(gè)是為了實(shí)現(xiàn)wait..notify類似效果，由condition.await和signal實(shí)現(xiàn)，這個(gè)可以針對單個(gè)喚醒
private final Condition termination = mainLock.newCondition();
//記錄工作worker最大值，就是workers的set集合最大個(gè)數(shù)
private int largestPoolSize;
//執(zhí)行完成的任務(wù)數(shù)量，拋異常也算
private long completedTaskCount;
//創(chuàng)建線程的工程
private volatile ThreadFactory threadFactory;
//在執(zhí)行飽和或關(guān)閉時(shí)調(diào)用處理程序。拒絕策略
private volatile RejectedExecutionHandler handler;
//當(dāng)線程數(shù)大于內(nèi)核數(shù)時(shí)，這是多余的空閑線程將在終止之前等待新任務(wù)的最長時(shí)間
//當(dāng)線程空閑時(shí)間達(dá)到keepAliveTime時(shí)，線程會退出，直到線程數(shù)量=corePoolSize
//如果allowCoreThreadTimeout=true，則會直到線程數(shù)量=0
private volatile long keepAliveTime;
//默認(rèn)false，cpu核心線程閑置的話也會一直保持活著狀態(tài)，如果true,核心線程會以keepAliveTime等待獲取任務(wù)
//允許核心線程超時(shí)
private volatile boolean allowCoreThreadTimeOut;
//cpu核數(shù)
//核心線程會一直存活，即使沒有任務(wù)需要執(zhí)行
//當(dāng)線程數(shù)小于核心線程數(shù)時(shí)，即使有線程空閑，線程池也會優(yōu)先創(chuàng)建新線程處理
// 設(shè)置allowCoreThreadTimeout=true（默認(rèn)false）時(shí)，核心線程會超時(shí)關(guān)閉
private volatile int corePoolSize;
//線程池最大線程數(shù)，實(shí)際受CAPACITY限制
private volatile int maximumPoolSize;
//默認(rèn)拒絕策略，當(dāng)隊(duì)列滿了，線程數(shù)據(jù)也達(dá)到了最大，則執(zhí)行拒絕策略，默認(rèn)直接拋出異常
private static final RejectedExecutionHandler defaultHandler =
new AbortPolicy();
private static final RuntimePermission shutdownPerm =
new RuntimePermission("modifyThread");
/* The context to be used when executing the finalizer, or null. */
private final AccessControlContext acc;
//worker是工作者，里面維護(hù)線程讓線程執(zhí)行任務(wù)
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable {
private static final long serialVersionUID = 6138294804551838833L;
//運(yùn)行任務(wù)的線程
final Thread thread;
//初始化任務(wù)，可以是空
Runnable firstTask;
//完成的任務(wù)數(shù)量
volatile long completedTasks;
//初始化worker,初始任務(wù)，可以為空
Worker(Runnable firstTask) {
//設(shè)置狀態(tài)
//把狀態(tài)位設(shè)置成-1，這樣任何線程都不能得到Worker的鎖，除非調(diào)用了unlock方法。這個(gè)unlock方法會在runWorker方法中一開始就調(diào)用，
//這是為了確保Worker構(gòu)造出來之后，沒有任何線程能夠得到它的鎖，除非調(diào)用了runWorker之后，其他線程才能獲得Worker的鎖
setState(-1);
this.firstTask = firstTask;
//線程工廠創(chuàng)建線程
this.thread = getThreadFactory().newThread(this);
}
//worker執(zhí)行任務(wù)，在addWorker里新增成功后啟動線程
public void run() {
runWorker(this);
}
}
接著，我們來看看線程池實(shí)現(xiàn)中的幾個(gè)概念和處理流程。
我們先回顧下提交任務(wù)的幾個(gè)方法：
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(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
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;
}
當(dāng)然，上圖沒有考慮隊(duì)列是否有界，提交任務(wù)時(shí)隊(duì)列滿了怎么辦？什么情況下會創(chuàng)建新的線程？提交任務(wù)時(shí)線程池滿了怎么辦？空閑線程怎么關(guān)掉？這些問題下面我們會一一解決。
我們經(jīng)常會使用 Executors 這個(gè)工具類來快速構(gòu)造一個(gè)線程池，對于初學(xué)者而言，這種工具類是很有用的，開發(fā)者不需要關(guān)注太多的細(xì)節(jié)，只要知道自己需要一個(gè)線程池，僅僅提供必需的參數(shù)就可以了，其他參數(shù)都采用作者提供的默認(rèn)值。
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
這里先不說有什么區(qū)別，它們最終都會導(dǎo)向這個(gè)構(gòu)造方法：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
//corePoolSize：cpu核數(shù)不能小于0
//maximumPoolSize：線程池做大線程數(shù)不能小于等于0并不能小于cpu核心數(shù)
//keepAliveTime：當(dāng)線程數(shù)大于內(nèi)核數(shù)時(shí)，這是多余的空閑線程將在終止之前等待新任務(wù)的最長時(shí)間,也不能小于0
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
//工作線程所在的隊(duì)列和線程制造工廠和拒絕策略都不能空
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;
}
在這里，介紹下線程池中的各個(gè)狀態(tài)和狀態(tài)變化的轉(zhuǎn)換過程：
- RUNNING：這個(gè)沒什么好說的，這是最正常的狀態(tài)：接受新的任務(wù)，處理等待隊(duì)列中的任務(wù)
- SHUTDOWN：不接受新的任務(wù)提交，但是會繼續(xù)處理等待隊(duì)列中的任務(wù)
- STOP：不接受新的任務(wù)提交，不再處理等待隊(duì)列中的任務(wù)，中斷正在執(zhí)行任務(wù)的線程
- TIDYING：所有的任務(wù)都銷毀了，workCount 為 0。線程池的狀態(tài)在轉(zhuǎn)換為 TIDYING 狀態(tài)時(shí)，會執(zhí)行鉤子方法 terminated()
- TERMINATED：terminated() 方法結(jié)束后，線程池的狀態(tài)就會變成這個(gè)
RUNNING 定義為 -1，SHUTDOWN 定義為 0，其他的都比 0 大，所以等于 0 的時(shí)候不能提交任務(wù)，大于 0 的話，連正在執(zhí)行的任務(wù)也需要中斷。
看了這幾種狀態(tài)的介紹，讀者大體也可以猜到十之八九的狀態(tài)轉(zhuǎn)換了，各個(gè)狀態(tài)的轉(zhuǎn)換過程有以下幾種：
- RUNNING -> SHUTDOWN：當(dāng)調(diào)用了 shutdown() 后，會發(fā)生這個(gè)狀態(tài)轉(zhuǎn)換，這也是最重要的
- (RUNNING or SHUTDOWN) -> STOP：當(dāng)調(diào)用 shutdownNow() 后，會發(fā)生這個(gè)狀態(tài)轉(zhuǎn)換，這下要清楚shutDown() 和 shutDownNow() 的區(qū)別了
- SHUTDOWN -> TIDYING：當(dāng)任務(wù)隊(duì)列和線程池都清空后，會由 SHUTDOWN 轉(zhuǎn)換為 TIDYING
- STOP -> TIDYING：當(dāng)任務(wù)隊(duì)列清空后，發(fā)生這個(gè)轉(zhuǎn)換
- TIDYING -> TERMINATED：這個(gè)前面說了，當(dāng) terminated() 方法結(jié)束后
另外，我們還要看看一個(gè)內(nèi)部類 Worker，因?yàn)?Doug Lea 把線程池中的線程包裝成了一個(gè)個(gè) Worker，翻譯成工人，就是線程池中做任務(wù)的線程。所以到這里，我們知道任務(wù)是 Runnable（內(nèi)部變量名叫 task 或 command），線程是 Worker。
Worker 這里又用到了抽象類 AbstractQueuedSynchronizer。題外話，AQS 在并發(fā)中真的是到處出現(xiàn)，而且非常容易使用，寫少量的代碼就能實(shí)現(xiàn)自己需要的同步方式（對 AQS 源碼感興趣的讀者請參看我之前寫的幾篇文章http://www.itdecent.cn/p/54d372425e54）。
//worker是工作者，里面維護(hù)線程讓線程執(zhí)行任務(wù)
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable {
private static final long serialVersionUID = 6138294804551838833L;
//運(yùn)行任務(wù)的線程
final Thread thread;
//初始化任務(wù)，可以是空
Runnable firstTask;
//完成的任務(wù)數(shù)量
volatile long completedTasks;
//初始化worker,初始任務(wù)，可以為空
Worker(Runnable firstTask) {
//設(shè)置狀態(tài)
//把狀態(tài)位設(shè)置成-1，這樣任何線程都不能得到Worker的鎖，除非調(diào)用了unlock方法。這個(gè)unlock方法會在runWorker方法中一開始就調(diào)用，
//這是為了確保Worker構(gòu)造出來之后，沒有任何線程能夠得到它的鎖，除非調(diào)用了runWorker之后，其他線程才能獲得Worker的鎖
setState(-1);
this.firstTask = firstTask;
//線程工廠創(chuàng)建線程
this.thread = getThreadFactory().newThread(this);
}
//worker執(zhí)行任務(wù)，在addWorker里新增成功后啟動線程
public void run() {
runWorker(this);
}
...// 其他幾個(gè)方法沒什么好看的，就是用 AQS 操作，來獲取這個(gè)線程的執(zhí)行權(quán)，用了獨(dú)占鎖
worker的加鎖解鎖機(jī)制是基于AQS框架的，要完全弄明白它的加鎖解鎖機(jī)制請看AQS框架的實(shí)現(xiàn)，在這里只是簡單介紹一下：
//嘗試加鎖方法，將狀態(tài)從0設(shè)置為1；如果不是0則加鎖失敗,在worker線程沒有啟動前是-1狀態(tài)，無法加鎖
//該方法重寫了父類AQS的同名方法
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
//嘗試釋放鎖的方法，直接將state置為0
//該方法重寫了父類AQS的同名方法
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
//注意：tryAcquire與tryRelease是重寫了AQS父類的方法，且不可以直接調(diào)用，它們被以下方法調(diào)用實(shí)現(xiàn)加鎖解鎖操作
//加鎖:acquire法是它父類AQS類的方法,會調(diào)用tryAcquire方法加鎖
public void lock() { acquire(1); }
//嘗試加鎖
public boolean tryLock() { return tryAcquire(1); }
//解鎖:release方法是它父類AQS類的方法，會調(diào)用tryRelease方法
public void unlock() { release(1); }
//返回鎖狀態(tài)
public boolean isLocked() { return isHeldExclusively(); }
默認(rèn)工廠DefaultThreadFactory
線程的創(chuàng)建看下defaultThreadFactory，當(dāng)然是在Executors類創(chuàng)建
public static ThreadFactory defaultThreadFactory() {
return new DefaultThreadFactory();
}
static class DefaultThreadFactory implements ThreadFactory {
//線程池編號
private static final AtomicInteger poolNumber = new AtomicInteger(1);
//線程池中線程所屬線程組
private final ThreadGroup group;
//線程池中線程編號
private final AtomicInteger threadNumber = new AtomicInteger(1);
//線程名稱前綴
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
//設(shè)置線程名稱為"pool-線程池的編號-thread-線程的編號"
namePrefix = "pool-" +
poolNumber.getAndIncrement() +
"-thread-";
}
//創(chuàng)建新的線程
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
//設(shè)置為非守護(hù)線程
if (t.isDaemon())
t.setDaemon(false);
//設(shè)置優(yōu)先級為NORMAL為5
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}
一般我們最好不要用默認(rèn)的線程池，可以繼承該類，給線程指定一個(gè)識別度高的名字，出了問題好排查；
怎么去創(chuàng)建一個(gè)worker
worker是線程池執(zhí)行任務(wù)的線程
- 如果線程池處于TERMINATED、STOP、TIDYING和(SHUTDOWN+任務(wù)隊(duì)列為空)是沒必要創(chuàng)建線程了
- core是true則線程數(shù)大于等于cpu核心線程數(shù)沒必要創(chuàng)建線程或者core是false則大于等于線程最大數(shù)也沒必要創(chuàng)建線程
- cas比較線程數(shù)+1成功則創(chuàng)建worker并且初始化線程
- 線程添加到線程集合并記錄線程集合最大值，啟動線程
- 線程啟動失敗則從線程集合移除線程
- 如果當(dāng)前線程池狀態(tài)處于RUNNING和(SHUTDOWN+任務(wù)隊(duì)列不為空）則不能去停止線程池，如果處于TIDYING或TERMINATED就沒必要再去關(guān)閉線程池了
- 如果線程池線程不為0，則遍歷線程集合，判斷是否被中斷過是否是閑置，如果是發(fā)出中斷，喚醒獲取任務(wù)阻塞的線程，如果為0，設(shè)置TIDYING，成功則設(shè)置成TERMINATED，關(guān)閉線程池
先看看流程圖：
這個(gè)方法非常重要 addWorker(Runnable firstTask, boolean core) 方法，我們看看它是怎么創(chuàng)建新的線程的：
private boolean addWorker(Runnable firstTask, boolean core) {
//標(biāo)志，和break retry以及continue retry聯(lián)合使用
//break retry結(jié)束內(nèi)存循環(huán)直接往retry:下方代碼，不再執(zhí)行for
//continue retry結(jié)束內(nèi)存循環(huán)直接往retry:下方代碼，重新執(zhí)行for
retry:
//自旋
for (; ; ) {
//或許線程池狀態(tài)整數(shù)
int c = ctl.get();
//得到線程池狀態(tài)
// 線程池狀態(tài)只有高3位
int rs = runStateOf(c);
// 這個(gè)非常不好理解
// 如果線程池已關(guān)閉，并滿足以下條件之一，那么不創(chuàng)建新的 worker：
// 1. 線程池狀態(tài)大于 SHUTDOWN，其實(shí)也就是 STOP, TIDYING, 或 TERMINATED
// 簡單分析下：
// 還是狀態(tài)控制的問題，當(dāng)線程池處于 SHUTDOWN 的時(shí)候，不允許提交任務(wù)，但是已有的任務(wù)繼續(xù)執(zhí)行
// 當(dāng)狀態(tài)大于 SHUTDOWN 時(shí)，不允許提交任務(wù)，且中斷正在執(zhí)行的任務(wù)
// 多說一句：如果線程池處于 SHUTDOWN，但是 firstTask 為 null，且 workQueue 非空，那么是允許創(chuàng)建 worker 的
// 這是因?yàn)?SHUTDOWN 的語義：不允許提交新的任務(wù)，但是要把已經(jīng)進(jìn)入到 workQueue 的任務(wù)執(zhí)行完，所以在滿足條件的基礎(chǔ)上，是允許創(chuàng)建新的 Worker 的
if (rs >= SHUTDOWN &&
!(rs == SHUTDOWN &&
firstTask == null &&
!workQueue.isEmpty()))
return false;
for (; ; ) {
//獲得線程池的線程數(shù)
int wc = workerCountOf(c);
//線程數(shù)大于等于CAPACITY(2^29-1)，創(chuàng)建worker失敗返回
//當(dāng)core是true則線程數(shù)大于等于核心數(shù)就創(chuàng)建worker失敗返回，
// 當(dāng)是false則線程數(shù)大于等于線程最大數(shù)量就創(chuàng)建worker失敗返回
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//cas比較設(shè)置當(dāng)前線程數(shù)+1，成功（代表線程數(shù)量匹配上，并要創(chuàng)建新線程所以+1）退出自旋返回到retry:
//ctl的值就+1了
if (compareAndIncrementWorkerCount(c))
break retry;
//比較不成功，說明線程數(shù)可能已經(jīng)被其它線程修改過了，重新讀一次
c = ctl.get();
//得到的線程池狀態(tài)跟初始進(jìn)來的狀態(tài)不一樣，則continue retry到retry:重新執(zhí)行for
if (runStateOf(c) != rs)
continue retry;
}
}
//到這里已經(jīng)具備創(chuàng)建worker的條件，下面開始創(chuàng)建worker
//worker啟動標(biāo)志
boolean workerStarted = false;
//worker已經(jīng)被添加標(biāo)志
boolean workerAdded = false;
Worker w = null;
try {
//創(chuàng)建一新的Worker，在worker構(gòu)造方法里對線程池的狀態(tài)設(shè)置為-1
//并且線程工程創(chuàng)建線程，有第一個(gè)任務(wù)也會設(shè)置進(jìn)去
w = new Worker(firstTask);
//在worker創(chuàng)建時(shí)初始化的線程
final Thread t = w.thread;
//線程初始化成功
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
//這是整個(gè)線程池的全局鎖，持有這個(gè)鎖才能讓下面的操作順理成章
//因?yàn)殛P(guān)閉一個(gè)線程池需要這個(gè)鎖，至少我持有鎖的期間，線程池不會被關(guān)閉
mainLock.lock();
try {
//ctl這個(gè)值在前面cas已經(jīng)設(shè)置+1
//重新獲取線程池的狀態(tài)
int rs = runStateOf(ctl.get());
//rs < SHUTDOWN：處于RUNNING,正常情況
//(rs == SHUTDOWN && firstTask == null)：處于SHUTDOWN不接受新的任務(wù)，但是會執(zhí)行等待隊(duì)列的任務(wù)
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
//新增的worker里的線程可不能是已經(jīng)啟動的
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
//到這里，worker已經(jīng)初始化好，并添加到hashSet里
workers.add(w);
//查看worker集合的大小
int s = workers.size();
//largestPoolSize是用于記錄worker中個(gè)數(shù)的最大值
//當(dāng)worker集合數(shù)量已經(jīng)大于它，則將其調(diào)整為當(dāng)前worker集合數(shù)量
if (s > largestPoolSize)
largestPoolSize = s;
//表示worker添加成功
workerAdded = true;
}
} finally {
//worker一系列添加后釋放鎖
mainLock.unlock();
}
//如果worker已經(jīng)添加成，啟動線程會調(diào)用runWorker方法，并設(shè)置線程已經(jīng)啟動標(biāo)志
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
//worker線程啟動失敗
if (!workerStarted)
addWorkerFailed(w);
}
//返回線程是否啟動標(biāo)志
return workerStarted;
}
簡單看下 addWorkFailed 的處理：
tryTerminate();方法后面統(tǒng)一講，因?yàn)楹芏嗟胤秸{(diào)用它
//添加worker失敗
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//線程啟動失敗，但是worker已經(jīng)創(chuàng)建好了，則將創(chuàng)建好的worker從集合移除
if (w != null)
workers.remove(w);
//死循環(huán)，cas比較去將線程數(shù)整數(shù)減一，直到減成功，ctl的值就減一
decrementWorkerCount();
//重新檢查是否終止，以防該worker的存在阻止了終止
tryTerminate();
} finally {
//釋放鎖
mainLock.unlock();
}
}
cas更新線程數(shù)，線程數(shù)減一
//循環(huán)設(shè)置ctl減少，直到更新成功
private void decrementWorkerCount() {
do {
} while (!compareAndDecrementWorkerCount(ctl.get()));
}
每個(gè)worker怎么獲取任務(wù)呢？
- 自旋獲取線程狀態(tài)，線程狀態(tài)不處于RUNNING或者不處于（SHUTDOWN+任務(wù)隊(duì)列不為空），則線程數(shù)減一，任務(wù)返回空。
- 自旋獲取線程狀態(tài)，線程狀態(tài)處于RUNNING或者處于（SHUTDOWN+任務(wù)隊(duì)列不為空），則當(dāng)前線程數(shù)如果大于最大線程數(shù)并且任務(wù)隊(duì)列時(shí)刻，或者當(dāng)前線程數(shù)大于核心線程數(shù)并且超時(shí)并且隊(duì)列是空，那么cas設(shè)置線程減一，繼續(xù)自旋，否則指定keepAliveTime超時(shí)時(shí)間去讀取任務(wù)或者直接阻塞讀取任務(wù)。
- 讀取任務(wù)的時(shí)候會被阻塞，如果tryTerminal里的中斷信號發(fā)出，這里就會識別，則拋出中斷異常，線程醒過來，設(shè)置超時(shí)標(biāo)識，繼續(xù)進(jìn)行自旋。
- 如果讀取的任務(wù)不為空，則直接返回任務(wù)。
流程圖如下：
// 如果發(fā)生了以下四件事中的任意一件，那么Worker需要被回收：
// 1. Worker個(gè)數(shù)比線程池最大大小要大
// 2. 線程池處于STOP狀態(tài)
// 3. 線程池處于SHUTDOWN狀態(tài)并且阻塞隊(duì)列為空
// 4. 使用超時(shí)時(shí)間從阻塞隊(duì)列里拿數(shù)據(jù)，并且超時(shí)之后沒有拿到數(shù)據(jù)(allowCoreThreadTimeOut || workerCount > corePoolSize)
private Runnable getTask() {
// 如果使用超時(shí)時(shí)間并且也沒有拿到任務(wù)的標(biāo)識
boolean timedOut = false;
for (; ; ) {
int c = ctl.get();
int rs = runStateOf(c);
//worker減一是因?yàn)榍懊嫣砑觲orker的時(shí)候啟動線程已經(jīng)添加成功，而這盤點(diǎn)不符合執(zhí)行任務(wù)，所以把添加的worker數(shù)量減掉
//在processWorkerExit進(jìn)行回收
// 如果線程池是SHUTDOWN狀態(tài)并且阻塞隊(duì)列為空的話，worker數(shù)量減一，
// 直接返回null(SHUTDOWN狀態(tài)還會處理阻塞隊(duì)列任務(wù)，但是阻塞隊(duì)列為空的話就結(jié)束了)，
// 如果線程池是STOP狀態(tài)的話，worker數(shù)量減一，
// 直接返回null(STOP狀態(tài)不處理阻塞隊(duì)列任務(wù))
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
//當(dāng)前線程數(shù)
int wc = workerCountOf(c);
//如果設(shè)置了allowCoreThreadTimeOut則可能超時(shí)
//標(biāo)記從隊(duì)列中取任務(wù)時(shí)是否設(shè)置超時(shí)時(shí)間，
// 如果為true說明這個(gè)worker可能需要回收，
// 為false的話這個(gè)worker會一直存在，并且阻塞當(dāng)前線程等待阻塞隊(duì)列中有數(shù)據(jù)
// allowCoreThreadTimeOut屬性默認(rèn)為false，表示線程池中的核心線程在閑置狀態(tài)下還保留在池中；
// 如果是true表示核心線程使用keepAliveTime這個(gè)參數(shù)來作為超時(shí)時(shí)間
// 如果worker數(shù)量比基本大小要大的話，timed就為true，需要進(jìn)行回收worker
// timed變量用于判斷是否需要進(jìn)行超時(shí)控制。
// allowCoreThreadTimeOut默認(rèn)是false，也就是核心線程不允許進(jìn)行超時(shí)；
// wc > corePoolSize，表示當(dāng)前線程池中的線程數(shù)量大于核心線程數(shù)量；
// 對于超過核心線程數(shù)量的這些線程，需要進(jìn)行超時(shí)控制
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
//4中情況分析：
//第一：wc > maximumPoolSize && wc > 1 大于了maximumPoolSize必然大于1，所以不獲取任務(wù)
//第二：wc > maximumPoolSize && workQueue.isEmpty() 阻塞隊(duì)列的沒有任務(wù)了，也就沒必要獲取
//第三：(timed && timedOut) && wc>1超時(shí)了，有線程在運(yùn)行，就沒必要獲取任務(wù)
//第四：(timed && timedOut) && workQueue.isEmpty()超時(shí)了，任務(wù)隊(duì)列沒任務(wù)，則沒必要獲取
//(timed && timedOut)表示當(dāng)前線程大于核心數(shù)并且前面獲取任務(wù)poll超時(shí)沒獲取到任務(wù)，則timedOut=true,
//這個(gè)就體現(xiàn)了keepAliveTime的用處，超時(shí)沒獲取到任務(wù)，并且當(dāng)前線程超高了cpu核心，因?yàn)闆]獲取到任務(wù)worker就閑置
//所以這里線程數(shù)減一，在外面進(jìn)行worker的回收
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
//線程數(shù)減一
if (compareAndDecrementWorkerCount(c))
//不用執(zhí)行任務(wù)
return null;
continue;
}
try {
//poll:從BlockingQueue取出一個(gè)隊(duì)首的對象，如果在指定時(shí)間內(nèi)，隊(duì)列一旦有數(shù)據(jù)可取，則立即返回隊(duì)列中的數(shù)據(jù)。
// 否則直到時(shí)間超時(shí)還沒有數(shù)據(jù)可取，返回失敗。
//take:取走BlockingQueue里排在首位的對象,若BlockingQueue為空,
// 阻斷進(jìn)入等待狀態(tài)直到BlockingQueue有新的數(shù)據(jù)被加入
//設(shè)置了allowCoreThreadTimeOut或者當(dāng)前線程已經(jīng)大于cpu核心數(shù)則以keepAliveTime超時(shí)時(shí)間獲取任務(wù)
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
//任務(wù)取到，返回任務(wù)
if (r != null)
return r;
//到這里任務(wù)是空，設(shè)置超時(shí)標(biāo)識
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
每個(gè)worker是怎么執(zhí)行任務(wù)的呢
添加worker成功后會去啟動線程，這時(shí)候調(diào)用ThreadPoolExecutor的run方法，然后執(zhí)行runWorker方法。
那接下來看下runWorker方法
- 先釋放當(dāng)前當(dāng)前worker的鎖，這樣支持獲取任務(wù)的時(shí)候可以多個(gè)線程去搶占woker，在if (!t.isInterrupted() && w.tryLock())才能獲取到鎖，如果獲取到鎖說明線程是閑置狀態(tài)
- 獲取任務(wù)
- 獲取到任務(wù)，進(jìn)行全局加鎖
- 如果線程池已經(jīng)處于STOP狀態(tài)以上線程卻沒有終端，則中斷當(dāng)前線程，也就是不去獲取任務(wù)了，但是會繼續(xù)執(zhí)行當(dāng)前任務(wù)，然后完成任務(wù)數(shù)量+1，解鎖，異常完成任務(wù)標(biāo)識設(shè)置為false。
- 回收worker。
先看下流程圖：
// 此方法由 worker 線程啟動后調(diào)用，這里用一個(gè) while 循環(huán)來不斷地從等待隊(duì)列中獲取任務(wù)并執(zhí)行
// 前面說了，worker 在初始化的時(shí)候，可以指定 firstTask，那么第一個(gè)任務(wù)也就可以不需要從隊(duì)列中獲取
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
// 該線程的第一個(gè)任務(wù)(如果有的話)
Runnable task = w.firstTask;
w.firstTask = null;
//由于在創(chuàng)建Worker的時(shí)候設(shè)置狀態(tài)是-1，其它線程不能得到worker的鎖。這里釋放后其它線程可以獲得worker的鎖
//在getask的時(shí)候可以支持中斷,如果這邊不unlock的話，在if (!t.isInterrupted() && w.tryLock())就一直認(rèn)為它是忙的狀態(tài)
//則無法打斷因?yàn)楂@取任務(wù)而阻塞的線程，它只是在獲取任務(wù)而不是執(zhí)行任務(wù)，也是閑置的線程
w.unlock();
//是否正常結(jié)束任務(wù)標(biāo)識
boolean completedAbruptly = true;
try {
// 如果worker中的任務(wù)不為空，否則使用getTask獲得任務(wù)。一直死循環(huán)，除非得到的任務(wù)為空才退出
//當(dāng)然getTask如果沒拿到任務(wù)會一直阻塞直到拿到任務(wù)
while (task != null || (task = getTask()) != null) {
// 如果拿到了任務(wù)，給自己上鎖，表示當(dāng)前Worker已經(jīng)要開始執(zhí)行任務(wù)了，
// 已經(jīng)不是閑置Worker(閑置Worker的解釋請看下面的線程池關(guān)閉)
w.lock();
// 在執(zhí)行任務(wù)之前先做一些處理。
// 1. 如果線程池已經(jīng)大于等于STOP狀態(tài)并且當(dāng)前線程沒有被中斷，中斷線程
// 2. 如果線程池還處于RUNNING或SHUTDOWN狀態(tài)，并且當(dāng)前線程已經(jīng)被中斷了，
// 重新檢查一下線程池狀態(tài)，如果處于STOP狀態(tài)并且沒有被中斷，那么中斷線程
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
// 任務(wù)執(zhí)行前需要做什么，ThreadPoolExecutor是個(gè)空實(shí)現(xiàn)
beforeExecute(wt, task);
Throwable thrown = null;
try {
// 真正的開始執(zhí)行任務(wù)，調(diào)用的是run方法，而不是start方法。這里run的時(shí)候可能會被中斷，比如線程池調(diào)用了shutdownNow方法
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 {
// 任務(wù)執(zhí)行結(jié)束需要做什么，ThreadPoolExecutor是個(gè)空實(shí)現(xiàn)
afterExecute(task, thrown);
}
} finally {
task = null;
// 記錄執(zhí)行任務(wù)的個(gè)數(shù)
w.completedTasks++;
// 執(zhí)行完任務(wù)之后，解鎖，Worker變成閑置Worker
w.unlock();
}
}
//正常執(zhí)行完任務(wù)，改成false
completedAbruptly = false;
} finally {
// 回收Worker
processWorkerExit(w, completedAbruptly);
}
}
任務(wù)執(zhí)行的worker應(yīng)該怎么處理？
會對worker進(jìn)行一個(gè)回收
- 判斷在執(zhí)行任務(wù)的時(shí)候有沒有異常，如果有則會將線程數(shù)量通過cas比較減一，因?yàn)樘砑觲orker的時(shí)候已加入隊(duì)列，失敗了就要減掉
- 任務(wù)正常執(zhí)行，統(tǒng)計(jì)線程完成任務(wù)數(shù)，移除當(dāng)前worker，嘗試關(guān)閉線程池
- 若線程池還處于RUNING和(SHUTDOWN+任務(wù)隊(duì)列非空)，則還不能關(guān)閉線程池，再判斷運(yùn)行任務(wù)的時(shí)候是否正常運(yùn)行，是就開始設(shè)置線程數(shù)量最小值，設(shè)置了allowCoreThreadTimeOut最小值為0，沒設(shè)置則cpu核心數(shù)，比較最小值，如果最小值為0并且還有任務(wù)為完成，則最小值改成1(執(zhí)行任務(wù))。當(dāng)前線程數(shù)大于等于最小線程，則不用再創(chuàng)建worker,如果當(dāng)前線程數(shù)量小于最小值添加新worker
只有以下幾種情況才會在回收worker還會補(bǔ)償創(chuàng)建worker：
- 線程池處于RUNING和(SHUTDOWN+任務(wù)隊(duì)列非空)則任務(wù)還沒執(zhí)行完，前面執(zhí)行任務(wù)又失敗worker被回收，這里需要補(bǔ)償創(chuàng)建
- 線程池處于RUNING和(SHUTDOWN+任務(wù)隊(duì)列非空)則任務(wù)還沒執(zhí)行完，前面執(zhí)行任務(wù)成功，但是當(dāng)前線程數(shù)小于線程數(shù)量最小值，這里需要補(bǔ)償創(chuàng)建
流程圖如下：
//回收Worker
private void processWorkerExit(Worker w, boolean completedAbruptly) {
// 如果Worker沒有正常結(jié)束流程調(diào)用processWorkerExit方法，worker數(shù)量減一。
// 如果是正常結(jié)束的話，在getTask方法里worker數(shù)量已經(jīng)減一了
if (completedAbruptly)
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//退出前，將本線程已完成的任務(wù)數(shù)量，添加到已經(jīng)完成任務(wù)的總數(shù)中
completedTaskCount += w.completedTasks;
// 線程池的worker集合刪除掉需要回收的Worker
workers.remove(w);
} finally {
mainLock.unlock();
}
// 嘗試結(jié)束線程池
tryTerminate();
int c = ctl.get();
// 如果線程池還處于RUNNING或者SHUTDOWN狀態(tài)
if (runStateLessThan(c, STOP)) {
//如果非異常狀況completedAbruptly=false，也就是沒有獲取到可執(zhí)行的任務(wù)，則獲取線程池允許的最小線程數(shù)，
// 如果allowCoreThreadTimeOut為true說明允許核心線程超時(shí)，則最小線程數(shù)為0，否則最小線程數(shù)為corePoolSize;
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
//如果allowCoreThreadTimeOut=true，且任務(wù)隊(duì)列有任務(wù)要執(zhí)行，則將最最小線程數(shù)設(shè)置為1
if (min == 0 && !workQueue.isEmpty())
min = 1;
//如果當(dāng)前線程數(shù)大于等于最小線程數(shù)，則直接返回
if (workerCountOf(c) >= min)
// 不需要新開一個(gè)Worker
return;
}
// 新開一個(gè)Worker代替原先的Worker
// 新開一個(gè)Worker需要滿足以下3個(gè)條件中的任意一個(gè)：
// 1. 獲取執(zhí)行的任務(wù)發(fā)生了異常
// 2. Worker數(shù)量比線程池基本大小要小
// 3. 阻塞隊(duì)列不空但是沒有任何Worker在工作
addWorker(null, false);
}
}
execute
先看下流程圖：
- 如果當(dāng)前線程數(shù)小于核心數(shù)量，直接創(chuàng)建worker
- 如果線程池狀態(tài)是RUNNING并且任務(wù)加到等待隊(duì)列成功，還要重新確認(rèn)一次線程池的狀態(tài)，若還是RUNNING，則當(dāng)前線程數(shù)等于0的話直接創(chuàng)建worker，如果不是RUNNING并且隊(duì)列移除任務(wù)成功，執(zhí)行拒絕策略，默認(rèn)拋出異常
- 如果 workQueue 隊(duì)列滿了，以 maximumPoolSize 為界創(chuàng)建新的 worker， 如果失敗，說明當(dāng)前線程數(shù)已經(jīng)達(dá)到 maximumPoolSize，執(zhí)行拒絕策略
有了上面的這些基礎(chǔ)后，我們終于可以看看 ThreadPoolExecutor 的 execute 方法了，前面源碼分析的時(shí)候也說了，各種方法都最終依賴于 execute 方法，也就是線程池執(zhí)行任務(wù)的入口：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
// 前面說的那個(gè)表示 “線程池狀態(tài)” 和 “線程數(shù)” 的整數(shù)
int c = ctl.get();
//得到線程池的線程數(shù)，如果線程數(shù)小于cpu核心數(shù)
//新增worker加入hashSet成功，就直接返回了
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
//新增worker加入hashSet失敗，則重新確認(rèn)ctl的值
c = ctl.get();
}
//到這里添加worker失敗了，需要把任務(wù)放到等待隊(duì)列
//線程池還在RUNNING狀態(tài)，阻塞隊(duì)列也沒滿的情況，加到阻塞隊(duì)列里
if (isRunning(c) && workQueue.offer(command)) {
/* 這里面說的是，如果任務(wù)進(jìn)入了 workQueue，我們是否需要開啟新的線程
* 因?yàn)榫€程數(shù)在 [0, corePoolSize) 是無條件開啟新的線程
* 如果線程數(shù)已經(jīng)大于等于 corePoolSize，那么將任務(wù)添加到隊(duì)列中，然后進(jìn)到這里
*/
int recheck = ctl.get();
// 如果線程池已不處于 RUNNING 狀態(tài)，那么移除已經(jīng)入隊(duì)的這個(gè)任務(wù)，并且執(zhí)行拒絕策略
if (!isRunning(recheck) && remove(command))
//默認(rèn)的拒絕策略直接拋異常
reject(command);
// 如果線程池還是 RUNNING 的，并且線程數(shù)為 0，那么開啟新的線程
// 到這里，我們知道了，這塊代碼的真正意圖是：擔(dān)心任務(wù)提交到隊(duì)列中了，但是線程都關(guān)閉了
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 如果 workQueue 隊(duì)列滿了，那么進(jìn)入到這個(gè)分支
// 以 maximumPoolSize 為界創(chuàng)建新的 worker，
// 如果失敗，說明當(dāng)前線程數(shù)已經(jīng)達(dá)到 maximumPoolSize，執(zhí)行拒絕策略
else if (!addWorker(command, false))
reject(command);
}
如果等待隊(duì)列滿了后對任務(wù)怎么處理的
是有四種不同的拒絕策略
final void reject(Runnable command) {
handler.rejectedExecution(command, this);
}
- AbortPolicy(默認(rèn)策略)直接拋出異常拒絕接收任務(wù)
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
- CallerRunsPolicy如果線程池在運(yùn)行，以當(dāng)前線程完成任務(wù)
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
- DiscardOldestPolicy丟棄等待最久的任務(wù)，繼續(xù)執(zhí)行其它任務(wù)
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
- DiscardPolicy不做任何處理，新的任務(wù)直接忽略
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
線程池的關(guān)閉
線程池的關(guān)閉有兩個(gè)方法shutdown() 與shutdownNow() ；
shutdown會將線程池狀態(tài)設(shè)置為SHUTDOWN狀態(tài)，然后中斷所有空閑線程，然后執(zhí)行tryTerminate()方法（tryTerminate這個(gè)方法很重要，會在后面分析），來嘗試終止線程池;
shutdownNow會將線程池狀態(tài)設(shè)置為STOP狀態(tài)，然后中斷所有線程（不管有沒有執(zhí)行任務(wù)都設(shè)置為中斷狀態(tài)），然后執(zhí)行tryTerminate()方法，來嘗試終止線程池;
- shutdown
流程圖如下：
這里的中斷只是發(fā)出中斷信號，而在獲取任務(wù)阻塞的線程會識別到對應(yīng)的中斷信號，然后拋出異常，執(zhí)行完方法退出
//嘗試停止線程池，此時(shí)不接受新的任務(wù)，但是會處理等待隊(duì)列的任務(wù)
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//檢查關(guān)閉線程池的權(quán)限
checkShutdownAccess();
//把線程池狀態(tài)更新到SHUTDOWN
advanceRunState(SHUTDOWN);
//中斷閑置的Worker
interruptIdleWorkers();
// 鉤子方法，默認(rèn)不處理。ScheduledThreadPoolExecutor會做一些處理
onShutdown();
} finally {
mainLock.unlock();
}
//嘗試結(jié)束線程池
tryTerminate();
}
//更新指定狀態(tài)
private void advanceRunState(int targetState) {
for (; ; ) {
int c = ctl.get();
//假設(shè)targetState=SHUTDOWN以下兩種情況會結(jié)束自旋
//1.當(dāng)前線程池狀態(tài)不是運(yùn)行狀態(tài)
//2.當(dāng)前線程狀態(tài)是運(yùn)行狀態(tài)，但是cas比較更新當(dāng)前線程數(shù)和線程池狀態(tài)為SHUTDOWN成功
if (runStateAtLeast(c, targetState) ||
ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
break;
}
}
中斷閑置線程
//傳true,中斷一個(gè)worker,false則中斷全部閑置的worker，
private void interruptIdleWorkers() {
interruptIdleWorkers(false);
}
//傳入了參數(shù)false，表示要中斷所有的正在運(yùn)行的閑置Worker，如果為true表示只打斷一個(gè)閑置Worker
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
Thread t = w.thread;
// Worker中的線程沒有被打斷并且Worker可以獲取鎖，這里Worker能獲取鎖說明Worker是個(gè)閑置Worker，
// 在阻塞隊(duì)列里拿數(shù)據(jù)一直被阻塞，沒有數(shù)據(jù)進(jìn)來。如果沒有獲取到Worker鎖，說明Worker還在執(zhí)行任務(wù)，
// 不進(jìn)行中斷(shutdown方法不會中斷正在執(zhí)行的任務(wù))
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
- shutdownNow
流程圖如下：
這里的中斷只是發(fā)出中斷信號，而在獲取任務(wù)阻塞的線程會識別到對應(yīng)的中斷信號，然后拋出異常，執(zhí)行完方法退出
//嘗試停止線程池，此時(shí)不接受新的任務(wù)，也不會處理等待隊(duì)列的任務(wù)
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//檢查關(guān)閉線程池的權(quán)限
checkShutdownAccess();
//把線程池狀態(tài)更新到STOP
advanceRunState(STOP);
//中斷所有Worker，不管是否閑置
interruptWorkers();
//清空等待隊(duì)列的任務(wù)，并且返回清除成功的這些任務(wù)的集合
tasks = drainQueue();
} finally {
mainLock.unlock();
}
//嘗試結(jié)束線程池
tryTerminate();
//返回清除成功的這些任務(wù)的集合
return tasks;
}
更新指定狀態(tài)
//更新指定狀態(tài)
private void advanceRunState(int targetState) {
for (; ; ) {
int c = ctl.get();
//假設(shè)targetState=SHUTDOWN以下兩種情況會結(jié)束自旋
//1.當(dāng)前線程池狀態(tài)不是運(yùn)行狀態(tài)
//2.當(dāng)前線程狀態(tài)是運(yùn)行狀態(tài)，但是cas比較更新當(dāng)前線程數(shù)和線程池狀態(tài)為SHUTDOWN成功
if (runStateAtLeast(c, targetState) ||
ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
break;
}
}
停止所有的啟動線程，不管是否閑置
//停止所有的啟動線程，不管是否閑置
private void interruptWorkers() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers)
w.interruptIfStarted();
} finally {
mainLock.unlock();
}
}
傳入了參數(shù)false，表示要中斷所有的正在運(yùn)行的閑置Worker，如果為true表示只打斷一個(gè)閑置Worker
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
對阻塞隊(duì)列的任務(wù)進(jìn)行清空
//對阻塞隊(duì)列的任務(wù)進(jìn)行清空
private List<Runnable> drainQueue() {
BlockingQueue<Runnable> q = workQueue;
ArrayList<Runnable> taskList = new ArrayList<Runnable>();
// 該方法會將阻塞隊(duì)列中的所有項(xiàng)添加到 taskList 中
// 然后清空任務(wù)隊(duì)列，該方法是線程安全的
q.drainTo(taskList);
//隊(duì)列還沒清空？
if (!q.isEmpty()) {
// 將 List 轉(zhuǎn)換為 數(shù)組，傳入的 Runnable[0] 用來說明是轉(zhuǎn)為 Runnable 數(shù)組
//也就是q轉(zhuǎn)成了Runnable[],q有幾個(gè)元素，Runnable[]有幾個(gè)元素
for (Runnable r : q.toArray(new Runnable[0])) {
//從隊(duì)列里移除指定元素
if (q.remove(r))
//移除成功，添加到任務(wù)集合
taskList.add(r);
}
}
//返回清空成功的任務(wù)集合
return taskList;
}
關(guān)于finalize方法說明：
垃圾回收時(shí)，如果判斷對象不可達(dá)，且覆蓋了finalize方法，則會將對象放入到F-Queue隊(duì)列 ，有一個(gè)名為”Finalizer”的守護(hù)線程執(zhí)行finalize方法，它的優(yōu)先級為8，做最后的清理工作，執(zhí)行finalize方法完畢后，GC會再次判斷該對象是否可達(dá)，若不可達(dá)，則進(jìn)行回收，否則，對象復(fù)活
注意：網(wǎng)上很多人說 ，F(xiàn)inalizer線程的優(yōu)先級低，個(gè)人認(rèn)為這是不對的，F(xiàn)inalizer線程在jdk1.8的優(yōu)先級是8，比我們創(chuàng)建線程默認(rèn)優(yōu)先級5要高，之前其它版本的jdk我記得導(dǎo)出的線程棧信息里面優(yōu)先級是5，忘記是哪個(gè)版本的jdk了，即使是5優(yōu)先級也不比自建的線程默認(rèn)優(yōu)先級低，總之我沒見過優(yōu)先級低于5的Finalizer線程；
這個(gè)線程會不停的循環(huán)等待java.lang.ref.Finalizer.ReferenceQueue中的新增對象。一旦Finalizer線程發(fā)現(xiàn)隊(duì)列中出現(xiàn)了新的對象，它會彈出該對象，調(diào)用它的finalize()方法，將該引用從Finalizer類中移除，因此下次GC再執(zhí)行的時(shí)候，這個(gè)Finalizer實(shí)例以及它引用的那個(gè)對象就可以回垃圾回收掉了。
大多數(shù)時(shí)候，F(xiàn)inalizer線程能夠趕在下次GC帶來更多的Finalizer對象前清空這個(gè)隊(duì)列,但是當(dāng)它的處理速度沒法趕上新對象創(chuàng)建的速度，對象創(chuàng)建的速度要比Finalizer線程調(diào)用finalize()結(jié)束它們的速度要快，這導(dǎo)致最后堆中所有可用的空間都被耗盡了;
當(dāng)我們大量線程頻繁創(chuàng)建重寫了finalizer（）方法的對象的情況下，高并發(fā)情況下，它可能會導(dǎo)致你內(nèi)存的溢出；雖然Finalizer線程優(yōu)先級高，但是畢竟它只有一個(gè)線程；最典型的例子就是數(shù)據(jù)庫連接池,proxool，對要釋放資源的操作加了鎖，并在finalized方法中調(diào)用該加鎖方法，在高并發(fā)情況下，鎖競爭嚴(yán)重，finalized競爭到鎖的幾率減少，finalized無法立即釋放資源，越來越多的對象finalized()方法無法被執(zhí)行，資源無法被回收，最終導(dǎo)致導(dǎo)致oom；所以覆蓋finalized方法，執(zhí)行一定要快，不能有鎖競爭的操作，否則在高并發(fā)下死的很慘；
嘗試終止線程池tryTerminate
該方法會在很多地方調(diào)用，如添加worker線程失敗的addWorkerFailed（)方法，worker線程跳出執(zhí)行任務(wù)的while 循環(huán)退出時(shí)的processWorkerExit（）方法，關(guān)閉線程池的shutdown()和shutdownNow()方法，從任務(wù)隊(duì)列移除任務(wù)的remove()方法；
該方法的作用是檢測當(dāng)前線程池的狀態(tài)是否可以將線程池終止，如果可以終止則嘗試著去終止線程，否則直接返回；
STOP->TIDYING 與SHUTDOWN->TIDYING狀態(tài)的轉(zhuǎn)換，就是在該方法中實(shí)現(xiàn)的，最終執(zhí)行terminated()方法后會把線程狀態(tài)設(shè)置為TERMINATED的狀態(tài)；
嘗試終止線程池執(zhí)行過程；
- 重點(diǎn)內(nèi)容先判斷線程池的狀態(tài)是否允許被終止
以下狀態(tài)不可被終止：
1.如果線程池的狀態(tài)是RUNNING（不可終止）
或者是TIDYING（該狀態(tài)一定執(zhí)行過了tryTerminate方法，正在執(zhí)行或即將執(zhí)行terminated()方法，所以不需要重復(fù)執(zhí)行），
或者是TERMINATED（該狀態(tài)已經(jīng)執(zhí)行完成terminated()鉤子方法，已經(jīng)是被終止?fàn)顟B(tài)了），
以上三種狀態(tài)直接返回。
2.如果線程池狀態(tài)是SHUTDOWN，而且任務(wù)隊(duì)列不是空的（該狀態(tài)需要繼續(xù)處理任務(wù)隊(duì)列中的任務(wù)，不可被終止），也直接返回。
以下兩種狀態(tài)線程池可以被終止：
1.如果線程池狀態(tài)是SHUTDOWN，而且任務(wù)隊(duì)列是空的（shutdown狀態(tài)下，任務(wù)隊(duì)列為空，可以被終止），向下進(jìn)行。
2.如果線程池狀態(tài)是STOP（該狀態(tài)下，不接收新任務(wù)，不執(zhí)行任務(wù)隊(duì)列中的任務(wù)，并中斷正在執(zhí)行中的線程，可以被終止），向下進(jìn)行。
- 線程池狀態(tài)可以被終止，如果線程池中仍然有線程，則嘗試中斷線程池中的線程
則嘗試中斷一個(gè)線程然后返回,被中斷的這個(gè)線程執(zhí)行完成退出后，又會調(diào)用tryTerminate()方法，中斷其它線程，直到線程池中的線程數(shù)為0，則繼續(xù)往下執(zhí)行； - 如果線程池中的線程為0，則將狀態(tài)設(shè)置為TIDYING，設(shè)置成功后執(zhí)行 terminated()方法，最后將線程狀態(tài)設(shè)置為TERMINATED
源碼如下：
//嘗試關(guān)閉線程池
//嘗試關(guān)閉線程池
final void tryTerminate() {
for (; ; ) {
int c = ctl.get();
// 滿足3個(gè)條件中的任意一個(gè)，不終止線程池
// 1. 線程池還在運(yùn)行，不能終止
// 2. 線程池處于TIDYING或TERMINATED狀態(tài)，說明已經(jīng)在關(guān)閉了，不允許繼續(xù)處理
// 3. 線程池處于SHUTDOWN狀態(tài)并且阻塞隊(duì)列不為空，這時(shí)候還需要處理阻塞隊(duì)列的任務(wù)，不能終止線程池
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && !workQueue.isEmpty()))
return;
// 走到這一步說明線程池已經(jīng)不在運(yùn)行，阻塞隊(duì)列已經(jīng)沒有任務(wù)，但是還要回收正在工作的Worker
//還剩下 SHUTDOWN&&workQueue.isEmpty 、STOP這兩種狀態(tài)
if (workerCountOf(c) != 0) {
// 由于線程池不運(yùn)行了，調(diào)用了線程池的關(guān)閉方法，在解釋線程池的關(guān)閉原理的時(shí)候會說道這個(gè)方法
// 中斷閑置Worker，直到回收全部的Worker。這里沒有那么暴力，只中斷一個(gè)閑置線程，
//發(fā)出中斷信號，中斷阻塞在獲取任務(wù)的線程，然后還是會調(diào)用tryTerminate方法，如果還有閑置線程，那么繼續(xù)中斷
interruptIdleWorkers(ONLY_ONE);
return;
}
// 走到這里說明worker已經(jīng)全部回收了，并且線程池已經(jīng)不在運(yùn)行，阻塞隊(duì)列已經(jīng)沒有任務(wù)?？梢詼?zhǔn)備結(jié)束線程池了
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// cas操作，將線程池狀態(tài)改成TIDYING
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
//鉤子方法，沒有任何實(shí)現(xiàn)
terminated();
} finally {
//terminated方法調(diào)用完畢之后，狀態(tài)變?yōu)門ERMINATED
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
傳入了參數(shù)false，表示要中斷所有的正在運(yùn)行的閑置Worker，如果為true表示只打斷一個(gè)閑置Worker
//傳入了參數(shù)false，表示要中斷所有的正在運(yùn)行的閑置Worker，如果為true表示只打斷一個(gè)閑置Worker
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
Thread t = w.thread;
// Worker中的線程沒有被打斷并且Worker可以獲取鎖，這里Worker能獲取鎖說明Worker是個(gè)閑置Worker，
// 在阻塞隊(duì)列里拿數(shù)據(jù)一直被阻塞，沒有數(shù)據(jù)進(jìn)來。如果沒有獲取到Worker鎖，說明Worker還在執(zhí)行任務(wù)，
// 不進(jìn)行中斷(shutdown方法不會中斷正在執(zhí)行的任務(wù))
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
總結(jié)幾個(gè)問題：
- 說說線程池中的線程創(chuàng)建時(shí)機(jī)？
1. 線程池處于RUNNING和（SHUTDOWN+等待隊(duì)列不為空）狀態(tài)，并且當(dāng)前線程數(shù)量小于核心數(shù)，
那么提交任務(wù)的時(shí)候創(chuàng)建一個(gè)新的線程，并由這個(gè)線程執(zhí)行這個(gè)任務(wù)。
2. 如果當(dāng)前線程數(shù)已經(jīng)達(dá)到 corePoolSize，那么將提交的任務(wù)添加到隊(duì)列中，
等待線程池中的線程去隊(duì)列中取任務(wù)。
3. 如果隊(duì)列已滿，那么創(chuàng)建新的線程來執(zhí)行任務(wù)，需要保證池中的線程數(shù)不會超過 maximumPoolSize，
如果此時(shí)線程數(shù)超過了 maximumPoolSize，那么執(zhí)行拒絕策略。
- 注意：如果將隊(duì)列設(shè)置為無界隊(duì)列，那么線程數(shù)達(dá)到 corePoolSize 后，其實(shí)線程數(shù)就不會再增長了。因?yàn)楹竺娴娜蝿?wù)直接往隊(duì)列塞就行了，此時(shí) maximumPoolSize 參數(shù)就沒有什么意義。
- 任務(wù)執(zhí)行過程中發(fā)生異常怎么處理？
如果某個(gè)任務(wù)執(zhí)行出現(xiàn)異常，那么執(zhí)行任務(wù)的線程會被關(guān)閉，而不是繼續(xù)接收其他任務(wù)。
然后經(jīng)過一系列確認(rèn)，比如線程池的狀態(tài)是否啟動、有沒有未完成的任務(wù)不夠線程執(zhí)行等等，
然后會啟動一個(gè)新的線程來代替它。
- 什么時(shí)候會執(zhí)行拒絕策略？
1、workers 的數(shù)量達(dá)到了 corePoolSize（任務(wù)此時(shí)需要進(jìn)入任務(wù)隊(duì)列），任務(wù)入隊(duì)成功，
與此同時(shí)線程池被關(guān)閉了并且剛剛加入隊(duì)列的任務(wù)移除成功，那么執(zhí)行拒絕策略。
2、workers 的數(shù)量大于等于 corePoolSize，將任務(wù)加入到任務(wù)隊(duì)列，可是隊(duì)列滿了，任務(wù)入隊(duì)失敗，
那么準(zhǔn)備開啟新的線程，可是線程數(shù)已經(jīng)達(dá)到 maximumPoolSize，那么執(zhí)行拒絕策略。
線程池看了幾天，一開始看得有點(diǎn)蒙蔽，慢慢才看懂，喜歡給個(gè)贊?。。。?！
