java併發編程——FutureTask源碼分析

FutureTask的簡單示例：

FutureTask的應用場景，若是在當前線程中須要執行比較耗時的操做，但又不想阻塞當前線程時，能夠把這些做業交給FutureTask，另開一個線程在後臺完成，噹噹前線程未來須要時，就能夠經過FutureTask對象得到後臺做業的計算結果或者執行狀態。

public static void main(String[] args) throws InterruptedException{
        FutureTask<Integer> ft = new FutureTask<>(()->{//Callable接口的實現類
            int num = new Random().nextInt(10);
            TimeUnit.SECONDS.sleep(num);
            return num;
        });
        Thread t = new Thread(ft);
        t.start();
        //模擬主線程作一些其餘操做，跟futureTask任務並行
        //等須要futureTask的運行結果時，能夠調用get方法獲取。
        TimeUnit.SECONDS.sleep(2);
        try {
            //等待任務執行完成，獲取返回值
            Integer num = ft.get();
            System.out.println(num);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

Callable接口

Java如今的多線程機制，核心方法run是沒有返回值的；若是要保存run方法裏面的計算結果，必須等待run方法計算完，不管計算過程多麼耗時。而Callable接口能夠看做是Runnable接口的補充，call方法帶有返回值，而且能夠拋出異常。

public interface Callable<V> {
    V call() throws Exception;
}

Future接口

當咱們啓動一個線程（線程t1）去完成一個計算耗時的操做（調用耗時方法method()）時，若是另一個線程（線程t2）一直等待線程t1的計算結果這顯然是不明智的。可是，咱們能夠在調用method()的時候當即返回一個Future，而咱們能夠經過Feature獲取method()的各類執行信息（計算是否取消，是否計算完成，獲取執行結果）

public interface Future<V> {
    //還沒計算完，能夠取消計算過程
    boolean cancel(boolean mayInterruptIfRunning);
    //判斷計算是否被取消
    boolean isCancelled();
    //判斷是否計算完
    boolean isDone();
    //阻塞式獲取任務執行結果
    V get() throws InterruptedException, ExecutionException;
    //在指定的時間內獲取計算結果
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

RunnableFuture接口

RunnableFuture實現了Runnable和Future。所以FutureTask能夠傳遞到線程對象Thread或Excutor(線程池)來執行。

public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run();
}

FutureTask實現類

FutureTask實現了RunnableFuture接口。

public class FutureTask<V> implements RunnableFuture<V>

FutureTask的源碼分析 任務的執行狀態, 當咱們把FutureTask看做一個Future，那麼它的做用就是控制Callable的call方法的執行過程，在執行的過程當中天然會有狀態的轉換：

/**
     * 一般一個FutureTask新建出來，state就是NEW狀態；
     * COMPETING和INTERRUPTING用的進行時，表示瞬時狀態，存在時間極短；
     * NORMAL表明順利完成；
     * EXCEPTIONAL表明執行過程出現異常；
     * CANCELED表明執行過程被取消；
     * INTERRUPTED被中斷
     *
     * 可能的狀態轉移:
     * （執行過程順利完成）NEW -> COMPLETING -> NORMAL
     * （執行過程出現異常）NEW -> COMPLETING -> EXCEPTIONAL
     * （執行過程被取消）NEW -> CANCELLED
     * （執行過程當中，線程中斷）NEW -> INTERRUPTING -> INTERRUPTED
     */
    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;

    //將要執行的任務
    private Callable<V> callable;
    //用於get()返回的結果，也多是用於get()方法拋出的異常
    private Object outcome; // non-volatile, protected by state reads/writes
    //執行callable的線程，調用FutureTask.run()方法經過CAS設置
    private volatile Thread runner;
    //棧結構的等待隊列，該節點是棧中的最頂層節點。
    private volatile WaitNode waiters;

FutureTask的run()方法

public void run() {
        //保證callable任務只被運行一次，若是state狀態不爲New或者設置運行線程runner失敗則直接返回false，說明線程已經啓動過。
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;//callable從構造方法中傳入
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    result = c.call();//執行任務，並返回result
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex);//保存call方法拋出的異常，將state狀態設置成NORMAL
                }
                if (ran)
                    set(result);//保存call方法的執行結果，將state狀態設置成EXCEPTIONAL
            }
        } finally {
            runner = null;
            int s = state;
            //判斷該任務是否正在響應中斷，若是中斷沒有完成，則等待中斷操做完成
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }

FutureTask的set方法

protected void set(V v) {
    //經過CAS把state的NEW狀態修改爲COMPLETING狀態
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        //修改爲功則把v值賦給outcome變量。
        outcome = v;
        //而後再把state狀態修改爲NORMAL，表示如今能夠獲取返回值。
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        finishCompletion();//喚醒等待隊列中的全部節點。
    }
}

FutureTask的setException 方法

protected void setException(Throwable t) {
    //經過CAS把state的NEW狀態修改爲COMPLETING狀態。
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = t;//將異常值賦給outcome
        //把state狀態修改爲EXCEPTIONAL，表示待返回的異常信息設置成功。
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
        finishCompletion();//醒等待隊列中的全部節點。
    }
}

FutureTask的handlePossibleCancellationInterrupt方法

