FutureTask源碼解析

第一部分:What

在Java中一般通過繼承Thread類或者實現Runnable接口這兩種方式來創建多線程，但是這兩種方式都有個缺陷，就是不能在執行完成后獲取執行的結果，因此Java 1.5之后提供了Callable和Future接口，通過它們就可以在任務執行完畢之后得到任務的執行結果。本文會簡要的介紹使用方法，然后會從源代碼角度分析下具體的實現原理。
本文以Java 1.7的代碼進行分析。

第二部分:How

Callable接口

對於需要執行的任務需要實現Callable接口，Callable接口定義如下:

public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    V call() throws Exception;
}

可以看到Callable是個泛型接口，泛型V就是要call()方法返回的類型。Callable接口和Runnable接口很像，都可以被另外一個線程執行，但是正如前面所說的，Runnable不會返回數據也不能拋出異常。

Future接口

Future接口代表異步計算的結果，通過Future接口提供的方法可以查看異步計算是否執行完成，或者等待執行結果並獲取執行結果，同時還可以取消執行。Future接口的定義如下:

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

• cancel():cancel()方法用來取消異步任務的執行。如果異步任務已經完成或者已經被取消，或者由於某些原因不能取消，則會返回false。如果任務還沒有被執行，則會返回true並且異步任務不會被執行。如果任務已經開始執行了但是還沒有執行完成，若mayInterruptIfRunning為true，則會立即中斷執行任務的線程並返回true，若mayInterruptIfRunning為false，則會返回true且不會中斷任務執行線程。
• isCanceled():判斷任務是否被取消，如果任務在結束(正常執行結束或者執行異常結束)前被取消則返回true，否則返回false。
• isDone():判斷任務是否已經完成，如果完成則返回true，否則返回false。需要注意的是：任務執行過程中發生異常、任務被取消也屬於任務已完成，也會返回true。
• get():獲取任務執行結果，如果任務還沒完成則會阻塞等待直到任務執行完成。如果任務被取消則會拋出CancellationException異常，如果任務執行過程發生異常則會拋出ExecutionException異常，如果阻塞等待過程中被中斷則會拋出InterruptedException異常。
• get(long timeout,Timeunit unit):帶超時時間的get()版本，如果阻塞等待過程中超時則會拋出TimeoutException異常。

FutureTask

Future只是一個接口，不能直接用來創建對象，FutureTask是Future的實現類，
FutureTask的繼承圖如下:

可以看到,FutureTask實現了RunnableFuture接口，則RunnableFuture接口繼承了Runnable接口和Future接口，所以FutureTask既能當做一個Runnable直接被Thread執行，也能作為Future用來得到Callable的計算結果。

使用

FutureTask一般配合ExecutorService來使用，也可以直接通過Thread來使用。

package com.beautyboss.slogen.callback;
 
import java.util.concurrent.*;
 
/**
 * Author  : Slogen
 * AddTime : 17/6/4
 * Email   : [e-mail]
 */
public class CallDemo {
 
