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方法邏輯。
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
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
我們不妨猜測一下整體返數主流程，在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)方法裡——
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主執行緒——
以上就是執行執行緒任務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());
}