【Java併發編程】Callable、Future和FutureTask的實現

時間 2019-11-11

標籤 Java併發編程 callable future futuretask 實現欄目 Java 简体版

原文原文鏈接

啓動線程執行任務，若是須要在任務執行完畢以後獲得任務執行結果，可使用從Java 1.5開始提供的Callable和Future
下面就分析一下Callable、Future以及FutureTask的具體實現及使用方法
源碼分析基於JDK 1.7java

1、Callable 與 Runnable

java.lang.Runnable是一個接口，只有一個run()方法node

public interface Runnable {
    public abstract void run();
}

run()方法的返回值是void，故在執行完任務後沒法返回任何結果less

Callable是java.util.concurrent包下的，也是一個接口，也只有一個call()方法，相似於java.lang.Runnable的run()方法，實現Callable接口的類和實現Runnable接口的類都是能夠被其它線程執行的任務異步

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

能夠看到call()方法是有返回值的，能夠將執行的結果返回ide

Callable和Runnable的區別：
一、Callable中定義的是call()方法，Runnable中定義的是run()方法
二、Callable中的call()方法能夠返回執行任務後的結果，Runnable中的run()方法沒法得到返回值
三、Callable中的call()方法定義了throws Exception拋出異常，拋出的異常能夠在主線程Future.get()時被主線程捕獲；Runnable中的run()方法沒有定義拋出異常，運行任務時發生異常時也會上拋，由於即便不加默認也會上拋RuntimeException，但異常沒法被主線程獲取
四、運行Callable任務能夠拿到一個Future對象表明異步運算的結果源碼分析

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

Future是java.util.concurrent包下的一個接口，表明着一個異步計算的結果，能夠經過get()獲取線程執行的返回值，cancel()取消任務執行，isCancelled()和isDone()得到任務執行的狀況this

boolean cancel(boolean mayInterruptIfRunning)

嘗試取消任務的執行，取消成功返回true，取消失敗返回false
mayInterruptIfRunning表示是否容許中斷正在執行的任務
一、若是任務還未開始，cancel返回true，且任務永遠不會被執行
二、若是任務正在執行，根據mayInterruptIfRunning的值判斷是否須要中斷執行中的任務，且若是mayInterruptIfRunning爲true，會調用中斷邏輯，返回true；若是mayInterruptIfRunning爲false，不會調用線程中斷，只是將任務取消
三、若是任務結束（多是正常完成、異常終止、被取消），返回false
四、若是cancel()操做返回true，後續調用isDone()、isCancelled()都返回true線程

boolean isCancelled()

表示任務是否被取消成功，若是在任務正常完成前被取消成功，則返回truecode

boolean isDone()

表示任務是否已經完成，則返回true，注意：正常完成、異常或取消操做都表明任務完成對象

V get() 和 V get(long timeout, TimeUnit unit)

get()用來獲取執行結果，這個方法會產生阻塞，會一直等到任務執行完畢才返回
get(long timeout, TimeUnit unit)用來獲取執行結果，若是在指定時間內還沒獲取到結果，會拋出TimeoutException

Future提供了三種功能：
一、獲取任務執行的結果
二、取消任務
三、判斷任務是否完成或是否取消

由於Future只是一個接口，因此是沒法直接用來建立對象使用的，所以就有了下面的FutureTask

3、FutureTask

public class FutureTask<V> implements RunnableFuture<V>

FutureTask實現了RunnableFuture接口，那麼RunnableFuture又是什麼呢？

public interface RunnableFuture<V> extends Runnable, Future<V> {
    /**
     * Sets this Future to the result of its computation
     * unless it has been cancelled.
     */
    void run();
}

RunnableFuture接口繼承了Runnable和Future，因此它既是一個可讓線程執行的Runnable任務，又是一個能夠獲取Callable返回值的Future

FutureTask的屬性

/** The run state of this task */
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; 

/** The underlying callable; nulled out after running */
private Callable<V> callable;

/** The result to return or exception to throw from get() */
private Object outcome;

/** The thread running the callable; CASed during run() */
private volatile Thread runner;

/** Treiber stack of waiting threads */
private volatile WaitNode waiters;

state 是任務的運行狀態

初始化時是NEW
任務終止的狀態有NORMAL（正常結束）、EXCEPTIONAL（異常結束）、CANCELLED（被取消）、INTERRUPTED（執行中被中斷），這些狀態是經過set()、setException、cancel()方法觸發的
COMPLETING 和 INTERRUPTING是兩個中間狀態，當正常結束設置outcome屬性前是COMPLETING，設置後變成NORMAL；當中斷運行中線程前是INTERRUPTING，調用thread.interrupt()後是INTERRUPTED

可能的狀態轉換：
NEW -> COMPLETING -> NORMAL
NEW -> COMPLETING -> EXCEPTIONAL
NEW -> CANCELLED
NEW -> INTERRUPTING -> INTERRUPTED

