硬核詳解 FutureTask 設計與實現
最近看到一篇比較不錯的FutureTask實踐，于是對FutureTask源碼進行了研究，而本文將從實踐和源碼兩個角度分析FutureTask的設計與實現思路，希望對你有幫助。
FutureTask使用示例
我們的批量線程會進行嘗試創(chuàng)建一些任務執(zhí)行，同時我們希望每個任務只有有一個線程去執(zhí)行，其他線程如果拿到這個任務準備執(zhí)行時發(fā)現這個任務已經在執(zhí)行，則等待這個任務的返回結果。
如下圖，假設我們的第一個線程提交task-1至清單成功后，這個線程就會執(zhí)行該任務，而線程2同樣想提交這個任務發(fā)現該任務已存在，則直接等待清單中記錄的這個任務的結果返回。而線程3則也是因為第一次提交任務，所以提交清單成功并執(zhí)行。
我們完全可以通過FutureTask做到這一點，先來說說任務清單，它用于存儲正在執(zhí)行的任務和任務名，為了保證多線程并發(fā)操作安全，筆者直接采用ConcurrentHashMap：
//保存執(zhí)行的任務清單
private static ConcurrentHashMap<String, Future<String>> taskCache = new ConcurrentHashMap<>();每一個線程執(zhí)行任務則都是通過executionTask方法，該邏輯一旦檢查任務不存在則創(chuàng)建，然后通過樂觀鎖方式保證任務不存在時才能提交，完成這些操作后通過get等待結果返回。 需要注意的是，這里筆者為了保證邏輯可用做了一點小小的計數處理，每當一個任務通過run執(zhí)行時，筆者會用原子類自增一下，以此判斷多線程執(zhí)行該方法時有沒有重復執(zhí)行任務的情況出現。
privatestatic AtomicInteger counter = new AtomicInteger(0);
public static String executionTask(String taskName) {
while (true){//執(zhí)行到任務成功
if (!taskCache.containsKey(taskName)) {//如果任務不存在則創(chuàng)建任務
Callable<String> callable = () -> taskName;
FutureTask<String> futureTask = new FutureTask<>(callable);
//雙重鎖校驗避免任務重復提交
Future<String> future = taskCache.putIfAbsent(taskName, futureTask);
if (future == null) {//如果返回空則說明這個任務之前沒有提交過，當前線程直接執(zhí)行
futureTask.run();
//累加執(zhí)行的任務數
counter.incrementAndGet();
}
}
//任務非空或提交任務完成的線程在這里等待任務返回
try {
return taskCache.get(taskName).get();
} catch (Exception e) {
//如果保存則從清單中移除
taskCache.remove(taskName);
}
}
}對應的測試代碼如下，筆者提交10w個線程執(zhí)行10個任務，從計數器的輸出結果來看，執(zhí)行了10個任務，沒有問題：
public static void main(String[] args) throws Exception {
//創(chuàng)建線程池
ExecutorService executorService = Executors.newFixedThreadPool(10_000);
for (int i = 0; i < 10_000; i++) {
//通過取模運算保證10w個線程只會創(chuàng)建 名為10以內的任務
int finalI = i % 10;
//提交任務
executorService.submit(() -> FutureTaskExample.executionTask(String.valueOf(finalI)));
}
//等待結束
executorService.shutdown();
while (!executorService.isTerminated()) {
}
//查看計數
System.out.println(counter.get());//10
}FutureTask在閉鎖下的哲學
本質上FutureTask也可以是一種閉鎖，即在FutureTask對應線程未完成運算前，FutureTask這個閉鎖就像一個大門一樣不允許所有線程通過，只有FutureTask完成運算進入完成狀態(tài)后，其它線程才能通過。
例如我們希望通過FutureTask執(zhí)行一些耗時的運算，此時就可以：
- 通過FutureTask提交任務
- 異步任務運算期間，執(zhí)行一些其它任務
- 通過get阻塞等待FutureTask結果返回
- FutureTask任務返回線程通過，打印輸出結果
所以FutureTask也是一個高效的異步工具，我們可以將一些耗時的操作提前啟動，著手其它耗時操作等待完成后拿結果：
對應的我們也給出上述實現的代碼示例：
public class Task {
//休眠完成后，返回一個隨機數
privatefinal FutureTask<Integer> futureTask = new FutureTask<>(() -> {
ThreadUtil.sleep(1000);
return RandomUtil.randomInt();
});
privatefinal Thread thread = new Thread(futureTask);
//啟動任務執(zhí)行
public void start() {
thread.start();
}
//阻塞獲取結果
public int get() throws ExecutionException, InterruptedException {
return futureTask.get();
}
}使用實例如下，即通過FutureTask執(zhí)行異步運算后，通過get執(zhí)行閉鎖控制異步流程：
Task task = new Task();
long begin = System.currentTimeMillis();
task.start();
//futureTask異步運行期間，執(zhí)行一些業(yè)務邏輯
System.out.println("do something...");
//阻塞等待futureTask完成，即利用get方法實現一個閉鎖的操作
int result = task.get();
long end = System.currentTimeMillis();
Console.log("result: {},cost:{}ms", result, end - begin);輸出結果如下，可以看到整體來說利用了異步運算期間執(zhí)行一些其它操作，同時還使用get保證整體流程順序正常：
do something...
result: -1158881871,cost:1009msFutureTask狀態(tài)機的扭轉
在正式進行源碼介紹之前，筆者先簡單介紹一下FutureTask執(zhí)行狀態(tài)的扭轉，FutureTask在創(chuàng)建時是全新的任務，此時它的狀態(tài)就是NEW，一旦調用run之后就會開始運行，此時就會出現4個分支：
- 當任務正確執(zhí)行完成之后，先將狀態(tài)設置為完成中COMPLETING，然后將執(zhí)行結果存入outcome變量中，完成后再將狀態(tài)設置為正常結束，即NORMAL。
- 一旦任務執(zhí)行出現異常，FutureTask則同樣將任務設置為完成中，將結果設置為null之后，調整狀態(tài)為執(zhí)行出錯EXCEPTIONAL。
- 當任務在執(zhí)行過程中，需要進行打斷操作時，FutureTask會將狀態(tài)設置為打斷中INTERRUPTING，一旦線程正常被打斷，任務就會被設置為終態(tài)INTERRUPTED。
- FutureTask同樣支持不打斷當前執(zhí)行線程，這一點筆者會在后文中說明，這種情況則直接將線程設置為CANCELLED。
一旦狀態(tài)按照預期調整為終態(tài)后，FutureTask就會喚醒那些等待任務執(zhí)行完成的線程：
這4種狀態(tài)扭轉，我們也可以通過閱讀FutureTask上關于狀態(tài)變量的注釋了解：
/*
* Possible state transitions:
* NEW -> COMPLETING -> NORMAL
* NEW -> COMPLETING -> EXCEPTIONAL
* NEW -> CANCELLED
* NEW -> INTERRUPTING -> INTERRUPTED
*/
privatevolatileint state;
privatestaticfinalint NEW = 0;
privatestaticfinalint COMPLETING = 1;
privatestaticfinalint NORMAL = 2;
privatestaticfinalint EXCEPTIONAL = 3;
privatestaticfinalint CANCELLED = 4;
privatestaticfinalint INTERRUPTING = 5;
privatestaticfinalint INTERRUPTED = 6;FutureTask如何執(zhí)行
FutureTask在創(chuàng)建之初時的初始態(tài)為NEW，也就是整數變量0，一旦某個線程執(zhí)行這個任務，JDK8版本會通過原子操作將記錄運行線程的地址(這個位置的偏移量用runnerOffset記錄)設置為當前線程，一旦操作成功則執(zhí)行該task封裝的Callable任務，如果運行成功則將result設置為運行后的結果，并將ran標志為true，并將任務狀態(tài)設置為NORMAL：
反之如果運行失敗，則將ran設置為false，result設置為空，并將錯誤信息設置到stateOffset標志位上了，將任務運行狀態(tài)設置為終態(tài)EXCEPTIONAL，然后喚醒其他需要處理這個任務的線程：
對此我們給出FutureTask的底層實現，可以看到FutureTask只有狀態(tài)為NEW且通過CAS操作將runner設置為自己的線程才能執(zhí)行任務，而后續(xù)的線程如果看到state不為new則只能獲取結果，而不能執(zhí)行，這就FutureTask避免重復運行的核心設計所在。 進行樂觀鎖上鎖拿到執(zhí)行權之后，就會基于CAS上鎖的線程進行任務調用，對應的結果扭轉就如上文所說，這里讀者可以參考筆者注釋自行閱讀：
public void run() {
//1. 為NEW 說明是第一次運行，則可以通過CAS操作獲取執(zhí)行權
//2. 不為NEW，直接返回
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
//如果任務不為空且狀態(tài)為NEW則調用call運行
if (c != null && state == NEW) {
V result;
boolean ran;
try {
//運行成功則記錄結果到result 并將ran 設置為true
result = c.call();
ran = true;
} catch (Throwable ex) {
//如果報錯則result設置為空，并將ran設置為false
result = null;
ran = false;
//并將任務狀態(tài)設置為錯誤
setException(ex);
}
//如果運行成功則將結果存到outcome中
if (ran)
set(result);
}
} finally {
//......
}
}這里我們直接查看獲取結果成功后的狀態(tài)設置方法set方法，可以看到其內部先通過cas將status即狀態(tài)字段設置為COMPLETING，完成結果設置之后，再通過putOrderedInt將狀態(tài)設置為終態(tài)NORMAL，這么做的原因是為什么呢？
protected void set(V v) {
//先CAS設置為完成中
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
//設置結果完成后，才能設置狀態(tài)為正常結束
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}多線程執(zhí)行FutureTask情況下，大部分判斷都需要依賴于COMPLETING這個中間態(tài)(利用get和COMPLETING方法)，所以這個狀態(tài)的可見性要求相對高一些，所以在進行結果設置之前，先通過CAS的方式進行更新status字段狀態(tài)，這種操作是需要將storeLoad屏障，雖然性能表現差一些，但可以保證可見性和有序性，所以先通過這個操作保證其他線程對于這個中態(tài)狀態(tài)可見保證并發(fā)操作一致性。
然后完成任務處理結果設置，此時在邏輯上我們可以認定這個任務是處理完成了，因為大部分的邏輯判斷都是依賴于COMPLETING，對于終態(tài)(NORMAL、EXCEPTIONAL、CANCELLED、INTERRUPTED)的可見性要求不高，所以FutureTask的設計者直接采用putOrderedInt這種操作保證寫入不會被重排序，但不會立即刷到一致性內存行上，所以在性能表現上會出色一些。
上述對于正確處理結果的設置，可以在set這個方法的源碼上得以印證，讀者可以結合上文筆者所說和注釋自行了解這段邏輯：
protected void set(V v) {
//通過cas操作volatile變量status完成中態(tài)設置，又保證的多核心可見性
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
//設置處理結果到outcome上，后續(xù)其他線程get都是從這個outcome變量上獲取
outcome = v;
//因為終態(tài)可見性要求不高，所以通過putOrderedInt設置終態(tài)，保證寫入有序性
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
//喚醒其他線程處理當前任務
finishCompletion();
}
}多線程如何獲取FutureTask執(zhí)行結果
其他線程需要從get方法獲取結果時，其內部本質就是調用awaitDone等待完成，假設我們用putOrderedInt寫入，且狀態(tài)對于當前線程不可見，那么這個線程要做的也僅僅是yield讓出處理器的使用權，相比之下volatile寫這種需要增加內存屏障寫入的開銷，這種內存消耗無論從概率還是消耗上，都是劃得來的。
最后完成上述操作，通過finishCompletion通知其他的等待線程可以開始處理FutureTask了。
對此我們也給出get的源碼和注釋，讀者可結合上文感知一下邏輯：
public V get() throws InterruptedException, ExecutionException {
int s = state;
//狀態(tài)在COMPLETING及其之前調用awaitDone等待完成
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
finallong deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) {
//......
//如果狀態(tài)處于中態(tài)完成中，則讓出線程對于處理器的使用權
elseif (s == COMPLETING) // cannot time out yet
Thread.yield();
//......
}我們給出狀態(tài)設置為終態(tài)之后，finishCompletion的邏輯，比較簡單，可以看到它僅是獲取當前等待節(jié)點的后一個線程的WaitNode ，通過unpark將其喚醒，然后獲取其后繼節(jié)點繼續(xù)進行喚醒操作：
private void finishCompletion() {
// assert state > COMPLETING;
//獲取后繼節(jié)點
for (WaitNode q; (q = waiters) != null;) {
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
for (;;) {
//將該等待節(jié)點直接喚醒
Thread t = q.thread;
if (t != null) {
q.thread = null;
LockSupport.unpark(t);
}
//指向已喚醒節(jié)點的后繼節(jié)點
WaitNode next = q.next;
//若為空直接退出
if (next == null)
break;
//將后繼指針設置為空，輔助gc，并讓q指向后續(xù)節(jié)點，繼續(xù)進行喚醒操作
q.next = null; // unlink to help gc
q = next;
}
break;
}
}
//......
}FutureTask的異常處理
有了上文設置成功的邏輯解說，相信讀者對于setException的邏輯處理也就比較熟悉了，這里筆者就不再展開，讀者可自行查看代碼和注釋：
protected void setException(Throwable t) {
//原子操作設置為完成中，保證可見性
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
//結果設置為null
outcome = t;
//通過高性能有序寫putOrderedInt設置終態(tài)為錯誤態(tài)
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
//喚醒其他線程
finishCompletion();
}
}FutureTask如何取消
任務取消操作有兩種情況，如果我們傳參為true，則會通過原子操作將狀態(tài)設置為打斷中，再嘗試打斷正在執(zhí)行的線程，然后將狀態(tài)設置為已打斷這個終態(tài)INTERRUPTED，再喚醒其他線程。
若傳參設置為false，則直接原子操作設置為已取消，然后直接喚醒其他線程。
public boolean cancel(boolean mayInterruptIfRunning) {
//如果狀態(tài)不為new則進行狀態(tài)原子設置操作
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
returnfalse;
try {
//如果入參為true則嘗試打斷線程
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
//完成打斷設置狀態(tài)為INTERRUPTED
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
//喚醒其他線程
finishCompletion();
}
returntrue;
}FutureTask如何避免任務重復執(zhí)行
我們假設當前任務為初始化NEW，這意味所有任務都可以嘗試進行上樂觀鎖獲取執(zhí)行勸，如下所示只有設置成功的線程才可以進入后續(xù)的執(zhí)行，而其他線程則直接返回：
public void run() {
//只有狀態(tài)為NEW的情況下，當前執(zhí)行的線程才會通過CAS獲取樂觀鎖，之后獲取樂觀鎖成功的才能執(zhí)行任務，其他的線程會直接返回
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
//......
//執(zhí)行任務邏輯
}JDK8版本FutureTask的變化
這一點，筆者參閱了1.6版本的FutureTask實現，其內部是通過AQS來維護需要執(zhí)行任務的線程及其狀態(tài)，而JDK8版本則專門為FutureTask創(chuàng)建了一個state字段，多線程之間通過CAS操作進行維護。
我們先來說明一下早期版本的設計再說明原因，在較早的版本FutureTask內部線程競爭關系和任務狀態(tài)都采用AQS進行維護，假設當前任務被取消，則執(zhí)行這個操作線程會通過原子操作將AQS隊列的state字段更新為CANCELLED。
同理執(zhí)行任務時，也是先通過AQS的原子操作將狀態(tài)設置為RUNNING，執(zhí)行完成之后將操作結果原子修改為RAN，并將結果記錄到FutureTask的reuslt中：
對此我們給出老版本的FutureTask的run方法，邏輯如上文所說，筆者這里就不多做贅述：
public void run() {
sync.innerRun();
}
void innerRun() {
//原子操作將狀態(tài)設置為RUNNING
if (!compareAndSetState(0, RUNNING))
return;
try {
//獲取當前線程，檢查當前狀態(tài)還是RUNNING則用innerSet當前執(zhí)行執(zhí)行callable，然后將結果設置到result中，并將狀態(tài)修改為RAN
runner = Thread.currentThread();
if (getState() == RUNNING) // recheck after setting thread
innerSet(callable.call());
else
releaseShared(0); // cancel
} catch (Throwable ex) {
//......
}
}我們來看看cancel方法，可以看到只要不是RAN這種終態(tài)，就可以嘗試打斷線程：
public boolean cancel(boolean mayInterruptIfRunning) {
return sync.innerCancel(mayInterruptIfRunning);
}
boolean innerCancel(boolean mayInterruptIfRunning) {
for (; ; ) {
//獲取狀
int s = getState();
//如果是RAN或者canceled則直接返回
if (ranOrCancelled(s))
returnfalse;
//將狀態(tài)設置為CANCELLED
if (compareAndSetState(s, CANCELLED))
break;
}
// 按照mayInterruptIfRunning的布爾值決定是否打斷線程
if (mayInterruptIfRunning) {
Thread r = runner;
if (r != null)
r.interrupt();
}
releaseShared(0);
done();
returntrue;
}有意思的來了，我們試想這樣一個場景：
- 假設我們執(zhí)行上述run方法的任務處于RUNNING狀態(tài)，此時當前線程1就可以調用interrupt打斷這個線程
- 此時線程2看到當前任務處于Running(源碼說明不是ran或者canceled即可打斷)直接將其打斷
- 又假設我們這個線程跑去執(zhí)行別的task任務
所以線程1就可能在執(zhí)行別的任務期間被打斷，進而出現幻覺死：
于是就有了JDK7之后版本，FutureTask專門使用一個volatile的state維護任務的狀態(tài)，并且在打斷前設置一個中態(tài)INTERRUPTING 即打斷中，執(zhí)行任務的線程1在運行完成將結果存入outcome之后，看到打斷中這個中態(tài)就會循環(huán)調用yield讓出執(zhí)行權，直到執(zhí)行cancel操作的線程完成，由此保證了打斷操作永遠被限制在當前FutureTask生命周期以內。
對應我們給出cancel操作的源碼，可以看到它是先設置為打斷中INTERRUPTING 然后在進行打斷再設置打斷完成INTERRUPTED的終態(tài)：
public boolean cancel(boolean mayInterruptIfRunning) {
//原子操作設置狀態(tài)為打斷中
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
returnfalse;
try {
//嘗試打斷線程，完成后設置為INTERRUPTED
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
returntrue;
}假設上一步的線程的打斷操作還未完成，這里可以直接理解為執(zhí)行interrupt打斷之前的代碼段，而執(zhí)行我們的run的線程已經運行完成并將結果設置到outcome時，如下所示會執(zhí)行finally 語句塊的handlePossibleCancellationInterrupt，它在看到打斷中的操作后會循環(huán)調用Thread.yield()讓當前線程讓出CPU執(zhí)行權，直到其他線程的cancel操作完成：
public void run() {
//.....
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
//執(zhí)行邏輯運算
result = c.call();
ran = true;
} catch (Throwable ex) {
//......
}
//將結果設置到outcome中
if (ran)
set(result);
}
} finally {
//......
//讀取到打斷中的狀態(tài)，調用handlePossibleCancellationInterrupt讓線程yield出去，保證打斷操作限定在當前線程內
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
private void handlePossibleCancellationInterrupt(int s) {
//循環(huán)yield直到cacnel操作完成
if (s == INTERRUPTING)
while (state == INTERRUPTING)
Thread.yield(); // wait out pending interrupt
}關于這個問題我們也可以參考源碼上的注釋：
/*
* Revision notes: This differs from previous versions of this
* class that relied on AbstractQueuedSynchronizer, mainly to
* avoid surprising users about retaining interrupt status during
* cancellation races. Sync control in the current design relies
* on a "state" field updated via CAS to track completion, along
* with a simple Treiber stack to hold waiting threads.
*
* Style note: As usual, we bypass overhead of using
* AtomicXFieldUpdaters and instead directly use Unsafe intrinsics.
*/
