【高併發】深度解析執行緒池中那些重要的頂層介面和抽象類
大家好，我是冰河~~
在上一篇《【高併發】不得不說的執行緒池與ThreadPoolExecutor類淺析》一文中，從整體上介紹了Java的執行緒池。如果細細品味執行緒池的底層原始碼實現，你會發現整個執行緒池體系的設計是非常優雅的！這些程式碼的設計值得我們去細細品味和研究，從中學習優雅程式碼的設計規範，形成自己的設計思想，為我所用！哈哈，說多了，接下來，我們就來看看執行緒池中那些非常重要的介面和抽象類，深度分析下執行緒池中是如何將抽象這一思想運用的淋漓盡致的！
通過對執行緒池中介面和抽象類的分析，你會發現，整個執行緒池設計的是如此的優雅和強大，從執行緒池的程式碼設計中，我們學到的不只是程式碼而已！！
題外話：膜拜Java大神Doug Lea，Java中的併發包正是這位老爺子寫的，他是這個世界上對Java影響力最大的一個人。
一、介面和抽象類總覽
說起執行緒池中提供的重要的介面和抽象類，基本上就是如下圖所示的介面和類。
介面與類的簡單說明：
- Executor介面：這個介面也是整個執行緒池中最頂層的介面，提供了一個無返回值的提交任務的方法。
- ExecutorService介面：派生自Executor介面，擴充套件了很過功能，例如關閉執行緒池，提交任務並返回結果資料、喚醒執行緒池中的任務等。
- AbstractExecutorService抽象類：派生自ExecutorService介面，實現了幾個非常實現的方法，供子類進行呼叫。
- ScheduledExecutorService定時任務介面，派生自ExecutorService介面，擁有ExecutorService介面定義的全部方法，並擴充套件了定時任務相關的方法。
接下來，我們就分別從原始碼角度來看下這些介面和抽象類從頂層設計上提供了哪些功能。
二、Executor介面
Executor介面的原始碼如下所示。
public interface Executor {
//提交執行任務，引數為Runnable介面物件，無返回值
void execute(Runnable command);
}
從原始碼可以看出，Executor介面非常簡單，只提供了一個無返回值的提交任務的execute(Runnable)方法。
由於這個介面過於簡單，我們無法得知執行緒池的執行結果資料，如果我們不再使用執行緒池，也無法通過Executor介面來關閉執行緒池。此時，我們就需要ExecutorService介面的支援了。
三、ExecutorService介面
ExecutorService介面是非定時任務類執行緒池的核心介面，通過ExecutorService介面能夠向執行緒池中提交任務（支援有返回結果和無返回結果兩種方式）、關閉執行緒池、喚醒執行緒池中的任務等。ExecutorService介面的原始碼如下所示。
package java.util.concurrent;
import java.util.List;
import java.util.Collection;
public interface ExecutorService extends Executor {
//關閉執行緒池，執行緒池中不再接受新提交的任務，但是之前提交的任務繼續執行，直到完成
void shutdown();
//關閉執行緒池，執行緒池中不再接受新提交的任務，會嘗試停止執行緒池中正在執行的任務。
List<Runnable> shutdownNow();
//判斷執行緒池是否已經關閉
boolean isShutdown();
//判斷執行緒池中的所有任務是否結束，只有在呼叫shutdown或者shutdownNow方法之後呼叫此方法才會返回true。
boolean isTerminated();
//等待執行緒池中的所有任務執行結束，並設定超時時間
boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException;
//提交一個Callable介面型別的任務，返回一個Future型別的結果
<T> Future<T> submit(Callable<T> task);
//提交一個Callable介面型別的任務，並且給定一個泛型型別的接收結果資料引數，返回一個Future型別的結果
<T> Future<T> submit(Runnable task, T result);
//提交一個Runnable介面型別的任務，返回一個Future型別的結果
Future<?> submit(Runnable task);
//批量提交任務並獲得他們的future，Task列表與Future列表一一對應
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException;
//批量提交任務並獲得他們的future，並限定處理所有任務的時間
<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介面中每個方法的含義，直接上述介面原始碼中的註釋即可，這些介面方法都比較簡單，我就不一一重複列舉描述了。這個介面也是我們在使用非定時任務類的執行緒池中最常使用的介面。
四、AbstractExecutorService抽象類
AbstractExecutorService類是一個抽象類，派生自ExecutorService介面，在其基礎上實現了幾個比較實用的方法，提供給子類進行呼叫。我們還是來看下AbstractExecutorService類的原始碼。
注意：大家可以到java.util.concurrent包下檢視完整的AbstractExecutorService類的原始碼，這裡，我將AbstractExecutorService原始碼進行拆解，詳解每個方法的作用。
- newTaskFor方法
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);
}
RunnableFuture類用於獲取執行結果，在實際使用時，我們經常使用的是它的子類FutureTask，newTaskFor方法的作用就是將任務封裝成FutureTask物件，後續將FutureTask物件提交到執行緒池。
- doInvokeAny方法
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {
//提交的任務為空，丟擲空指標異常
if (tasks == null)
throw new NullPointerException();
//記錄待執行的任務的剩餘數量
int ntasks = tasks.size();
//任務集合中的資料為空，丟擲非法引數異常
if (ntasks == 0)
throw new IllegalArgumentException();
ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks);
//以當前例項物件作為引數構建ExecutorCompletionService物件
// ExecutorCompletionService負責執行任務，後面呼叫用poll返回第一個執行結果
ExecutorCompletionService<T> ecs =
new ExecutorCompletionService<T>(this);
try {
// 記錄可能丟擲的執行異常
ExecutionException ee = null;
// 初始化超時時間
final long deadline = timed ? System.nanoTime() + nanos : 0L;
Iterator<? extends Callable<T>> it = tasks.iterator();
//提交任務，並將返回的結果資料新增到futures集合中
//提交一個任務主要是確保在進入迴圈之前開始一個任務
futures.add(ecs.submit(it.next()));
--ntasks;
//記錄正在執行的任務數量
int active = 1;
for (;;) {
//從完成任務的BlockingQueue佇列中獲取並移除下一個將要完成的任務的結果。
//如果BlockingQueue佇列中中的資料為空，則返回null
//這裡的poll()方法是非阻塞方法
Future<T> f = ecs.poll();
//獲取的結果為空
if (f == null) {
//集合中仍有未執行的任務數量
if (ntasks > 0) {
//未執行的任務數量減1
--ntasks;
//提交完成並將結果新增到futures集合中
futures.add(ecs.submit(it.next()));
//正在執行的任務數量加•1
++active;
}
//所有任務執行完成，並且返回了結果資料，則退出迴圈
//之所以處理active為0的情況，是因為poll()方法是非阻塞方法，可能導致未返回結果時active為0
else if (active == 0)
break;
//如果timed為true，則執行獲取結果資料時設定超時時間，也就是超時獲取結果表示
else if (timed) {
f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
if (f == null)
throw new TimeoutException();
nanos = deadline - System.nanoTime();
}
//沒有設定超時，並且所有任務都被提交了，則一直阻塞，直到返回一個執行結果
else
f = ecs.take();
}
//獲取到執行結果，則將正在執行的任務減1，從Future中獲取結果並返回
if (f != null) {
--active;
try {
return f.get();
} catch (ExecutionException eex) {
ee = eex;
} catch (RuntimeException rex) {
ee = new ExecutionException(rex);
}
}
}
if (ee == null)
ee = new ExecutionException();
throw ee;
} finally {
//如果從所有執行的任務中獲取到一個結果資料，則取消所有執行的任務，不再向下執行
for (int i = 0, size = futures.size(); i < size; i++)
futures.get(i).cancel(true);
}
}
這個方法是批量執行執行緒池的任務，最終返回一個結果資料的核心方法，通過原始碼的分析，我們可以發現，這個方法只要獲取到一個結果資料，就會取消執行緒池中所有執行的任務，並將結果資料返回。這就好比是很多要進入一個居民小區一樣，只要有一個人有門禁卡，門衛就不再檢查其他人是否有門禁卡，直接放行。
在上述程式碼中，我們看到提交任務使用的ExecutorCompletionService物件的submit方法，我們再來看下ExecutorCompletionService類中的submit方法，如下所示。
public Future<V> submit(Callable<V> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<V> f = newTaskFor(task);
executor.execute(new QueueingFuture(f));
return f;
}
public Future<V> submit(Runnable task, V result) {
if (task == null) throw new NullPointerException();
RunnableFuture<V> f = newTaskFor(task, result);
executor.execute(new QueueingFuture(f));
return f;
}
可以看到，ExecutorCompletionService類中的submit方法本質上呼叫的還是Executor介面的execute方法。
- invokeAny方法
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {
try {
return doInvokeAny(tasks, false, 0);
} catch (TimeoutException cannotHappen) {
assert false;
return null;
}
}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
return doInvokeAny(tasks, true, unit.toNanos(timeout));
}
這兩個invokeAny方法本質上都是在呼叫doInvokeAny方法，線上程池中提交多個任務，只要返回一個結果資料即可。
直接看上面的程式碼，大家可能有點暈。這裡，我舉一個例子，我們在使用執行緒池的時候，可能會啟動多個執行緒去執行各自的任務，比如執行緒A負責task_a，執行緒B負責task_b，這樣可以大規模提升系統處理任務的速度。如果我們希望其中一個執行緒執行完成返回結果資料時立即返回，而不需要再讓其他執行緒繼續執行任務。此時，就可以使用invokeAny方法。
- invokeAll方法
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
if (tasks == null)
throw new NullPointerException();
ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
//標識所有任務是否完成
boolean done = false;
try {
//遍歷所有任務
for (Callable<T> t : tasks) {
將每個任務封裝成RunnableFuture物件提交任務
RunnableFuture<T> f = newTaskFor(t);
//將結果資料新增到futures集合中
futures.add(f);
//執行任務
execute(f);
}
//遍歷結果資料集合
for (int i = 0, size = futures.size(); i < size; i++) {
Future<T> f = futures.get(i);
//任務沒有完成
if (!f.isDone()) {
try {
//阻塞等待任務完成並返回結果
f.get();
} catch (CancellationException ignore) {
} catch (ExecutionException ignore) {
}
}
}
//任務完成（不管是正常結束還是異常完成）
done = true;
//返回結果資料集合
return futures;
} finally {
//如果發生中斷異常InterruptedException 則取消已經提交的任務
if (!done)
for (int i = 0, size = futures.size(); i < size; i++)
futures.get(i).cancel(true);
}
}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {
if (tasks == null)
throw new NullPointerException();
long nanos = unit.toNanos(timeout);
ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
boolean done = false;
try {
for (Callable<T> t : tasks)
futures.add(newTaskFor(t));
final long deadline = System.nanoTime() + nanos;
final int size = futures.size();
for (int i = 0; i < size; i++) {
execute((Runnable)futures.get(i));
// 在新增執行任務時超時判斷，如果超時則立刻返回futures集合
nanos = deadline - System.nanoTime();
if (nanos <= 0L)
return futures;
}
// 遍歷所有任務
for (int i = 0; i < size; i++) {
Future<T> f = futures.get(i);
if (!f.isDone()) {
//對結果進行判斷時進行超時判斷
if (nanos <= 0L)
return futures;
try {
f.get(nanos, TimeUnit.NANOSECONDS);
} catch (CancellationException ignore) {
} catch (ExecutionException ignore) {
} catch (TimeoutException toe) {
return futures;
}
//重置任務的超時時間
nanos = deadline - System.nanoTime();
}
}
done = true;
return futures;
} finally {
if (!done)
for (int i = 0, size = futures.size(); i < size; i++)
futures.get(i).cancel(true);
}
}
invokeAll方法同樣實現了無超時時間設定和有超時時間設定的邏輯。
無超時時間設定的invokeAll方法總體邏輯為：將所有任務封裝成RunnableFuture物件，呼叫execute方法執行任務，將返回的結果資料新增到futures集合，之後對futures集合進行遍歷判斷，檢測任務是否完成，如果沒有完成，則呼叫get方法阻塞任務，直到返回結果資料，此時會忽略異常。最終在finally程式碼塊中對所有任務是否完成的標識進行判斷，如果存在未完成的任務，則取消已經提交的任務。
有超時設定的invokeAll方法總體邏輯與無超時時間設定的invokeAll方法總體邏輯基本相同，只是在兩個地方添加了超時的邏輯判斷。一個是在新增執行任務時進行超時判斷，如果超時，則立刻返回futures集合；另一個是每次對結果資料進行判斷時添加了超時處理邏輯。
invokeAll方法中本質上還是呼叫Executor介面的execute方法來提交任務。
- submit方法
submit方法的邏輯比較簡單，就是將任務封裝成RunnableFuture物件並提交，執行任務後返回Future結果資料。如下所示。
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方法提交任務時，本質上還是呼叫的Executor介面的execute方法。
綜上所述，在非定時任務類的執行緒池中提交任務時，本質上都是呼叫的Executor介面的execute方法。至於呼叫的是哪個具體實現類的execute方法，我們在後面的文章中深入分析。
五、ScheduledExecutorService介面
ScheduledExecutorService介面派生自ExecutorService介面，繼承了ExecutorService介面的所有功能，並提供了定時處理任務的能力，ScheduledExecutorService介面的原始碼比較簡單，如下所示。
package java.util.concurrent;
public interface ScheduledExecutorService extends ExecutorService {
//延時delay時間來執行command任務，只執行一次
public ScheduledFuture<?> schedule(Runnable command,
long delay, TimeUnit unit);
//延時delay時間來執行callable任務，只執行一次
public <V> ScheduledFuture<V> schedule(Callable<V> callable,
long delay, TimeUnit unit);
//延時initialDelay時間首次執行command任務，之後每隔period時間執行一次
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit);
//延時initialDelay時間首次執行command任務，之後每延時delay時間執行一次
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit);
}
至此，我們分析了執行緒池體系中重要的頂層介面和抽象類。
通過對這些頂層介面和抽象類的分析，我們需要從中感悟並體會軟體開發中的抽象思維，深入理解抽象思維在具體編碼中的實現，最終，形成自己的程式設計思維，運用到實際的專案中，這也是我們能夠從原始碼中所能學到的眾多細節之一。這也是高階或資深工程師和架構師必須瞭解原始碼細節的原因之一。
好了，今天就到這兒吧，我是冰河，我們下期見~~