ExecutorService Callable Future多執行緒返回結果原理解析

2022-09-28 14:00:15

正文

在並行多執行緒場景下，存在需要獲取各執行緒的非同步執行結果，這時，就可以通過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 &lt;= 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主執行緒——

以上就是執行執行緒任務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;
}

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

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多執行緒返回結果原理解析的詳細內容，更多關於ExecutorService Callable Future多執行緒的資料請關注it145.com其它相關文章！

ExecutorService Callable Future多執行緒返回結果原理解析

目錄

正文

簡單例子

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

FutureTask的run方法

set(result)方法

熱門文章