    public static void main(String[] args) throws ExecutionException, InterruptedException {
 
        /**
         * 第一種方式:Future + ExecutorService
         * Task task = new Task();
         * ExecutorService service = Executors.newCachedThreadPool();
         * Future<Integer> future = service.submit(task1);
         * service.shutdown();
         */
 
 
        /**
         * 第二種方式: FutureTask + ExecutorService
         * ExecutorService executor = Executors.newCachedThreadPool();
         * Task task = new Task();
         * FutureTask<Integer> futureTask = new FutureTask<Integer>(task);
         * executor.submit(futureTask);
         * executor.shutdown();
         */
 
        /**
         * 第三種方式:FutureTask + Thread
         */
 
        // 2. 新建FutureTask,需要一個實現了Callable接口的類的實例作為構造函數參數
        FutureTask<Integer> futureTask = new FutureTask<Integer>(new Task());
        // 3. 新建Thread對象並啟動
        Thread thread = new Thread(futureTask);
        thread.setName("Task thread");
        thread.start();
 
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
 
        System.out.println("Thread [" + Thread.currentThread().getName() + "] is running");
 
        // 4. 調用isDone()判斷任務是否結束
        if(!futureTask.isDone()) {
            System.out.println("Task is not done");
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        int result = 0;
        try {
            // 5. 調用get()方法獲取任務結果,如果任務沒有執行完成則阻塞等待
            result = futureTask.get();
        } catch (Exception e) {
            e.printStackTrace();
        }
 
        System.out.println("result is " + result);
 
    }
 
    // 1. 繼承Callable接口,實現call()方法,泛型參數為要返回的類型
    static class Task  implements Callable<Integer> {
 
        @Override
        public Integer call() throws Exception {
            System.out.println("Thread [" + Thread.currentThread().getName() + "] is running");
            int result = 0;
            for(int i = 0; i < 100;++i) {
                result += i;
            }
 
            Thread.sleep(3000);
            return result;
        }
    }
}

第三部分:Why

構造函數

先從FutureTask的構造函數看起，FutureTask有兩個構造函數，其中一個如下：

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

這個構造函數會把傳入的Callable變量保存在this.callable字段中，該字段定義為private Callable<V> callable;用來保存底層的調用，在被執行完成以后會指向null,接着會初始化state字段為NEW。state字段用來保存FutureTask內部的任務執行狀態，一共有7中狀態，每種狀態及其對應的值如下：

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是volatile類型的，也就是說只要有任何一個線程修改了這個變量，那么其他所有的線程都會知道最新的值。

為了后面更好的分析FutureTask的實現，這里有必要解釋下各個狀態。

• NEW:表示是個新的任務或者還沒被執行完的任務。這是初始狀態。
• COMPLETING:任務已經執行完成或者執行任務的時候發生異常，但是任務執行結果或者異常原因還沒有保存到outcome字段(outcome字段用來保存任務執行結果，如果發生異常，則用來保存異常原因)的時候，狀態會從NEW變更到COMPLETING。但是這個狀態會時間會比較短，屬於中間狀態。
• NORMAL:任務已經執行完成並且任務執行結果已經保存到outcome字段，狀態會從COMPLETING轉換到NORMAL。這是一個最終態。
• EXCEPTIONAL:任務執行發生異常並且異常原因已經保存到outcome字段中后，狀態會從COMPLETING轉換到EXCEPTIONAL。這是一個最終態。
• CANCELLED:任務還沒開始執行或者已經開始執行但是還沒有執行完成的時候，用戶調用了cancel(false)方法取消任務且不中斷任務執行線程，這個時候狀態會從NEW轉化為CANCELLED狀態。這是一個最終態。
• INTERRUPTING: 任務還沒開始執行或者已經執行但是還沒有執行完成的時候，用戶調用了cancel(true)方法取消任務並且要中斷任務執行線程但是還沒有中斷任務執行線程之前，狀態會從NEW轉化為INTERRUPTING。這是一個中間狀態。
• INTERRUPTED:調用interrupt()中斷任務執行線程之后狀態會從INTERRUPTING轉換到INTERRUPTED。這是一個最終態。

有一點需要注意的是，所有值大於COMPLETING的狀態都表示任務已經執行完成(任務正常執行完成，任務執行異常或者任務被取消)。

各個狀態之間的可能轉換關系如下圖所示:

另外一個構造函數如下，

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

這個構造函數會把傳入的Runnable封裝成一個Callable對象保存在callable字段中，同時如果任務執行成功的話就會返回傳入的result。這種情況下如果不需要返回值的話可以傳入一個null。

順帶看下Executors.callable()這個方法，這個方法的功能是把Runnable轉換成Callable，代碼如下:

public static <T> Callable<T> callable(Runnable task, T result) {
    if (task == null)
        throw new NullPointerException();
    return new RunnableAdapter<T>(task, result);
}

可以看到這里采用的是適配器模式，調用RunnableAdapter<T>(task, result)方法來適配，實現如下:

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

這個適配器很簡單，就是簡單的實現了Callable接口，在call()實現中調用Runnable.run()方法，然后把傳入的result作為任務的結果返回。

在new了一個FutureTask對象之后，接下來就是在另一個線程中執行這個Task,無論是通過直接new一個Thread還是通過線程池，執行的都是run()方法，接下來就看看run()方法的實現。

run()

run()方法實現如下:

public void run() {
    // 1. 狀態如果不是NEW，說明任務或者已經執行過，或者已經被取消，直接返回
    // 2. 狀態如果是NEW，則嘗試把當前執行線程保存在runner字段中
    // 如果賦值失敗則直接返回
    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 {
                // 3. 執行任務
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                // 4. 任務異常
                setException(ex);
            }
            if (ran)
                // 4. 任務正常執行完畢
                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;
        // 5. 如果任務被中斷，執行中斷處理
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

run()方法首先會

1. 判斷當前任務的state是否等於NEW,如果不為NEW則說明任務或者已經執行過，或者已經被取消，直接返回。
2. 如果狀態為NEW則接着會通過unsafe類把任務執行線程引用CAS的保存在runner字段中，如果保存失敗，則直接返回。
3. 執行任務。
4. 如果任務執行發生異常，則調用setException()方法保存異常信息。setException()方法如下：

protected void setException(Throwable t) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = t;
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
        finishCompletion();
    }
}

