java多線程系列：ThreadPoolExecutor源碼分析
前言
這篇主要講述ThreadPoolExecutor的源碼分析，貫穿類的建立、任務的添加到線程池的關閉整個流程，讓你知其然因此然。但願你能夠經過本篇博文知道ThreadPoolExecutor是怎麼添加任務、執行任務的，以及延伸的知識點。那麼先來看看ThreadPoolExecutor的繼承關係吧。java
繼承關係
Executor接口
public interface Executor {
void execute(Runnable command);
}
Executor接口只有一個方法execute,傳入線程任務參數git
ExecutorService接口
public interface ExecutorService extends Executor {
void shutdown();
List<Runnable> shutdownNow();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException;
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException;
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
ExecutorService接口繼承Executor接口，並增長了submit、shutdown、invokeAll等等一系列方法。github
AbstractExecutorService抽象類
public abstract class AbstractExecutorService implements ExecutorService {
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);
}
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;
}
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {...}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {... }
public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {...}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {...}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {...}
}
AbstractExecutorService抽象類實現ExecutorService接口，而且提供了一些方法的默認實現，例如submit方法、invokeAny方法、invokeAll方法。數據庫
像execute方法、線程池的關閉方法（shutdown、shutdownNow等等）就沒有提供默認的實現。編程
ThreadPoolExecutor
先介紹下ThreadPoolExecutor線程池的狀態吧安全
線程池狀態
int 是4個字節，也就是32位（注：一個字節等於8位）多線程
//記錄線程池狀態和線程數量（總共32位，前三位表示線程池狀態，後29位表示線程數量）
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//線程數量統計位數29 Integer.SIZE=32
private static final int COUNT_BITS = Integer.SIZE - 3;
//容量 000 11111111111111111111111111111
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
//運行中 111 00000000000000000000000000000
private static final int RUNNING = -1 << COUNT_BITS;
//關閉 000 00000000000000000000000000000
private static final int SHUTDOWN = 0 << COUNT_BITS;
//中止 001 00000000000000000000000000000
private static final int STOP = 1 << COUNT_BITS;
//整理 010 00000000000000000000000000000
private static final int TIDYING = 2 << COUNT_BITS;
//終止 011 00000000000000000000000000000
private static final int TERMINATED = 3 << COUNT_BITS;
//獲取運行狀態（獲取前3位）
private static int runStateOf(int c) { return c & ~CAPACITY; }
//獲取線程個數（獲取後29位）
private static int workerCountOf(int c) { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }
- RUNNING：接受新任務而且處理阻塞隊列裏的任務
- SHUTDOWN：拒絕新任務可是處理阻塞隊列裏的任務
- STOP：拒絕新任務而且拋棄阻塞隊列裏的任務同時會中斷正在處理的任務
- TIDYING：全部任務都執行完（包含阻塞隊列裏面任務）當前線程池活動線程爲0，將要調用terminated方法
- TERMINATED：終止狀態。terminated方法調用完成之後的狀態
線程池狀態轉換併發
RUNNING -> SHUTDOWN 顯式調用shutdown()方法, 或者隱式調用了finalize()方法 (RUNNING or SHUTDOWN) -> STOP 顯式調用shutdownNow()方法 SHUTDOWN -> TIDYING 當線程池和任務隊列都爲空的時候 STOP -> TIDYING 當線程池爲空的時候 TIDYING -> TERMINATED 當 terminated() hook 方法執行完成時候
構造函數
有四個構造函數，其餘三個都是調用下面代碼中的這個構造函數函數
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
}
參數介紹源碼分析
| 參數 | 類型 | 含義 |
|---|---|---|
| corePoolSize | int | 核心線程數 |
| maximumPoolSize | int | 最大線程數 |
| keepAliveTime | long | 存活時間 |
| unit | TimeUnit | 時間單位 |
| workQueue | BlockingQueue
|
存放線程的隊列 |
| threadFactory | ThreadFactory | 建立線程的工廠 |
| handler | RejectedExecutionHandler | 多餘的的線程處理器（拒絕策略） |
提交任務
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;
}
流程步驟以下
- 調用submit方法，傳入Runnable或者Callable對象
- 判斷傳入的對象是否爲null，爲null則拋出異常，不爲null繼續流程
- 將傳入的對象轉換爲RunnableFuture對象
- 執行execute方法，傳入RunnableFuture對象
- 返回RunnableFuture對象
流程圖以下
execute
public void execute(Runnable command) {
//傳進來的線程爲null，則拋出空指針異常
if (command == null)
throw new NullPointerException();
//獲取當前線程池的狀態+線程個數變量
int c = ctl.get();
/**
* 3個步驟
*/
//1.判斷當前線程池線程個數是否小於corePoolSize,小於則調用addWorker方法建立新線程運行,且傳進來的Runnable當作第一個任務執行。
//若是調用addWorker方法返回false，則直接返回
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//2.若是線程池處於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);
}
//3.新增線程，新增失敗則執行拒絕策略
else if (!addWorker(command, false))
reject(command);
}
其實從上面代碼註釋中能夠看出就三個判斷，
- 核心線程數是否已滿
- 隊列是否已滿
- 線程池是否已滿
而後根據這三個條件進行不一樣的操做，下圖是Java併發編程的藝術書中的線程池的主要處理流程，或許會比較容易理解些
下面是整個流程的詳細步驟
- 調用execute方法，傳入Runable對象
- 判斷傳入的對象是否爲null，爲null則拋出異常，不爲null繼續流程
- 獲取當前線程池的狀態和線程個數變量
- 判斷當前線程數是否小於核心線程數，是走流程5，不然走流程6
- 添加線程數，添加成功則結束，失敗則從新獲取當前線程池的狀態和線程個數變量,
- 判斷線程池是否處於RUNNING狀態，是則添加任務到阻塞隊列，不然走流程10，添加任務成功則繼續流程7
- 從新獲取當前線程池的狀態和線程個數變量
- 從新檢查線程池狀態，不是運行狀態則移除以前添加的任務，有一個false走流程9，都爲true則走流程11
- 檢查線程池線程數量是否爲0，不然結束流程，是調用addWorker(null, false)，而後結束
- 調用!addWorker(command, false)，爲true走流程11，false則結束
- 調用拒絕策略reject(command)，結束
可能看上面會有點繞，不清楚的能夠看下面的流程圖
addWorker
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 檢查當前線程池狀態是不是SHUTDOWN、STOP、TIDYING或者TERMINATED
// 且！（當前狀態爲SHUTDOWN、且傳入的任務爲null，且隊列不爲null）
// 條件都成立則返回false
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
//循環
for (;;) {
int wc = workerCountOf(c);
//若是當前的線程數量超過最大容量或者大於（根據傳入的core決定是核心線程數仍是最大線程數）核心線程數 || 最大線程數，則返回false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//CAS增長c，成功則跳出retry
if (compareAndIncrementWorkerCount(c))
break retry;
//CAS失敗執行下面方法，查看當前線程數是否變化，變化則繼續retry循環，沒變化則繼續內部循環
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
}
}
//CAS成功
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//新建一個線程
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
//加鎖
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//從新檢查線程池狀態
//避免ThreadFactory退出故障或者在鎖獲取前線程池被關閉
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // 先檢查線程是不是可啓動的
throw new IllegalThreadStateException();
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
//判斷worker是否添加成功，成功則啓動線程，而後將workerStarted設置爲true
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
//判斷線程有沒有啓動成功，沒有則調用addWorkerFailed方法
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
這裏能夠將addWorker分爲兩部分，第一部分增長線程池個數，第二部分是將任務添加到workder裏面並執行。
第一部分主要是兩個循環，外層循環主要是判斷線程池狀態，下面描述來自Java中線程池ThreadPoolExecutor原理探究
rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())展開！運算後等價於
s >= SHUTDOWN && (rs != SHUTDOWN || firstTask != null || workQueue.isEmpty())也就是說下面幾種狀況下會返回false：
- 當前線程池狀態爲STOP，TIDYING，TERMINATED
- 當前線程池狀態爲SHUTDOWN而且已經有了第一個任務
- 當前線程池狀態爲SHUTDOWN而且任務隊列爲空
內層循環做用是使用cas增長線程個數，若是線程個數超限則返回false，否者進行cas，cas成功則退出雙循環，否者cas失敗了，要看當前線程池的狀態是否變化了，若是變了，則從新進入外層循環從新獲取線程池狀態，否者進入內層循環繼續進行cas嘗試。
到了第二部分說明CAS成功了，也就是說線程個數加一了，可是如今任務還沒開始執行，這裏使用全局的獨佔鎖來控制workers裏面添加任務，其實也可使用併發安全的set，可是性能沒有獨佔鎖好（這個從註釋中知道的）。這裏須要注意的是要在獲取鎖後從新檢查線程池的狀態，這是由於其餘線程可可能在本方法獲取鎖前改變了線程池的狀態，好比調用了shutdown方法。添加成功則啓動任務執行。
因此這裏也將流程圖分爲兩部分來描述
第一部分流程圖
第二部分流程圖
Worker對象
Worker是定義在ThreadPoolExecutor中的finnal類，其中繼承了AbstractQueuedSynchronizer類和實現Runnable接口，其中的run方法以下
public void run() {
runWorker(this);
}
線程啓動時調用了runWorker方法，關於類的其餘方面這裏就不在敘述。
runWorker
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock();
boolean completedAbruptly = true;
try {
//循環獲取任務
while (task != null || (task = getTask()) != null) {
w.lock();
// 當線程池是處於STOP狀態或者TIDYING、TERMINATED狀態時，設置當前線程處於中斷狀態
// 若是不是，當前線程就處於RUNNING或者SHUTDOWN狀態，確保當前線程不處於中斷狀態
// 從新檢查當前線程池的狀態是否大於等於STOP狀態
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;
//統計當前worker完成了多少個任務
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
//整個線程結束時調用，線程退出操做。統計整個線程池完成的任務個數之類的工做
processWorkerExit(w, completedAbruptly);
}
}
getTask
getTask方法的主要做用其實從方法名就能夠看出來了，就是獲取任務
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
//循環
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
//線程線程池狀態和隊列是否爲空
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
//線程數量
int wc = workerCountOf(c);
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
//（當前線程數是否大於最大線程數或者）
//且（線程數大於1或者任務隊列爲空）
//這裏有個問題(timed && timedOut)timedOut = false，好像(timed && timedOut)一直都是false吧
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;
}
}
}
關閉線程池
shutdown
當調用shutdown方法時，線程池將不會再接收新的任務，而後將先前放在隊列中的任務執行完成。
下面是shutdown方法的源碼
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(SHUTDOWN);
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}
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;
}
shutdown和shutdownNow區別
shutdown和shutdownNow這兩個方法的做用都是關閉線程池，流程大體相同，只有幾個步驟不一樣，以下
- 加鎖
- 檢查關閉權限
- CAS改變線程池狀態
- 設置中斷標誌(線程池不在接收任務，隊列任務會完成)/中斷當前執行的線程
- 調用onShutdown方法（給子類提供的方法）/獲取隊列中的任務
- 解鎖
- 嘗試將線程池狀態變成終止狀態TERMINATED
- 結束/返回隊列中的任務
總結
線程池能夠給咱們多線程編碼上提供極大便利，就好像數據庫鏈接池同樣，減小了線程的開銷，提供了線程的複用。並且ThreadPoolExecutor也提供了一些未實現的方法，供咱們來使用，像beforeExecute、afterExecute等方法，咱們能夠經過這些方法來對線程進行進一步的管理和統計。
在使用線程池上好須要注意，提交的線程任務能夠分爲CPU 密集型任務和IO 密集型任務，而後根據任務的不一樣進行分配不一樣的線程數量。
- CPU密集型任務：
- 應當分配較少的線程，好比
CPU個數至關的大小
- 應當分配較少的線程，好比
- IO 密集型任務：
- 因爲線程並非一直在運行，因此能夠儘量的多配置線程，好比 CPU 個數 * 2
- 混合型任務：
- 能夠將其拆分爲
CPU密集型任務以及IO密集型任務，這樣來分別配置。
- 能夠將其拆分爲
好了，這篇博文到這裏就結束了，文中可能會有些紕漏，歡迎留言指正。
若是本文對你有所幫助，給個star唄，謝謝。本文GitHub地址：點這裏點這裏
參考資料
- 併發編程網-Java中線程池ThreadPoolExecutor原理探究
- Java併發編程的藝術
相關文章
- 1. java多線程系列：ThreadPoolExecutor
- 2. Java線程池：ThreadPoolExecutor源碼分析
- 3. Java線程池ThreadPoolExecutor源碼分析
- 4. Java線程池之ThreadPoolExecutor源碼分析
- 5. Java ThreadPoolExecutor 線程池源碼分析
- 6. java ThreadPoolExecutor線程池源碼分析
- 7. Java 多線程線程池-ThreadPoolExecutor分析
- 8. Java線程池ThreadPoolExecutor類源碼分析 java線程池ThreadPoolExecutor類使用詳解
- 9. Java源碼解析 - ThreadPoolExecutor 線程池
- 10. java線程池——ThreadPoolExecutor源碼解析
- 更多相關文章...
- • C# 多線程 - C#教程
- • Hibernate多對多映射關係 - Hibernate教程
- • Java Agent入門實戰（二）-Instrumentation源碼概述
- • Git五分鐘教程
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. java多線程系列：ThreadPoolExecutor
- 2. Java線程池：ThreadPoolExecutor源碼分析
- 3. Java線程池ThreadPoolExecutor源碼分析
- 4. Java線程池之ThreadPoolExecutor源碼分析
- 5. Java ThreadPoolExecutor 線程池源碼分析
- 6. java ThreadPoolExecutor線程池源碼分析
- 7. Java 多線程線程池-ThreadPoolExecutor分析
- 8. Java線程池ThreadPoolExecutor類源碼分析 java線程池ThreadPoolExecutor類使用詳解
- 9. Java源碼解析 - ThreadPoolExecutor 線程池
- 10. java線程池——ThreadPoolExecutor源碼解析