ExecutorService?Callable?Future多線程返回結果的原理是什么

本篇內容介紹了“ExecutorService Callable Future多線程返回結果的原理是什么”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

正文

在并發多線程場景下，存在需要獲取各線程的異步執行結果，這時，就可以通過ExecutorService線程池結合Callable、Future來實現。

簡單例子

public class ExecutorTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executor = Executors.newSingleThreadExecutor();
        Callable callable = new MyCallable();
        Future future = executor.submit(callable);
        System.out.println("打印線程池返回值：" + future.get());
    }
}
class MyCallable implements Callable<String>{
    @Override
    public String call() throws Exception {
        return "測試返回值";
    }
}

執行完成后，會打印出以下結果：

打印線程池返回值：測試返回值

可見，線程池執行完異步線程任務，我們是可以獲取到異步線程里的返回值。

那么，ExecutorService、Callable、Future實現有返回結果的多線程是如何實現的呢？

首先，我們需要創建一個實現函數式接口Callable的類，該Callable接口只定義了一個被泛型修飾的call方法，這意味著，需要返回什么類型的值可以由具體實現類來定義——

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

因此，我自定義了一個實現Callable接口的類，該類的重寫了call方法，我們在執行多線程時希望返回什么樣的結果，就可以在該重寫的call方法定義。

class MyCallable implements Callable<String>{
    @Override
    public String call() throws Exception {
        return "測試返回值";
    }
}

在自定義的MyCallable類中，我在call方法里設置一個很簡單的String返回值 “測試返回值”，這意味著，我是希望在線程池執行完異步線程任務時，可以返回“測試返回值”這個字符串給我。

接下來，我們就可以創建該MyCallable類的對象，然后通過executor.submit(callable)丟到線程池里，線程池里會利用空閑線程來幫我們執行一個異步線程任務。

ExecutorService executor = Executors.newSingleThreadExecutor();
Callable callable = new MyCallable();
Future future = executor.submit(callable);

值得注意一點是，若需要實現獲取線程返回值的效果，只能通過executor.submit(callable)去執行，而不能通過executor.execute(Runnable command)執行，因為executor.execute(Runnable command)只能傳入實現Runnable 接口的對象，但這類對象是不具備返回線程效果的功能。

進入到executor.submit(callable)底層，具體實現在AbstractExecutorService類中。可以看到，執行到submit方法內部時，會將我們傳進來的new MyCallable()對象作為參數傳入到newTaskFor(task)方法里——

public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}

這個newTaskFor(task)方法內部具體實現，是將new MyCallable()對象傳入構造器中，生成了一個FutureTask對象。

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    return new FutureTask<T>(callable);
}

這個FutureTask對象實現RunableFuture接口，這個RunableFuture接口又繼承了Runnable，說明FutureTask類內部會實現一個run方法，然后本身就可以當做一個Runnable線程任務，借助線程Thread(new FutureTask(.....)).start()方式開啟一個新線程，去異步執行其內部實現的run方法邏輯。

異步執行內部實現run方法邏輯

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

分析到這里，可以知道FutureTask的核心方法一定是run方法，線程執行start方法后，最后會去調用FutureTask的run方法。在講解這個run方法前，我們先去看一下創建FutureTask的初始化構造方法底層邏輯new FutureTask(callable)

public class FutureTask<V> implements RunnableFuture<V> {
private Callable<V> callable;
......//省略其余源碼
public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    //通過構造方法初始化Callable<V> callable賦值
    this.callable = callable;
    this.state = NEW;       // ensure visibility of callable
}
......//省略其余源碼
}

可以看到，FutureTask(Callable callable)構造器，主要是將我們先前創建的new MyCallable()對象傳進來，賦值給FutureTask內部定義的Callable callable引用，實現子類對象指向父類引用。這一點很關鍵，這就意味著，在初始化創建FutureTask對象后，我們是可以通過callable.call()來調用我們自定義設置可以返回“測試返回值”的call方法，這不就是我們希望在異步線程執行完后能夠返回的值嗎？

我們不妨猜測一下整體返數主流程，在Thread(new FutureTask(.....)).start()開啟一個線程后，當線程獲得了CPU時間片，就會去執行FutureTask對象里的run方法，這時run方法里可以通過callable.call()調用到我們自定義的MyCallable#call()方法，進而得到方法返回值 “測試返回值”——到這一步，只需要將這個返回值賦值給FutureTask里某個定義的對象屬性，那么，在主線程在通過獲取FutureTask里被賦值的X對象屬性值，不就可以拿到返回字符串值 “測試返回值”了嗎？

