溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
（手機橫屏看源碼更方便）
注：java源碼分析部分如無特殊說明均基于 java8 版本。
注：線程池源碼部分如無特殊說明均指ThreadPoolExecutor類。
前面我們一起學習了線程池中普通任務的執行流程，但其實線程池中還有一種任務，叫作未來任務（future task），使用它您可以獲取任務執行的結果，它是怎么實現的呢？
建議學習本章前先去看看彤哥之前寫的《死磕 java線程系列之自己動手寫一個線程池（續）》，有助于理解本章的內容，且那邊的代碼比較短小，學起來相對容易一些。
（1）線程池中的未來任務是怎么執行的？
（2）我們能學到哪些比較好的設計模式？
（3）對我們未來學習別的框架有什么幫助？
我們還是從一個例子入手，來講解來章的內容。
我們定義一個線程池，并使用它提交5個任務，這5個任務分別返回0、1、2、3、4，在未來的某一時刻，我們再取用它們的返回值，做一個累加操作。
public class ThreadPoolTest02 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
// 新建一個固定5個線程的線程池
ExecutorService threadPool = Executors.newFixedThreadPool(5);
List<Future<Integer>> futureList = new ArrayList<>();
// 提交5個任務，分別返回0、1、2、3、4
for (int i = 0; i < 5; i++) {
int num = i;
// 任務執行的結果用Future包裝
Future<Integer> future = threadPool.submit(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("return: " + num);
// 返回值
return num;
});
// 把future添加到list中
futureList.add(future);
}
// 任務全部提交完再從future中get返回值，并做累加
int sum = 0;
for (Future<Integer> future : futureList) {
sum += future.get();
}
System.out.println("sum=" + sum);
}
}這里我們思考兩個問題：
（1）如果這里使用普通任務，要怎么寫，時間大概是多少？
如果使用普通任務，那么就要把累加操作放到任務里面，而且并不是那么好寫（final的問題），總時間大概是1秒多一點。但是，這樣有一個缺點，就是累加操作跟任務本身的內容耦合到一起了，后面如果改成累乘，還要修改任務的內容。
（2）如果這里把future.get()放到for循環里面，時間大概是多少？
這個問題我們先不回答，先來看源碼分析。
submit方法，它是提交有返回值任務的一種方式，內部使用未來任務（FutureTask）包裝，再交給execute()去執行，最后返回未來任務本身。
public <T> Future<T> submit(Callable<T> task) {
// 非空檢測
if (task == null) throw new NullPointerException();
// 包裝成FutureTask
RunnableFuture<T> ftask = newTaskFor(task);
// 交給execute()方法去執行
execute(ftask);
// 返回futureTask
return ftask;
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
// 將普通任務包裝成FutureTask
return new FutureTask<T>(callable);
}這里的設計很巧妙，實際上這兩個方法都是在AbstractExecutorService這個抽象類中完成的，這是模板方法的一種運用。
我們來看看FutureTask的繼承體系：
FutureTask實現了RunnableFuture接口，而RunnableFuture接口組合了Runnable接口和Future接口的能力，而Future接口提供了get任務返回值的能力。
問題：submit()方法返回的為什么是Future接口而不是RunnableFuture接口或者FutureTask類呢？
答：這是因為submit()返回的結果，對外部調用者只想暴露其get()的能力（Future接口），而不想暴露其run()的能力（Runaable接口）。
經過上一章的學習，我們知道execute()方法最后調用的是task的run()方法，上面我們傳進去的任務，最后被包裝成了FutureTask，也就是說execute()方法最后會調用到FutureTask的run()方法，所以我們直接看這個方法就可以了。
public void run() {
// 狀態不為NEW，或者修改為當前線程來運行這個任務失敗，則直接返回
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
// 真正的任務
Callable<V> c = callable;
// state必須為NEW時才運行
if (c != null && state == NEW) {
// 運行的結果
V result;
boolean ran;
try {
// 任務執行的地方【本文由公從號“彤哥讀源碼”原創】
result = c.call();
// 已執行完畢
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
// 處理異常
setException(ex);
}
if (ran)
// 處理結果
set(result);
}
} finally {
// 置空runner
runner = null;
// 處理中斷
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}可以看到代碼也比較簡單，先做狀態的檢測，再執行任務，最后處理結果或異常。
執行任務這里沒啥問題，讓我們看看處理結果或異常的代碼。
protected void setException(Throwable t) {
// 將狀態從NEW置為COMPLETING
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
// 返回值置為傳進來的異常（outcome為調用get()方法時返回的）
outcome = t;
// 最終的狀態設置為EXCEPTIONAL
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
// 調用完成方法
finishCompletion();
}
}
protected void set(V v) {
// 將狀態從NEW置為COMPLETING
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
// 返回值置為傳進來的結果（outcome為調用get()方法時返回的）
outcome = v;
// 最終的狀態設置為NORMAL
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
// 調用完成方法
finishCompletion();
}
}咋一看，這兩個方法似乎差不多，不同的是出去的結果不一樣且狀態不一樣，最后都調用了finishCompletion()方法。
private void finishCompletion() {
// 如果隊列不為空（這個隊列實際上為調用者線程）
for (WaitNode q; (q = waiters) != null;) {
// 置空隊列
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
for (;;) {
// 調用者線程
Thread t = q.thread;
if (t != null) {
q.thread = null;
// 如果調用者線程不為空，則喚醒它
// 【本文由公從號“彤哥讀源碼”原創】
LockSupport.unpark(t);
}
WaitNode next = q.next;
if (next == null)
break;
q.next = null; // unlink to help gc
q = next;
}
break;
}
}
// 鉤子方法，子類重寫
done();
// 置空任務
callable = null; // to reduce footprint
}整個run()方法總結下來：
（1）FutureTask有一個狀態state控制任務的運行過程，正常運行結束state從NEW->COMPLETING->NORMAL，異常運行結束state從NEW->COMPLETING->EXCEPTIONAL；
（2）FutureTask保存了運行任務的線程runner，它是線程池中的某個線程；
（3）調用者線程是保存在waiters隊列中的，它是什么時候設置進去的呢？
（4）任務執行完畢，除了設置狀態state變化之外，還要喚醒調用者線程。
調用者線程是什么時候保存在FutureTask中（waiters）的呢？查看構造方法：
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}發現并沒有相關信息，我們再試想一下，如果調用者不調用get()方法，那么這種未來任務是不是跟普通任務沒有什么區別？確實是的哈，所以只有調用get()方法了才有必要保存調用者線程到FutureTask中。
所以，我們來看看get()方法中是什么鬼。
get()方法調用時如果任務未執行完畢，會阻塞直到任務結束。
public V get() throws InterruptedException, ExecutionException {
int s = state;
// 如果狀態小于等于COMPLETING，則進入隊列等待
if (s <= COMPLETING)
s = awaitDone(false, 0L);
// 返回結果（異常）
return report(s);
}是不是很清楚，如果任務狀態小于等于COMPLETING，則進入隊列等待。
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
// 我們這里假設不帶超時
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) {
// 處理中斷
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
// 4. 如果狀態大于COMPLETING了，則跳出循環并返回
// 這是自旋的出口
int s = state;
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
// 如果狀態等于COMPLETING，說明任務快完成了，就差設置狀態到NORMAL或EXCEPTIONAL和設置結果了
// 這時候就讓出CPU，優先完成任務
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
// 1. 如果隊列為空
else if (q == null)
// 初始化隊列（WaitNode中記錄了調用者線程）
q = new WaitNode();
// 2. 未進入隊列
else if (!queued)
// 嘗試入隊
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
// 超時處理
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
// 3. 阻塞當前線程（調用者線程）
else
// 【本文由公從號“彤哥讀源碼”原創】
LockSupport.park(this);
}
}這里我們假設調用get()時任務還未執行，也就是其狀態為NEW，我們試著按上面標示的1、2、3、4走一遍邏輯：
（1）第一次循環，狀態為NEW，直接到1處，初始化隊列并把調用者線程封裝在WaitNode中；
（2）第二次循環，狀態為NEW，隊列不為空，到2處，讓包含調用者線程的WaitNode入隊；
（3）第三次循環，狀態為NEW，隊列不為空，且已入隊，到3處，阻塞調用者線程；
（4）假設過了一會任務執行完畢了，根據run()方法的分析最后會unpark調用者線程，也就是3處會被喚醒；
（5）第四次循環，狀態肯定大于COMPLETING了，退出循環并返回；
問題：為什么要在for循環中控制整個流程呢，把這里的每一步單獨拿出來寫行不行？
答：因為每一次動作都需要重新檢查狀態state有沒有變化，如果拿出去寫也是可以的，只是代碼會非常冗長。這里只分析了get()時狀態為NEW，其它的狀態也可以自行驗證，都是可以保證正確的，甚至兩個線程交叉運行（斷點的技巧）。
OK，這里返回之后，再看看是怎么處理最終的結果的。
private V report(int s) throws ExecutionException {
Object x = outcome;
// 任務正常結束
if (s == NORMAL)
return (V)x;
// 被取消了
if (s >= CANCELLED)
throw new CancellationException();
// 執行異常
throw new ExecutionException((Throwable)x);
}還記得前面分析run的時候嗎，任務執行異常時是把異常放在outcome里面的，這里就用到了。
（1）如果正常執行結束，則返回任務的返回值；
（2）如果異常結束，則包裝成ExecutionException異常拋出；
通過這種方式，線程中出現的異常也可以返回給調用者線程了，不會像執行普通任務那樣調用者是不知道任務執行到底有沒有成功的。
FutureTask除了可以獲取任務的返回值以外，還能夠取消任務的執行。
public boolean cancel(boolean mayInterruptIfRunning) {
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try { // in case call to interrupt throws exception
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
return true;
}這里取消任務是通過中斷執行線程來處理的，有興趣的同學可以自己分析一下。
如果這里把future.get()放到for循環里面，時間大概是多少？
答：大概會是5秒多一點，因為每提交一個任務，都要阻塞調用者線程直到任務執行完畢，每個任務執行都是1秒多，所以總時間就是5秒多點。
（1）未來任務是通過把普通任務包裝成FutureTask來實現的。
（2）通過FutureTask不僅能夠獲取任務執行的結果，還有感知到任務執行的異常，甚至還可以取消任務；
（3）AbstractExecutorService中定義了很多模板方法，這是一種很重要的設計模式；
（4）FutureTask其實就是典型的異常調用的實現方式，后面我們學習到Netty、Dubbo的時候還會見到這種設計思想的。
RPC框架中異步調用是怎么實現的？
答：RPC框架常用的調用方式有同步調用、異步調用，其實它們本質上都是異步調用，它們就是用FutureTask的方式來實現的。
一般地，通過一個線程（我們叫作遠程線程）去調用遠程接口，如果是同步調用，則直接讓調用者線程阻塞著等待遠程線程調用的結果，待結果返回了再返回；如果是異步調用，則先返回一個未來可以獲取到遠程結果的東西FutureXxx，當然，如果這個FutureXxx在遠程結果返回之前調用了get()方法一樣會阻塞著調用者線程。
有興趣的同學可以先去預習一下dubbo的異步調用（它是把Future扔到RpcContext中的）。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