private void handlePossibleCancellationInterrupt(int s) {
    if (s == INTERRUPTING)
        while (state == INTERRUPTING)
            Thread.yield(); // 若是正在響應中斷，則調用Thread.yield()等待響應中斷結束（INTERRUPTED）。
}

FutureTask的finishCompletion方法

private void finishCompletion() {
    for (WaitNode q; (q = waiters) != null;) {
        //經過CAS把棧頂的元素置爲null，至關於彈出棧頂元素
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            for (;;) {
                Thread t = q.thread;
                if (t != null) {
                    q.thread = null;
                    LockSupport.unpark(t);//喚醒每個節點，通知每一個線程，該任務執行完成（多是執行完成，也可能cancel，異常等）
                }
                WaitNode next = q.next;
                if (next == null)
                    break;
                q.next = null; // unlink to help gc
                q = next;
            }
            break;
        }
    }
    done();
    callable = null;        // to reduce footprint
}

FutureTask的runAndReset 方法 （可被子類重寫，外部沒法直接調用）

protected boolean runAndReset() {
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
        return false;
    boolean ran = false;
    int s = state;
    try {
        Callable<V> c = callable;
        if (c != null && s == NEW) {
            try {
                c.call(); // don't set result
                ran = true;
            } catch (Throwable ex) {
                setException(ex);
            }
        }
    } finally {
        runner = null;
        s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
    return ran && s == NEW;
}

該方法和run方法的區別是，run方法只能被運行一次任務，而該方法能夠屢次運行任務。而runAndReset這個方法不會設置任務的執行結果值,若是該任務成功執行完成後，不修改state的狀態，仍是可運行（NEW）狀態，若是取消任務或出現異常，則不會再次執行。

FutureTask的get方法

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)//若是state狀態小於等於COMPLETING，說明任務還沒開始執行或還未執行完成
        s = awaitDone(false, 0L);//調用awaitDone方法阻塞該調用線程（將當前線程掛起等待）
    return report(s);//若是state的狀態大於COMPLETING，則說明任務執行完成，或發生異常、中斷、取消狀態。直接經過report方法返回執行結果。
}

FutureTask的get(long timeout, TimeUnit unit) 方法

public V get(long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException {
    if (unit == null)
        throw new NullPointerException();
    int s = state;
    if (s <= COMPLETING &&
        (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
        throw new TimeoutException();
    return report(s);
}

同get方法，該get方法支持阻塞等待多長時間，若是超時直接拋出TimeoutException異常。

FutureTask的report方法

private V report(int s) throws ExecutionException {
    Object x = outcome;
    //若是state的狀態爲NORMAL，說明任務正確執行完成，直接返回計算後的值。
    if (s == NORMAL)
        return (V)x;
    if (s >= CANCELLED)//state的狀態大於等於CANCELLED，說明任務被成功取消執行、或響應中斷，直接返回CancellationException異常
        throw new CancellationException();
    throw new ExecutionException((Throwable)x);//不然返回ExecutionException異常。
}

FutureTask的awaitDone 方法

構建棧鏈表的節點元素，並將該節點入棧，同時阻塞當前線程等待運行主任務的線程喚醒該節點。

JDK1.7版本是使用AQS的雙向鏈表隊列實現的。

private int awaitDone(boolean timed, long nanos)
    throws InterruptedException {
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    WaitNode q = null;
    boolean queued = false;
    for (;;) {
        //若是該線程執行interrupt()方法，則從隊列中移除該節點，並拋出異常
        if (Thread.interrupted()) {
            removeWaiter(q);
            throw new InterruptedException();
        }
        int s = state;
        //若是state狀態大於COMPLETING 則說明任務執行完成，或取消
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            return s;
        }
        //若是state=COMPLETING，則使用yield，由於此狀態的時間特別短，經過yield比掛起響應更快。
        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);
        //若是須要阻塞指定時間，則使用LockSupport.parkNanos阻塞指定時間
        //若是到指定時間還沒執行完，則從隊列中移除該節點，並返回當前狀態
        else if (timed) {
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) {
                removeWaiter(q);
                return state;
                }
                LockSupport.parkNanos(this, nanos);
            }
            //阻塞當前線程
            else
                LockSupport.park(this);
        }
}

FutureTask的removeWaiter 方法

移除棧中的節點元素，須要使用CAS自旋來保障移除成功。

private void removeWaiter(WaitNode node) {
    if (node != null) {
        node.thread = null;
        retry:
        for (;;) {          // restart on removeWaiter race
            for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
                s = q.next;
                if (q.thread != null)
                    pred = q;
                else if (pred != null) {
                    pred.next = s;
                    if (pred.thread == null) // check for race
                        continue retry;
                }
                else if (!UNSAFE.compareAndSwapObject(this, waitersOffset, q, s))
                    continue retry;
            }
            break;
        }
    }
}

FutureTask的cancel方法

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) {
            //調用runner.interupt()，設置狀態爲INTERRUPTED.喚醒全部在get()方法等待的線程
            try {
                Thread t = runner;
                if (t != null)
                    t.interrupt();
            } finally { // final state
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {
        finishCompletion();
    }
    return true;
}

總結： 任務開始運行後，不能在次運行，保證只運行一次（runAndReset 方法除外） 任務還未開始，或者任務已被運行，但未結束，這兩種狀況下均可以取消； 若是任務已經結束，則不能夠被取消 。

參考地址：

• https://www.jianshu.com/p/1dd951412ff5