callable 是線程執行的有返回值的任務
outcome 是任務執行後的結果或異常
waiters 表示等待獲取結果的阻塞線程，鏈表結構，後等待線程的會排在鏈表前面

FutureTask的構造方法

FutureTask有兩個構造方法：
FutureTask(Callable callable)

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

構造方法參數是Callable定義的任務，並將state置爲NEW，只有當state爲NEW時，callable才能被執行

FutureTask(Runnable runnable, V result)

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

參數爲Runnable和帶泛型的result對象，因爲Runnable自己是沒有返回值的，故線程的執行結果經過result返回
能夠看到經過runnable和result封裝了個Callable，其實是new RunnableAdapter<T>(task, result)，這個Adapter適配器將Runnable和result轉換成Callable，並返回result

FutureTask.run()的實現

線程運行時真正執行的方法，Callable.call()會在其中執行，幷包含設置返回值或異常的邏輯

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

一、任務執行狀態不是NEW，直接返回；將runner屬性從null->當前線程不成功，直接返回
二、調用call()方法，調用成功，使用set()設置返回值
三、調用過程發生異常，使用setException()保存異常

set() 和 setException()

protected void set(V v) {
    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();
    }
}

set()和setException()的實現基本同樣，都是先將任務運行狀態從NEW->COMPLETING，分別設置返回值或異常給outcome，再將狀態分別置爲NORMAL和EXCEPTIONAL，最後調用finishCompletion()依次喚醒等待獲取結果的阻塞線程

finishCompletion()實現

/**
 * Removes and signals all waiting threads, invokes done(), and nulls out callable.
 */
private void finishCompletion() {
    // assert state > COMPLETING;
    for (WaitNode q; (q = waiters) != null;) {
        //將成員變量waiters置爲null
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            //循環喚醒WaitNode中的等待線程
            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
}

一、執行FutureTask類的get方法時，會把主線程封裝成WaitNode節點並保存在waiters鏈表中
二、FutureTask任務執行完成後，經過UNSAFE設置waiters的值爲null，並經過LockSupport.unpark方法依次喚醒等待獲取結果的線程

FutureTask.get()的實現

get()方法有兩個實現，一個是一直等待獲取結果，直到任務執行完；一個是等待指定時間，超時後任務還未完成會上拋TimeoutException

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

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

內部經過awaitDone()對主線程進行阻塞，具體實現以下：

/**
 * Awaits completion or aborts on interrupt or timeout.
 *
 * @param timed true if use timed waits
 * @param nanos time to wait, if timed
 * @return state upon completion
 */
private int awaitDone(boolean timed, long nanos)
    throws InterruptedException {
    final long deadline = timed ? System.nanoTime() + nanos : 0L; //截止時間
    WaitNode q = null;
    boolean queued = false;
    for (;;) {
        //若是主線程已經被中斷，removeWaiter()，並上拋InterruptedException
        //注意：Thread.interrupted()後會致使線程的中斷狀態爲false
        if (Thread.interrupted()) {
            removeWaiter(q); //線程被中斷的狀況下，從waiters鏈表中刪除q
            throw new InterruptedException();
        }

        int s = state;
        //若是任務已經完成（多是正常完成、異常、中斷），直接返回，即尚未開始等待，任務已經完成了
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            return s;
        }
        //若是任務正在完成，讓出CPU資源，等待state變成NORMAL或EXCEPTIONAL
        else if (s == COMPLETING) // cannot time out yet
            Thread.yield();
        //s<COMPLETING 且 尚未建立WaitNode
        else if (q == null)
            q = new WaitNode();
        //s<COMPLETING 且 已經建立WaitNode，但尚未入隊
        else if (!queued)
            /**
             * 一、將當前waiters賦值給q.next，即「q-->當前waiters」
             * 二、CAS，將waiters屬性，從「當前waiters-->q」
             * 因此後等待的會排在鏈表的前面，而任務完成時會從鏈表前面開始依次喚醒等待線程
             */
            queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                 q.next = waiters, q);
        //全部準備工做完成，判斷等待是否須要計時
        else if (timed) {
            nanos = deadline - System.nanoTime();
            //若是已經等待超時，remove當前WaiterNode
            if (nanos <= 0L) {
                removeWaiter(q); //等待超時的狀況下，從waiters鏈表中刪除q
                return state;
            }
            LockSupport.parkNanos(this, nanos); //掛起一段時間
        }
        else
            LockSupport.park(this); //一直掛起，等待喚醒
    }
}

一、判斷主線程是否被中斷，若是被中斷，將當前WaitNode節點從waiters鏈表中刪除，並上拋InterruptedException
二、若是任務已經完成（多是正常完成、異常、中斷），直接返回（即尚未開始等待，任務已經完成了，就返回了）
三、若是任務正在完成，讓出CPU資源，等待state變成NORMAL或EXCEPTIONAL
四、若是任務沒有被中斷，也沒有完成，new WaitNode()
五、若是任務沒有被中斷，也沒有完成，也建立了WaitNode，使用UNSAFE.CAS()操做將WaitNode加入waiters鏈表
六、全部準備工做完畢，經過LockSupport的park或parkNanos掛起線程