在setException()方法中

• 首先會CAS的把當前的狀態從NEW變更為COMPLETING狀態。
• 把異常原因保存在outcome字段中，outcome字段用來保存任務執行結果或者異常原因。
• CAS的把當前任務狀態從COMPLETING變更為EXCEPTIONAL。這個狀態轉換對應着上圖中的二。
• 調用finishCompletion()。關於這個方法后面在分析。

5. 如果任務成功執行則調用set()方法設置執行結果，該方法實現如下:

protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        finishCompletion();
    }
}

這個方法跟上面分析的setException()差不多，

• 首先會CAS的把當前的狀態從NEW變更為COMPLETING狀態。
• 把任務執行結果保存在outcome字段中。
• CAS的把當前任務狀態從COMPLETING變更為NORMAL。這個狀態轉換對應着上圖中的一。
• 調用finishCompletion()。

發起任務線程跟執行任務線程通常情況下都不會是同一個線程，在任務執行線程執行任務的時候，任務發起線程可以查看任務執行狀態、獲取任務執行結果、取消任務等等操作，接下來分析下這些操作。

get()

任務發起線程可以調用get()方法來獲取任務執行結果，如果此時任務已經執行完畢則會直接返回任務結果，如果任務還沒執行完畢，則調用方會阻塞直到任務執行結束返回結果為止。get()方法實現如下:

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

get()方法實現比較簡單，會

1. 判斷任務當前的state <= COMPLETING是否成立。前面分析過，COMPLETING狀態是任務是否執行完成的臨界狀態。
2. 如果成立，表明任務還沒有結束(這里的結束包括任務正常執行完畢，任務執行異常，任務被取消)，則會調用awaitDone()進行阻塞等待。
3. 如果不成立表明任務已經結束，調用report()返回結果。

awaitDone()

當調用get()獲取任務結果但是任務還沒執行完成的時候，調用線程會調用awaitDone()方法進行阻塞等待，該方法定義如下:

