異步任務 -- FutureTask

任務提交

以前在分析線程池的時候，提到過 AbstractExecutorService 的實現：html

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 方法。node

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 有下面幾種狀態：this

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，後面狀態可能有下面幾種演化：spa

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 設置異常：code

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 方法獲取結果時，每每是在另外一個線程：htm

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

若是任務尚未完成則等待任務完成：blog

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 方法返回結果：rem

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 感受沒有完全弄明白，也沒有一個好的結尾，如今這裏標記下，後面繼續更新。

原文出處：https://www.cnblogs.com/lucare/p/10316808.html