而WaitNode就是一個簡單的鏈表節點，記錄這等待的線程和下一個WaitNode

/**
 * 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
    WaitNode() { thread = Thread.currentThread(); }
}

FutureTask.cancel()的實現

public boolean cancel(boolean mayInterruptIfRunning) {
    if (state != NEW)
        return false;
    
    if (mayInterruptIfRunning) {
        if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, INTERRUPTING))
            return false;
        
        Thread t = runner;
        if (t != null)
            t.interrupt(); //中斷線程
        
        UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); // final state
    }
    else if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, CANCELLED))
        return false;
    
    finishCompletion();
    
    return true;
}

一、若是任務不是運行狀態，直接返回false失敗
二、若是mayInterruptIfRunning==true，中斷運行中的任務，使用CAS操做將狀態NEW-->INTERRUPTING，再調用runner.interrupt()，最後將狀態置爲INTERRUPTED
三、若是mayInterruptIfRunning==false，將任務置爲CANCELLED取消狀態
四、調用finishCompletion()依次喚醒等待獲取結果的線程，返回true取消成功

4、使用示例

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class TestFuture {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(1);
        Task task = new Task(); //callable任務
        Future<Integer> result = executor.submit(task);
        executor.shutdown();
         
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e1) {
            e1.printStackTrace();
        }
         
        System.out.println("主線程在執行任務");
         
        try {
            System.out.println("task運行結果："+result.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
         
        System.out.println("全部任務執行完畢");
    }
    
    static class Task implements Callable<Integer>{
        @Override
        public Integer call() throws Exception {
            System.out.println("子線程在進行計算");
            Thread.sleep(3000);
            int sum = 0;
            for(int i=0;i<100;i++)
                sum += i;
            return sum;
        }
    }
}

運行結果：

子線程在進行計算
主線程在執行任務
task運行結果：4950
全部任務執行完畢

若是隻是想控制在某些狀況下能夠將任務取消，可使用Future<?> future = executor.submit(runnable)，這樣返回結果確定爲null，但可使用future.cancel()取消任務執行

5、總結

一、有了Runnable，爲何還須要Callable，它們的區別是什麼？

Runnable和Callable都表示執行的任務，但不一樣的是Runnable.run()方法沒有返回值，Callable.call()有返回值
但其實線程在執行任務時仍是執行的Runnable.run()方法，因此在使用ThreadPoolExecutor.submit()時會將Callable封裝爲FutureTask，而FutureTask是Runnable和Future的實現類
因此在執行Callable的任務時，線程實際上是執行FutureTask這個Runnable的run()方法，其中封裝了調用Callable.call()並返回結果的邏輯

執行Runnable任務若是發生異常，主線程沒法知曉；而執行Callable任務若是發生異常，在Future.get()時會拋出java.util.concurrent.ExecutionException，其中封裝了真實異常

二、Future.get()是如何獲取線程返回值的？

首先得益於Callable.call()方法定義了返回值，提交Callable任務後，Callable會被封裝成FutureTask，其既能夠做爲Runnable被執行，也能夠做爲Future獲取返回值，FutureTask.run()方法會調用Callable.call()中的任務代碼
在任務執行完成前，若是主線程使用Future.get()，實際上是調用FutureTask.get()，其中會判斷任務狀態還沒有結束，將主線程加入waiters等待鏈表，並掛起主線程
待任務執行結束後，FutureTask會喚醒全部等待獲取返回值的線程，此時主線程的FutureTask.get()就會返回了

因此，主線程和運行線程是經過FutureTask做爲橋樑獲取線程返回值的

三、Future.cancel()真的能取消任務的執行嗎？

首先答案是「不必定」，根據JDK中的方法註釋「Attempts to cancel execution of this task」，即嘗試去取消執行的任務若是任務正在執行，且調用cancel()時參數mayInterruptIfRunning傳的是true，那麼會對執行線程調用interrupt()方法那麼問題就變成了interrupt()方法能中斷線程執行嗎？ interrupt()方法不會中斷正在運行的線程。這一方法實際上完成的是在線程受到阻塞時拋出一箇中斷信號，這樣線程就得以退出阻塞的狀態。更確切的說，若是線程被Object.wait()、Thread.join()、Thread.sleep()等阻塞，那麼它將接收到一箇中斷異常（InterruptedException），從而提前地終結被阻塞狀態。若是線程沒有被阻塞，調用interrupt()將不起做用那麼即便線程正在阻塞狀態，並拋出了InterruptedException，線程可否真的取消執行還要看代碼中是否捕獲了InterruptedException和有沒有作相應的對中斷標示的判斷邏輯