private int awaitDone(boolean timed, long nanos)
        throws InterruptedException {
    // 計算等待截止時間
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    WaitNode q = null;
    boolean queued = false;
    for (;;) {
        // 1. 判斷阻塞線程是否被中斷,如果被中斷則在等待隊
        // 列中刪除該節點並拋出InterruptedException異常
        if (Thread.interrupted()) {
            removeWaiter(q);
            throw new InterruptedException();
        }
 
        // 2. 獲取當前狀態，如果狀態大於COMPLETING
        // 說明任務已經結束(要么正常結束，要么異常結束，要么被取消)
        // 則把thread顯示置空，並返回結果
        int s = state;
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            return s;
        }
        // 3. 如果狀態處於中間狀態COMPLETING
        // 表示任務已經結束但是任務執行線程還沒來得及給outcome賦值
        // 這個時候讓出執行權讓其他線程優先執行
        else if (s == COMPLETING) // cannot time out yet
            Thread.yield();
        // 4. 如果等待節點為空，則構造一個等待節點
        else if (q == null)
            q = new WaitNode();
        // 5. 如果還沒有入隊列，則把當前節點加入waiters首節點並替換原來waiters
        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;
            }
            // 6. 阻塞等待特定時間
            LockSupport.parkNanos(this, nanos);
        }
        else
            // 6. 阻塞等待直到被其他線程喚醒
            LockSupport.park(this);
    }
}

awaitDone()中有個死循環，每一次循環都會

1. 判斷調用get()的線程是否被其他線程中斷，如果是的話則在等待隊列中刪除對應節點然后拋出InterruptedException異常。
2. 獲取任務當前狀態，如果當前任務狀態大於COMPLETING則表示任務執行完成，則把thread字段置null並返回結果。
3. 如果任務處於COMPLETING狀態，則表示任務已經處理完成(正常執行完成或者執行出現異常)，但是執行結果或者異常原因還沒有保存到outcome字段中。這個時候調用線程讓出執行權讓其他線程優先執行。
4. 如果等待節點為空，則構造一個等待節點WaitNode。
5. 如果第四步中新建的節點還沒如隊列，則CAS的把該節點加入waiters隊列的首節點。
6. 阻塞等待。

假設當前state=NEW且waiters為NULL,也就是說還沒有任何一個線程調用get()獲取執行結果，這個時候有兩個線程threadA和threadB先后調用get()來獲取執行結果。再假設這兩個線程在加入阻塞隊列進行阻塞等待之前任務都沒有執行完成且threadA和threadB都沒有被中斷的情況下(因為如果threadA和threadB在進行阻塞等待結果之前任務就執行完成或線程本身被中斷的話，awaitDone()就執行結束返回了)，執行過程是這樣的，以threadA為例:

1. 第一輪for循環，執行的邏輯是q == null,所以這時候會新建一個節點q。第一輪循環結束。
2. 第二輪for循環，執行的邏輯是!queue，這個時候會把第一輪循環中生成的節點的netx指針指向waiters，然后CAS的把節點q替換waiters。也就是把新生成的節點添加到waiters鏈表的首節點。如果替換成功，queued=true。第二輪循環結束。
3. 第三輪for循環，進行阻塞等待。要么阻塞特定時間，要么一直阻塞知道被其他線程喚醒。

在threadA和threadB都阻塞等待之后的waiters結果如圖

cancel(boolean)

用戶可以調用cancel(boolean)方法取消任務的執行，cancel()實現如下:

public boolean cancel(boolean mayInterruptIfRunning) {
    // 1. 如果任務已經結束，則直接返回false
    if (state != NEW)
        return false;
    // 2. 如果需要中斷任務執行線程
    if (mayInterruptIfRunning) {
        // 2.1. 把任務狀態從NEW轉化到INTERRUPTING
        if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, INTERRUPTING))
            return false;
        Thread t = runner;
        // 2.2. 中斷任務執行線程
        if (t != null)
            t.interrupt();
        // 2.3. 修改狀態為INTERRUPTED
        UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); // final state
    }
    // 3. 如果不需要中斷任務執行線程，則直接把狀態從NEW轉化為CANCELLED
    else if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, CANCELLED))
        return false;
    // 4.
    finishCompletion();
    return true;
}

cancel()方法會做下面幾件事:

1 .判斷任務當前執行狀態，如果任務狀態不為NEW，則說明任務或者已經執行完成，或者執行異常，不能被取消，直接返回false表示執行失敗。

2. 判斷需要中斷任務執行線程，則

• 把任務狀態從NEW轉化到INTERRUPTING。這是個中間狀態。
• 中斷任務執行線程。
• 修改任務狀態為INTERRUPTED。這個轉換過程對應上圖中的四。

3. 如果不需要中斷任務執行線程，直接把任務狀態從NEW轉化為CANCELLED。如果轉化失敗則返回false表示取消失敗。這個轉換過程對應上圖中的四。

4. 調用finishCompletion()。

finishCompletion()

根據前面的分析，不管是任務執行異常還是任務正常執行完畢，或者取消任務，最后都會調用finishCompletion()方法，該方法實現如下:

private void finishCompletion() {
    // assert state > COMPLETING;
    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
}

這個方法的實現比較簡單，依次遍歷waiters鏈表，喚醒節點中的線程，然后把callable置空。

被喚醒的線程會各自從awaitDone()方法中的LockSupport.park*()阻塞中返回，然后會進行新一輪的循環。在新一輪的循環中會返回執行結果(或者更確切的說是返回任務的狀態)。

report()

report()方法用在get()方法中，作用是把不同的任務狀態映射成任務執行結果。實現如下：

private V report(int s) throws ExecutionException {
    Object x = outcome;
    // 1. 任務正常執行完成，返回任務執行結果
    if (s == NORMAL)
        return (V)x;
    // 2. 任務被取消，拋出CancellationException異常
    if (s >= CANCELLED)
        throw new CancellationException();
    // 3. 其他狀態，拋出執行異常ExecutionException
    throw new ExecutionException((Throwable)x);
}

 

映射關系如下圖所示:

如果任務處於NEW、COMPLETING和INTERRUPTING這三種狀態的時候是執行不到report()方法的，所以沒必要對這三種狀態進行轉換。

get(long,TimeUnit)

帶超時等待的獲取任務結果，實現如下:

public V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
    if (unit == null)
        throw new NullPointerException();
    int s = state;
    if (s <= COMPLETING &&
        // 如果awaitDone()超時返回之后任務還沒結束，則拋出異常
        (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
        throw new TimeoutException();
    return report(s);
}

跟get()不同點在於get(long,TimeUnit)會在awaitDone()超時返回之后拋出TimeoutException異常。

isCancelled()和isDone()

這兩個方法分別用來判斷任務是否被取消和任務是否執行完成，實現都比較簡單，代碼如下:

public boolean isCancelled() {
    return state >= CANCELLED;
}

public boolean isDone() {
    return state != NEW;
}

根據前面的分析，這兩個方法就很容易理解不用多做解釋了，O(∩_∩)O。

總結下，其實FutureTask的實現還是比較簡單的，當用戶實現Callable()接口定義好任務之后，把任務交給其他線程進行執行。FutureTask內部維護一個任務狀態，任何操作都是圍繞着這個狀態進行，並隨時更新任務狀態。任務發起者調用get*()獲取執行結果的時候，如果任務還沒有執行完畢，則會把自己放入阻塞隊列中然后進行阻塞等待。當任務執行完成之后，任務執行線程會依次喚醒阻塞等待的線程。調用cancel()取消任務的時候也只是簡單的修改任務狀態，如果需要中斷任務執行線程的話則調用Thread.interrupt()中斷任務執行線程。

第四部分:Other

有個值得關注的問題就是當任務還在執行的時候用戶調用cancel(true)方法能否真正讓任務停止執行呢？
在前面的分析中我們直到，當調用cancel(true)方法的時候，實際執行還是Thread.interrupt()方法，而interrupt()方法只是設置中斷標志位，如果被中斷的線程處於sleep()、wait()或者join()邏輯中則會拋出InterruptedException異常。

因此結論是:cancel(true)並不一定能夠停止正在執行的異步任務。