深入理解Java多執行緒--執行緒池（ThreadPool）
在java多執行緒開發中，我們需要使用執行緒的時候一般是建立一個Thread物件，然後呼叫start（）方法去執行執行緒操作。這樣做沒有什麼問題，但是如果我們有很多工需要多個執行緒來非同步執行的時候，在我們建立了很多執行緒的情況下，會造成很大的效能方面的問題。
1.大量的執行緒的建立和銷燬，本身就是一個很大的效能開銷。
2.大量執行緒同時運作的時候， 會造成資源緊張，我們知道執行緒的底層機制就是切分CPU的時間，在大量執行緒互相搶佔資源的情況下，可能會造成阻塞現象。
基於上面的緣由，JDK提供了執行緒池可以很好地解決這種問題。執行緒池可以幫我們管理一定數量的執行緒，重複使用執行緒可以避免執行緒的大量建立，控制執行緒任務的大量併發， 可以避免因為CPU資源的競爭而造成的阻塞。
執行緒池的使用：
關於執行緒池，我們首先要了解ThreadPoolExecutor 類，我們建立執行緒池就是通過這個類去建立的，它的建構函式如下：
public ThreadPoolExecutor(int corePoolSize, //核心執行緒數量 int maximumPoolSize, //最大執行緒數量 long keepAliveTime, //超出核心執行緒數量以外的執行緒空餘存活時間 TimeUnit unit, //存活時間的單位 BlockingQueue<Runnable> workQueue, //儲存待執行任務的佇列 ThreadFactory threadFactory, //建立新執行緒使用的工廠 RejectedExecutionHandler handler // 當任務無法執行時的處理器 ) { //省略程式碼 }
下面我們來詳細解析一下各個引數的含義：
1.corePoolSize （核心執行緒數量）
預設情況下，核心執行緒會一直處於存活狀態，即時它們是閒置的。如果將ThreadPoolExecutor的allowCoreThreadTimeOut屬性設定為true，那麼核心執行緒也會在超時之後結束，超時時間就是引數中設定的keepAlivTime的值。
值得注意的是，當執行緒數量小於核心執行緒數時，有任務要執行時，即使當前有空閒執行緒也會建立一個新的執行緒並且作為核心執行緒存在。
2.maximumPoolSize（最大執行緒數量）
包括核心執行緒數+核心之外的執行緒數量，如果任務佇列在滿的情況下，而且執行緒數小於最大執行緒數，此時會建立一個新的執行緒來執行任務。
3.keepAliveTime（非核心執行緒的存活時間）
意思是非核心執行緒在沒有任務執行的情況下的最大的存活時間，當然如果執行緒池設定 allowCoreThreadTimeOut(true)，核心執行緒也一樣會有這種存活時間限制。
4.unit （存活時間單位）
5.workQueue（任務佇列）
執行緒池中的任務佇列，我們提交給執行緒池的Runnable物件就儲存在這個佇列中。這是一個BlockQueue型別的結構，屬於阻塞佇列， 類似於生產者消費者模式，佇列中有任務的時候才能進行取出，當佇列滿的時候，新增也會被阻塞。
6.threadFactory（建立執行緒的工廠）
通過這個工廠類，我們可以給建立執行緒取名字或者優先順序之類的引數。
7.handler （任務飽和策略）
主要有四種策略：
CallerRunsPolicy：只要執行緒池沒有關閉，就直接用呼叫者所線上程來執行任務
AbortPolicy：直接丟擲RejectedExecutionException異常
DiscardPolicy：不通知地把任務拋棄了，不幹了
DiscardOldestPolicy：把佇列中呆了最久的那個任務拋棄了，然後再呼叫execute方法重試
workQueue（任務佇列）的分析：
執行緒池中使用的佇列是BlockQueue介面，常用的實現有以下幾種：
-
ArrayBlockingQueue：基於陣列，有界，先進先出原則，一般不用
-
LinkedBlockingQueue：基於連結串列，按先進先出排序，Executors.newFixedThreadPool() 使用的就是個佇列
-
SynchronousQueue：這個佇列不儲存任務，只負責傳遞，Executors.newCachedThreadPool就使用了這個佇列，當有任務到來時如果現有沒有空閒的執行緒，就會再建立一個執行緒來執行新任務。
-
PriorityBlockingQueue：具有優先順序的任務佇列
下面我們建立一個自己的執行緒池，程式碼如下：
public class MyThreadPool {
private final String TAG = this.getClass().getSimpleName();
private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2; // 核心執行緒數為 CPU數＊2
private static final int MAXIMUM_POOL_SIZE = 64; // 執行緒佇列最大執行緒數
private static final int KEEP_ALIVE_TIME = 1; // 保持存活時間 1秒
private final BlockingQueue<Runnable> mWorkQueue = new LinkedBlockingQueue<>(128);
private final ThreadFactory DEFAULT_THREAD_FACTORY = new ThreadFactory() {
private final AtomicInteger mCount = new AtomicInteger(1);
public Thread newThread(Runnable r) {
Thread thread = new Thread(r, TAG + " #" + mCount.getAndIncrement());
thread.setPriority(Thread.NORM_PRIORITY);
return thread;
}
};
private ThreadPoolExecutor mExecutor = new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_TIME,
TimeUnit.SECONDS, mWorkQueue, DEFAULT_THREAD_FACTORY,
new ThreadPoolExecutor.DiscardOldestPolicy());
private static volatile MyThreadPool mInstance = new MyThreadPool();
public static MyThreadPool getInstance() {
return mInstance;
}
//執行任務
public void execute(Runnable runnable) {
mExecutor.execute(runnable);
}
@Deprecated
public void shutdownNow() {
mExecutor.shutdownNow();
}
}可以看到，其實建立一個執行緒池很簡單。
JDK提供的常見執行緒池
JDK為我們提供的五種常見執行緒池的實現，可以用Executors類去建立。
1.newFixedThreadExecutor
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}可以看到核心執行緒數等於最大執行緒數，而且由構造方法傳參決定，佇列用LinkedBlockingQueue，佇列預設預設容量為Integer.MAX_VALUE，相當於沒有上限了。
這個執行緒池的執行規則為，當執行緒數小於核心執行緒數時，有新任務就會建立新執行緒去執行任務直到執行緒數量等於核心執行緒數，執行緒數等於核心執行緒數以後，就會把任務加入到佇列中，由執行緒去佇列中取任務進行執行操作。此種執行緒模式適合於負載量大的併發執行，當然要根據實際情況設定核心執行緒數。
2.newSingleThreadExecutor
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
可以看到，這種模式只有一個核心執行緒並且最大執行緒數也是一個，所有的任務都只有一個執行緒去執行，這個模式適用於序列執行任務，每個任務需要按順序執行並且不需要併發執行。
3.newCachedThreadPool
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}此種執行緒池沒有核心執行緒，最大執行緒數是Integer.MAX_VALUE，可以建立非常多的非核心執行緒，最大空閒時間為60秒，佇列用的SynchronousQueue。這中佇列不儲存任務，每當有任務到來的時候，如果沒有空閒執行緒就建立一個新的執行緒，而執行緒空閒60秒以後就會被結束。這種模式適用於需要併發執行大量輕任務。
4.newScheduledThreadPool
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
//核心執行緒數是固定的,非核心執行緒無限大,並且非核心執行緒數有10s的空閒存活時間
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE,
DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
new DelayedWorkQueue());
}
//super執行這裡
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}此執行緒池的核心數是固定的，而非核心執行緒沒有數量上限，非核心執行緒允許的最大閒置時間為10s。這種執行緒池適用於執行定時或週期性地執行任務。
其中DelayedWorkQueue是一個延時佇列，存入任務的時候會附帶一個延時值，代表這個任務要過多久才能取出。
兩種提交任務的方式：
1.execute（引數）：提交了任務沒有返回值
2.submit（引數）：提交了任務有一個返回值
我們來看看submit方法的原始碼：
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}其實submit裡面也是呼叫了execute方法，只是將task包裝成了RunnableFuture例項，然後返回這個Future物件，通過這個物件我們可以判斷任務是否執行成功。獲得執行結果呼叫Future.get()方法，這個方法被執行緒阻塞直到任務完成。
關閉執行緒池
執行緒池即時不執行任務也會佔用一些資源，所以不使用的時候最好關閉執行緒池。
關閉執行緒池的方法有兩種：
1.shutdown（）；
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess(); //獲取許可權
advanceRunState(SHUTDOWN); //修改執行狀態為SHUTDOWN
interruptIdleWorkers(); //遍歷停止未開啟的執行緒
onShutdown(); // 目前空實現
} finally {
mainLock.unlock();
}
tryTerminate();
}
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) { //遍歷所有執行緒
Thread t = w.thread;
//多了一個條件w.tryLock()，表示拿到鎖後就中斷，因為只有未開啟的執行緒才能拿到鎖
//正在執行的執行緒本來就已經上鎖了而且不是可重入鎖，所以會獲取鎖失敗
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
將執行緒設定為SHUTDOWN狀態，然後中斷尚未開始執行的執行緒，不能再新增執行緒了。
2.shutdownNow（）；
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP); //修改狀態
interruptWorkers(); //中斷所有執行緒
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
//中斷所有執行緒
private void interruptWorkers() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers)
w.interruptIfStarted();
} finally {
mainLock.unlock();
}
}設定為STOP狀態，嘗試中斷所有執行緒，不管有沒有在執行。
所以說，shutdown和shutdownNow的主要區別是前者中斷未執行的執行緒，後者中斷所有執行緒。
執行緒池使用的總結：
根據前面的介紹，我們小結一下JDK提供的幾種執行緒池的使用策略：
1.CachedThreadPool 用於併發執行大量短期的小任務，或者是負載較輕的伺服器
2.FixedThreadPool 用於負載比較重的伺服器，為了資源的合理利用，需要限制當前執行緒數量。
3.SingleThreadExecutor 用於序列執行任務的場景，每個任務必須按順序執行，不需要併發執行。
4.ScheduledThreadPoolExecutor 用於需要多個後臺執行緒執行週期任務，同時需要限制執行緒數量的場景。
當我們自定義執行緒池時，如果是CPU密集型的任務（需要進行大量計算等等），應該儘量配置少的執行緒數量，一般設定為CPU個數+1個執行緒，這樣可以避免太多執行緒爭搶資源的情況。
如果是IO密集型的任務（主要時間在IO工作，CPU空閒多一點），則應該多配置執行緒數，一般為CPU數的兩倍，這樣可以充分利用CPU功的能。
執行緒池的原始碼解析：
當我們用執行緒池執行任務的時候會呼叫execute方法或者submit方法，其實submit也是呼叫了execute方法，所以我們這裡先從execute方法的原始碼開始分析：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
//獲取當前執行緒池的生命週期狀態碼，其實裡面包含了執行緒池的生命週期和工作執行緒個數資訊
int c = ctl.get();
//判斷工作執行緒的數量是否小於核心執行緒數，如果小於則通過addWorker方法建立新的核心執行緒
//建立成功的話結束方法，失敗的繼續往下走
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//如果執行緒池為running狀態，則新增任務到佇列
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
//再次判斷執行緒狀態，如果不是running狀態，移除任務
if (! isRunning(recheck) && remove(command))
reject(command); //呼叫拒絕方法
//如果沒有工作執行緒，則單獨建立執行緒,而不指定任務
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//如果工作執行緒大於等於核心執行緒數，則建立一個非核心執行緒來執行任務，建立失敗的話呼叫拒絕方法
else if (!addWorker(command, false))
reject(command);
}從上面可以看到，其中addWorker方法是建立執行緒的核心方法，reject方法是處理建立失敗的回撥。
下面我們重點來看看addWorker方法，引數有兩個：
Runnable firstTask:
為傳遞進來需要執行的任務,也可以設定為null(在SHUTDOWN情況下,單純的建立執行緒來執行任務).
boolean core:
需要建立的執行緒是否為核心執行緒
private boolean addWorker(Runnable firstTask, boolean core) {
//for(;;)是死迴圈，retry主要用來跳出迴圈或者跳到retry所在地方重新執行
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c); //執行緒池生命週期
// Check if queue empty only if necessary.
//首先判斷如果為非running狀態，還有就是如果是SHUTDOWN狀態並且任務為null並且任務隊
//列不為空，這種情況還是要建立執行緒來加快執行任務的，所以也要排除這個情況，其他情況返
//回false新增失敗
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c); //工作執行緒數量
//工作執行緒數大於最大容量，或者作為核心執行緒新增的時候大於等於核心執行緒數，作為
//非核心執行緒新增時大於最大允許執行緒數，都會返回false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//設定工作執行緒數量增加1，如果成功則跳出迴圈，失敗的話重新迴圈
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
//下面是建立執行緒的過程，建立過程中需要加鎖保證安全，Worker是執行緒的包裝類
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask); //用Worker包裝任務
final Thread t = w.thread;
if (t != null) {
//加鎖
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int rs = runStateOf(ctl.get());
//可以看到，有兩種情況會建立執行緒，第一種是當前執行緒池為running狀態，第二是
//雖然為SHUTDOWN狀態但是task為null，此時也還要建立執行緒
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
//新增worker物件
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true; //設定為新增成功
}
} finally {
//釋放鎖
mainLock.unlock();
}
if (workerAdded) {
//執行緒設定為就緒狀態
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w); //處理建立失敗
}
return workerStarted;
}上面是建立執行緒的核心程式碼分析，註釋已經說明得比較清楚了。上面有一個比較重要的類是Worker，此類會對任務進行包裝，下面我們去看看這個Worker的核心原始碼：
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
*********省略程式碼************
final Thread thread;
Runnable firstTask;
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);
}
*******省略程式碼*******
}抽取一些關鍵程式碼，我們可以看到，worker類持有執行緒類thread和任務類runnable，而且worker本身也實現了Runnable介面，也是一個執行緒任務。再看構造方法裡面的this.thread = getThreadFactory().newThread(this)，可以看到thread的任務引數就是worker本身，如果呼叫thread的start方法最後運行了Worker的run（）方法，再去呼叫runWorker(this)方法。下面我們來看看這個方法的原始碼：
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
// 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 {
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(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}這個方法邏輯很簡單，就是通過while迴圈去執行task 的run方法，我們看while迴圈的條件有一個getTask方法，這是一個從任務佇列取任務的方法，然後把取出來的任務執行。我們來看看getTask方法原始碼：
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
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;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}getTask方法返回null的幾種情況：任務佇列為空並且是非running狀態、工作執行緒大於最大執行緒數、執行緒已經超時了。返回null的時候，runWorker方法就會跳出迴圈，結束任務。我們通過任務佇列workQueue去取任務，當然還有相關超時的情況，因為此佇列是阻塞佇列，所以當佇列為空的時候會阻塞在這裡，這也就是為什麼執行緒池中的執行緒在沒有執行任務的時候不會死亡的原因了。
以上就是對java執行緒池的核心原始碼的分析了，文章結束！