JAVA并發(fā)編程與高并發(fā)解決方案 - 并發(fā)編程 六
| 版本 | 作者 | 內容 |
|---|---|---|
| 2018.7.4 | chuIllusions | 線程池 |
相關文章
JAVA并發(fā)編程與高并發(fā)解決方案 - 并發(fā)編程 一 之 并發(fā)相關知識
JAVA并發(fā)編程與高并發(fā)解決方案 - 并發(fā)編程 二 之 線程安全性、安全發(fā)布對象
JAVA并發(fā)編程與高并發(fā)解決方案 - 并發(fā)編程 三 之 線程安全策略
JAVA并發(fā)編程與高并發(fā)解決方案 - 并發(fā)編程 四 之 J.U.C之AQS
JAVA并發(fā)編程與高并發(fā)解決方案 - 并發(fā)編程 五 之 J.U.C組件拓展
線程池
在前面使用的例子用，我們已經使用過線程池，基本上就是初始化線程池實例之后，把任務丟進去，等待調度執(zhí)行就可以了，使用起來非常簡單、方便。雖然使用很簡單，但線程池涉及到的知識點非常多。需要分析其實現。
JAVA中Thread這個類是線程類，在JAVA基礎時，對于線程的認識是基于此類，為什么不使用Thread直接執(zhí)行線程例子呢，而要使用線程池？可以試想，當并發(fā)數量很多，并且每個線程都是執(zhí)行一個時間很短的任務就結束了，這樣頻繁創(chuàng)建線程就會大大降低系統(tǒng)的效率，因為頻繁創(chuàng)建線程和銷毀線程需要時間。而線程池可以達到這樣的效果：線程可以復用，就是執(zhí)行完一個任務，并不被銷毀，而是可以繼續(xù)執(zhí)行其他的任務。
Thread的弊端：
- 每次
new Thread()新建對象，性能差； - 線程缺乏統(tǒng)一管理，可能無限制的新建線程，相互競爭，有可能占用過多系統(tǒng)資源導致死機或OOM；
- 缺少更多的功能，如更多執(zhí)行、定期執(zhí)行、線程中斷；
線程池的好處
- 重用存在的線程，減少對象創(chuàng)建、消亡的開銷，性能佳，降低資源消耗；
- 可有效控制最大并發(fā)線程數，提高系統(tǒng)資源利用率，同時可以避免過多資源競爭，避免阻塞，提高響應速度；
- 提供定時執(zhí)行、定期執(zhí)行、單線程、并發(fā)數控制等功能，以達到提高線程的可管理性。
阿里發(fā)布的 Java 開發(fā)手冊中強制線程池不允許使用 Executors 去創(chuàng)建，而是通過 ThreadPoolExecutor 的方式，這樣的處理方式讓寫的同學更加明確線程池的運行規(guī)則，規(guī)避資源耗盡的風險。
?? Executors利用工廠模式向我們提供了4種線程池實現方式，但是并不推薦使用，原因是使用Executors創(chuàng)建線程池不會傳入相關參數而使用默認值所以我們常常忽略了那些重要的參數（線程池大小、緩沖隊列的類型等），而且默認使用的參數會導致資源浪費，不可取。
ThreadPoolExecutor
Constructor And Parameters
java.uitl.concurrent.ThreadPoolExecutor 類是線程池中最核心的一個類，因此如果要透徹地了解Java中的線程池，必須先了解這個類，因此我們直接上源碼：
public class ThreadPoolExecutor extends AbstractExecutorService {
/** 構造函數 1 */
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {}
/** 構造函數 2 */
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory) {}
/** 構造函數 3 */
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {}
/** 構造函數 4 */
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {}
}
ThreadPoolExecutor 類繼承結構是： Executor(I) <- ExecutorService(I) <- AbstractExecutorService(C) <- TreadPoolExecutor
ThreadPoolExecutor類中提供了四個構造方法，在構造函數4中，參數最多，通過觀察其他3個構造函數，發(fā)現前面三個構造器都是調用的第四個構造器進行的初始化工作。
構造器中各個參數的含義：
-
corePoolSize： 核心池的大小，這個參數跟后面講述的線程池的實現原理有非常大的關系。在創(chuàng)建了線程池后，默認情況下，線程池中并沒有任何線程，而是等待有任務到來才創(chuàng)建線程去執(zhí)行任務，除非調用了預創(chuàng)建線程的方法，即在沒有任務到來之前就創(chuàng)建 corePoolSize 個線程或者 一個線程：
-
prestartCoreThread(): 預創(chuàng)建一個核心線程，使其閑置等待工作。 -
prestartAllCoreThreads(): 啟動所有核心線程，導致它們空閑地等待工作。
默認情況下，在創(chuàng)建了線程池后，線程池中的線程數為0，當有任務來之后，就會創(chuàng)建一個線程去執(zhí)行任務，當線程池中的線程數目達到corePoolSize后，就會把到達的任務放到緩存隊列當中；
-
maximumPoolSize：線程池最大線程數，這個參數也是一個非常重要的參數，它表示在線程池中最多能創(chuàng)建多少個線程；
keepAliveTime：表示線程沒有任務執(zhí)行時最多保持多久時間會終止。默認情況下，只有當線程池中的線程數大于
corePoolSize時，keepAliveTime才會起作用，直到線程池中的線程數不大于corePoolSize，即當線程池中的線程數大于corePoolSize時，如果一個線程空閑的時間達到keepAliveTime，則會終止，直到線程池中的線程數不超過corePoolSize。但是如果調用了allowCoreThreadTimeOut(boolean)方法，在線程池中的線程數不大于corePoolSize時，keepAliveTime參數也會起作用，直到線程池中的線程數為0；-
unit：參數
keepAliveTime的時間單位，有7種取值，在TimeUnit類中有7種靜態(tài)屬性：-
TimeUnit.DAYS: 以 天 為單位 ； -
TimeUnit.HOURS: 以 小時 為單位 ； -
TimeUnit.MINUTES: 以 分鐘 為單位 ； -
TimeUnit.SECONDS: 以 秒 為單位 ； -
TimeUnit.MILLISECONDS: 以 毫秒 為單位 ； -
TimeUnit.MICROSECONDS: 以 微秒 為單位 ； -
TimeUnit.NANOSECONDS: 以 納秒 為單位 ；
-
-
workQueue： 一個阻塞隊列，用來存儲等待執(zhí)行的任務，這個參數的選擇也很重要，會對線程池的運行過程產生重大影響，一般來說，這里的阻塞隊列有以下幾種選擇：
-
ArrayBlockingQueue：基于數組的先進先出隊列，此隊列創(chuàng)建時必須指定大小； -
LinkedBlockingQueue：基于鏈表的先進先出隊列，如果創(chuàng)建時沒有指定此隊列大小，則默認為Integer.MAX_VALUE； -
SynchronousQueue：這個隊列比較特殊，它不會保存提交的任務，而是將直接新建一個線程來執(zhí)行新來的任務。
一般使用
LinkedBlockingQueue和SynchronousQueue -
threadFactory：線程工廠，主要用來創(chuàng)建線程。 線程池最主要的一項工作，就是在滿足某些條件的情況下創(chuàng)建線程。而在
ThreadPoolExecutor線程池中，創(chuàng)建線程的工作交給ThreadFactory來完成。要使用線程池，就必須要指定ThreadFactory。 如果我們使用的構造函數時并沒有指定使用的ThreadFactory，這個時候ThreadPoolExecutor會使用一個默認的ThreadFactory：DefaultThreadFactory（這個類在Executors工具類中）;-
handler：在
ThreadPoolExecutor線程池中還有一個重要的接口：RejectedExecutionHandler。當提交給線程池的某一個新任務無法直接被線程池中“核心線程”直接處理，又無法加入等待隊列，也無法創(chuàng)建新的線程執(zhí)行；又或者線程池已經調用shutdown()方法停止了工作；又或者線程池不是處于正常的工作狀態(tài)；這時候ThreadPoolExecutor線程池會拒絕處理這個任務，觸發(fā)創(chuàng)建ThreadPoolExecutor線程池時定義的RejectedExecutionHandler接口的實現，表示當拒絕處理任務時的策略，有以下四種取值，四種值都為其靜態(tài)內部類：
-
ThreadPoolExecutor.AbortPolicy：丟棄任務并拋出RejectedExecutionException異常。 -
ThreadPoolExecutor.DiscardPolicy：也是丟棄任務，但是不拋出異常。 -
ThreadPoolExecutor.DiscardOldestPolicy：丟棄隊列最前面的任務，然后重新嘗試執(zhí)行新提交的任務。
-
Operating principle
介紹完上面的構造函數以及其參數之后，介紹一下ThreadPoolExecutor的運行原理，在網上瀏覽到一篇相關文章，因此將它直接引用過來
深入理解java線程池—ThreadPoolExecutor，以下內容引用此文章，并且增加了自己的一點理解
ThreadPoolExecutor.execute()
向線程池中提交一個不需要返回結果的任務
public void execute(Runnable command) {
//任務為null,則拋出異常
if (command == null)
throw new NullPointerException();
//取出記錄著runState和workerCount 的 ctl的當前值
int c = ctl.get();
/**
* 1.第一步：
* 通過workerCountOf方法從ctl所表示的int值中提取出低29位的值,也就是當前活動的線程數。
* 如果當前活動的線程數少于corePoolSize,則通過addWorker(command, true)新建一個線程,并將任務(command)添加到該線程中
*/
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
/**
* 2.第二步：
* 2.1 isRunning(c) 當前線程池是否處于運行狀態(tài)。源代碼是通過判斷c < SHUTDOWN 來確定返回值。由于RUNNING才會接收新任務，且只有這個值-1才小于SHUTDOWN
* 2.2 workQueue.offer(command) 任務添加到緩沖隊列
*/
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
/**
* 如果 線程池已經處于非運行狀態(tài),則從緩沖隊列中移除任務然后采用線程池指定的策略拒絕任務
*/
if (! isRunning(recheck) && remove(command))
reject(command);
/**
* 如果 線程池中任務數量為0,則通過addWorker(null, false)嘗試新建一個線程,新建線程對應的任務為null
*/
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
/**
* 3.第三步，也就是以上兩個步驟都不滿足：
* 3.1 當前線程池并不處于Running狀態(tài)
* 3.2 當前線程池處于Running狀態(tài),但是緩沖隊列已經滿了
*/
else if (!addWorker(command, false))
reject(command);
}
從上面execute()方法中，出現了ctl，跟蹤源碼分析起作用：
//將整型的32位分為高3位和低29位，高3位表示線程池的狀態(tài),低29位表示活動的線程數
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//獲得高三位
private static final int COUNT_BITS = Integer.SIZE - 3;
//29位能表示的最大二進制整數，也就是活動線程數
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
//運行狀態(tài)是存儲在高三位中
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;
線程池是通過Integer類型的高3位表述當前線程池的狀態(tài)RUNNING，SHUTDOWN，STOP，TIDYING，TERMINATED 。低29位表示當前線程的運行任務數量。然后通過位運算來計算運行狀態(tài)和任務的數量。
線程池在執(zhí)行execute(Runnable)，執(zhí)行流程如下（對應圖中的流程）
- 如果當前運行的線程少于
corePoolSize，則創(chuàng)建新線程來執(zhí)行任務（需要獲得全局鎖） - 如果運行的線程等于或多于
corePoolSize,則將任務加入BlockingQueue - 如果無法將任務加入
BlockingQueue(隊列已滿)，則創(chuàng)建新的線程來處理任務（需要獲得全局鎖） - 如果創(chuàng)建新線程將使當前運行的線程超出
maxiumPoolSize，任務將被拒絕，并調用RejectedExecutionHandler.rejectedExecution()方法。
線程池采取上述的流程進行設計是為了減少獲取全局鎖的次數。在線程池完成預熱（當前運行的線程數大于或等于corePoolSize）之后，幾乎所有的execute方法調用都執(zhí)行步驟2。
ThreadPoolExecutor.addWorker()
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get(); //獲取運行狀態(tài)和工作數量
int rs = runStateOf(c); //獲取當前線程池運行的狀態(tài)
// Check if queue empty only if necessary.
//條件代表著以下幾個場景，直接返回false說明當前工作線程創(chuàng)建失敗
//1.rs>SHUTDOWN 此時不再接收新任務，且所有的任務已經執(zhí)行完畢
//2.rs=SHUTDOWN 此時不再接收新任務，但是會執(zhí)行隊列中的任務
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
//先判斷當前活動的線程數是否大于最大值，如果超過了就直接返回false說明線程創(chuàng)建失敗
//如果沒有超過再根據core的值再進行以下判斷
//1\. core為true，則判斷當前活動的線程數是否大于corePoolSize
//2\. core為false，則判斷當前活動線程數是否大于maximumPoolSize
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//比較當前值是否和c相同，如果相同，則改為c+1，并且跳出大循環(huán)，直接執(zhí)行Worker進行線程創(chuàng)建
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
//檢查下當前線程池的狀態(tài)是否已經發(fā)生改變
//如果已經改變了，則進行外層retry大循環(huán)，否則只進行內層的循環(huán)
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的也是Runnable的實現類
w = new Worker(firstTask);
//因為不可以直接在Worker的構造方法中進行線程創(chuàng)建
//所以要把它的引用賦給t方便后面進行線程創(chuàng)建
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());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
workers.add(w);//將創(chuàng)建的線程添加到workers容器中
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
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;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
}
Worker在ThreadPoolExecutor為一個內部類實現了Runnable接口。只有一個構造方法，在上面的addWorker()中final Thread t = w.thread;知道其實是獲取了線程的對象，因為在構造方法中，線程的引用即是它自己。
因此在調用t.start()執(zhí)行的是（Worker類中的方法）：
/** Delegates main run loop to outer runWorker */
public void run() {
//這里執(zhí)行的是ThreadPoolExecutor中的runWorker
runWorker(this);
}
ThreadPoolExecutor.runWorker()
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;//獲取Worker中的任務
w.firstTask = null; //將Woeker中的任務置空
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//如果當前任務為空 那么就從getTask中獲得任務
/**
* 如果task不為空，執(zhí)行完task后則將task置空
* 繼續(xù)進入循環(huán)，則從getTask中獲取任務
*/
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 {
//任務執(zhí)行前調用的方法
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);
}
}
從上面可以簡單理解，就是執(zhí)行任務，只是執(zhí)行任務需要進行處理，包括獲得任務、任務開始前處理、任務執(zhí)行、任務執(zhí)行后處理。但是，關鍵代碼還是里面所調用的一個方法getTask() 。
beforeExecute(Thread t, Runnable r)與afterExecute(Runnable r, Throwable t)并未在類中有處理業(yè)務的邏輯，即可以通過繼承線程池的方式來重寫這兩個方法，這樣就能夠對任務的執(zhí)行進行監(jiān)控。
這里我有兩個疑問？
-
怎么退出這個While循環(huán)，也就是進入到
processWorkerExit()- 從While循環(huán)體中可以知道，當線程運行時出現異常，那么都會退出循環(huán)，進入到
processWorkerExit() - 從
getTask()獲得結果為null，則也會進到processWorkerExit()
- 從While循環(huán)體中可以知道，當線程運行時出現異常，那么都會退出循環(huán)，進入到
getTask()方法為什么是最關鍵的？分析其執(zhí)行代碼
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
//死循環(huán)
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?
//如果設置了allowCoreThreadTimeOut(true)
//或者當前運行的任務數大于設置的核心線程數
// timed = true
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
/** ------------------------以上的操作跟之前類似----------------------- */
/** ------------------------關鍵在于下面的代碼------------------------- */
/** ------------------------從阻塞隊列中獲取任務----------------------- */
try {
Runnable r = timed ?
//對于阻塞隊列，poll(long timeout, TimeUnit unit) 將會在規(guī)定的時間內去任務
//如果沒取到就返回null
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
//take會一直阻塞，等待任務的添加
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
到此，終于發(fā)現為什么線程池能夠保證一直等待任務而不被銷毀，其實就是進入了阻塞狀態(tài)。
ThreadPoolExecutor.processWorkerExit()
/**
* @param completedAbruptly 工作線程是否死與執(zhí)行任務出現的異常
*/
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) //如果突然被打斷，工作線程數不會被減少
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
tryTerminate();
int c = ctl.get();
//判斷運行狀態(tài)是否在STOP之前
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {//正常退出，也就是task == null
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
//新增一個工作線程，代替原來的工作線程
addWorker(null, false);
}
}
AbstractExecutorService.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(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;
}
在ThreadPoolExecutor中并未發(fā)現submit()，因此從父類，即抽象類AbstractExecutorService中找到submit()的方法實現，從方法實現中，可知：
-
submit()接收任務參數，并將參數封裝為FutureTask任務類 - 將封裝好的
FutureTask提交到execute()中
結論：submit()真正實現的任務處理流程跟execute()一樣，也可以說submit()就是調用了execute()
從上面的流程圖可以知道，向線程池提交一個任務后，共經歷以下流程：
- 提交任務到線程池；
- 線程池判斷核心線程池里是的線程是否都在執(zhí)行任務，如果不是，則創(chuàng)建一個新的工作線程來執(zhí)行任務。如果核心線程池里的線程都在執(zhí)行任務，則進入下一個流程。
- 線程池判斷工作隊列是否已滿。如果工作隊列沒有滿，則將新提交的任務儲存在這個工作隊列里。如果工作隊列滿了，則進入下一個流程。
- 線程池判斷其內部線程是否都處于工作狀態(tài)。如果沒有，則創(chuàng)建一個新的工作線程來執(zhí)行任務。如果已滿了，則交給飽和策略來處理這個任務。
ThreadPoolExecutor.shutdown()
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();//檢查終止線程池的線程是否有權限。
advanceRunState(SHUTDOWN);// 設置線程池的狀態(tài)為關閉狀態(tài)。
interruptIdleWorkers();// 中斷線程池中空閑的線程
onShutdown(); // 鉤子函數，在ThreadPoolExecutor中沒有任何動作
} finally {
mainLock.unlock();
}
tryTerminate();// 嘗試終止線程池
}