實現上，主體流程確實是這樣，只不過忽略了一些細節而已。

FutureTask的run方法

public void run() {
    //如果狀態不是NEW或者設置runner為當前線程時，說明FutureTask任務已經取消，無法繼續執行
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    try {
        //在該文中，callable被賦值為指向我們定義的new MyCallable()對象引用
        Callable<V> c = callable;
        if (c != null && state == NEW) {
            V result;
            boolean ran;
            try {
                //c.call最后會調用new MyCallable()的call()方法，得到字符串返回值“測試返回值”給result
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                setException(ex);
            }
            //正常執行完c.call()方法時，ran值為true,說明會執行set(result)方法。
            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);
    }
}

根據以上源碼簡單分析，可以看到run方法當中，最終確實會執行new MyCallable()的call()方法，得到字符串返回值“測試返回值”給result，然后執行set(result)方法，根據set方法名就不難猜出，這是一個會賦值給某個字段的方法。

這里分析會忽略一些狀態值的講解，這塊會包括線程的取消、終止等內容，后面我會出一片專門針對FutureTask源碼分析的文章再介紹，本文主要還是介紹異步線程返回結果的主要原理。

set(result)方法

沿著以上分析，追蹤至set(result)方法里——

protected void set(V v) {
    //通過CAS原子操作，將運行的線程設置為COMPLETING，說明線程已經執行完成中
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        //若CAS原子比較賦值成功，說明線程可以被正常執行完成的話，然后將result結果值賦值給outcome
        outcome = v;
        //線程正常完成結束
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        finishCompletion();
    }
}

這個方法的主要是，若該線程執行能夠正常完成話，就將得到的返回值賦值給outcome，這個outcome是FutureTask的一個Object變量——

private Object outcome;

至此，就完成了流程的這一步——

最后，就是執行主線程的根據ftask.get()獲取執行完成的值，這個get可以設置超時時間，例如 ftask.get(2,TimeUnit.SECONDS)表示超過2秒還沒有獲取到線程返回值的話，就直接結束該get方法，繼續主線程往下執行。

System.out.println("打印線程池返回值：" + ftask.get(2,TimeUnit.SECONDS));

進入到get方法，可以看到當狀態在s <= COMPLETING時，表示任務還沒有執行完，就會去執行awaitDone(false, 0L)方法，這個方法表示，將一直做死循環等待線程執行完成，才會跳出等待循環繼續往下走。若設置了超時時間，例如ftask.get(2,TimeUnit.SECONDS))，就會在awaitDone方法循環至2秒，在2秒內發現線程狀態被設置為正常完成時，就會跳出循環，若2秒后線程沒有執行完成，也會強制跳出循環了，但這種情況將無法獲取到線程結果值。

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        //循環等待線程執行狀態
        s = awaitDone(false, 0L);
    return report(s);
}

最后就是report(s)方法，可以看到outcome值最終賦值給Object x，若s==NORMAL表示線程任務已經正常完成結束，就可以根據我們定義的類型進行泛型轉換返回，我們定義的是String字符串類型，故而會返回字符串值，也就是 “測試返回值”。

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

你看，最后就能獲取到了異步線程執行的結果返回給main主線程&mdash;&mdash;

以上就是執行線程任務run方法后，如何將線程任務結果返回給主線程，其實，還少一個地方補充，就是如何將FutureTask任務丟給線程執行，我們這里用到了線程池， 但是execute(ftask)底層同樣是使用一個了線程通過執行start方法開啟一個線程，這個新運行的線程最終會執行FutureTask的run方法。

public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}

可以簡單優化下，直接用一個線程演示該案例，這樣看著更好理解些，當時，生產上是不會有這樣直接用一個線程來執行的，更多是通過原生線程池&mdash;&mdash;

public static void main(String[] args) throws Exception{
    Callable callable = new MyCallable();
    RunnableFuture<String> ftask = new FutureTask<String>(callable);
    new Thread(ftask).start();
    System.out.println("打印線程池返回值：" + ftask.get());
}

“ExecutorService Callable Future多線程返回結果的原理是什么”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！