非同步任務 -- FutureTask

Java · 發表 2019-01-24 21:03:00

摘要：任務提交之前在分析執行緒池的時候，提到過 AbstractExecutorService 的實現： public Future<?> submit(Runnable task) { if (task == null) throw new NullPointerEx...

任務提交

之前在分析執行緒池的時候，提到過 AbstractExecutorService 的實現：

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;
}

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);
}

對於 submit 提交的任務，不管是 Runnable 還是 Callable，最終都會統一為 FutureTask 並傳給 execute 方法。

public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW;// ensure visibility of callable
}

public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW;// ensure visibility of callable
}

對於 Runnable 還會建立一個介面卡：

static final class RunnableAdapter<T> implements Callable<T> {
final Runnable task;
final T result;
RunnableAdapter(Runnable task, T result) {
this.task = task;
this.result = result;
}
public T call() {
task.run();
return result;
}
}

任務狀態

FutureTask 有下面幾種狀態：

private volatile int state;
private static final int NEW= 0;
private static final int COMPLETING= 1;
private static final int NORMAL= 2;
private static final int EXCEPTIONAL= 3;
private static final int CANCELLED= 4;
private static final int INTERRUPTING = 5;
private static final int INTERRUPTED= 6;

初次建立的時候構造器中賦值 state = NEW，後面狀態可能有下面幾種演化：

NEW -> COMPLETING -> NORMAL （正常完成的過程）
NEW -> COMPLETING -> EXCEPTIONAL （執行過程中遇到異常）
NEW -> CANCELLED （執行前被取消）
NEW -> INTERRUPTING -> INTERRUPTED （取消時被中斷）

任務執行

當執行緒池執行任務的時候，最終都會執行 FutureTask 的 run 方法：

public void run() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
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 must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}

對於 Callable 直接執行其 call 方法。執行成功則呼叫 set 方法設定結果，如果遇到異常則呼叫 setException 設定異常：

protected void set(V v) {
// 首先 CAS 設定 state 為中間狀態 COMPLETING
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
// 設定為正常狀態
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}

protected void setException(Throwable t) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
// 設定為異常狀態
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}

這兩個方法都是對全域性變數 outcome 的賦值。當我們通過 get 方法獲取結果時，往往是在另一個執行緒：

public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}

如果任務還沒有完成則等待任務完成：

private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
// 通過 for 迴圈來阻塞當前執行緒
for (;;) {
// 響應中斷
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}

int s = state;
// 任務已完成或者已丟擲異常直接返回
if (s > COMPLETING) {
// WaitNode已建立此時也沒用了
if (q != null)
q.thread = null;
return s;
}
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
else if (q == null)
q = new WaitNode();
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);
}
else
LockSupport.park(this);
}
}

如果任務已完成或者等待任務直到完成後，呼叫 report 方法返回結果：

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);
}

如果 state == NORMAL，標識任務正常完成，返回實際結果。如果 state >= CANCELLED，則返回 CancellationException，否則返回 ExecutionException，這樣線上程池中執行的任務不管是異常還是正常返回了結果，都能被感知。

Treiber Stack

/**
 * Simple linked list nodes to record waiting threads in a Treiber
 * stack.See other classes such as Phaser and SynchronousQueue
 * for more detailed explanation.
 */
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}

在 awaitDone 方法中 WaitNode q = null，第一次會建立一個 WaitNode，這時即使有多個執行緒在等待結果，都會建立各自的 WaitNode：

else if (q == null)
q = new WaitNode();
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);

然後在for迴圈中會跳到第二個 else if，由於沒有入隊，這時會通過 CAS 將新建的 WaitNode 型別的 q 賦值給 waiters，這個時候同一時刻只有一個執行緒能賦值成功，後一個在失敗後又經歷一次迴圈，最終成功地將當前 WaitNode 插入到 waiters 的頭部。

任務取消

FutureTask 有一個 cancel 方法，包含一個 boolean 型別的引數（在執行中的任務是否可以中斷）：

public boolean cancel(boolean mayInterruptIfRunning) {
// 如果任務不是剛建立或者是剛建立但是更改為指定狀態失敗則返回 false
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;
}

最終都會呼叫 finishCompletion() ，在 set 方法和 setException 方法中也呼叫了這個 finishCompletion 方法：

private void finishCompletion() {
// assert state > COMPLETING;
// 如果任務執行完或者存在異常的話這個waiters已經為null了
for (WaitNode q; (q = waiters) != null;) {
// 首先不斷嘗試把 waiters 設定為 null，如果很多執行緒呼叫 task.cancel()，也只有一個能成功
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
}

當在 finishCompletion 方法中喚醒執行緒後，被喚醒的執行緒在 awaitDone 方法中繼續迴圈，發現狀態已完成：

int s = state;
// 任務已完成或者已丟擲異常直接返回
if (s > COMPLETING) {
// WaitNode已建立此時也沒用了
if (q != null)
q.thread = null;
return s;
}

接著呼叫 report 方法，發現狀態為異常的話將包裝成 ExecutionException((Throwable)x); 這個異常就是我們在使用 get 的時候需要捕獲的異常。

最近比較忙，這塊東西已經很久沒有看了， FutureTask 感覺沒有徹底弄明白，也沒有一個好的結尾，現在這裡標記下，後面繼續更新。