高并發(fā)之——兩種異步模型與深度解析Future接口

?

????????}

????????????}
????}?finally?{
????????????????????t.interrupt();
????if?(!(state?==?NEW?&&
????????????try?{
2.有返回結果的異步模型

盡管使用回調接口能夠獲取異步任務(wù)的結果，但是這種方式使用起來(lái)略顯復雜。在JDK中提供了可以直接返回異步結果的處理方案。最常用的就是使用Future接口或者其實(shí)現類(lèi)FutureTask來(lái)接收任務(wù)的返回結果。

（7）run()方法與runAndReset()方法

接下來(lái)，就是run()方法了，run()方法的源代碼如下所示。

????????else?if?(q?==?null)
?*?@version?1.0.0

????LockSupport.parkNanos(this,?nanos);

任務(wù)的執行類(lèi)是具體執行任務(wù)的類(lèi)，實(shí)現Runnable接口，在此類(lèi)中定義一個(gè)回調接口類(lèi)型的成員變量和一個(gè)String類(lèi)型的任務(wù)參數（模擬任務(wù)的參數），并在構造方法中注入回調接口和任務(wù)參數。在run方法中執行任務(wù)，任務(wù)完成后將任務(wù)的結果數據封裝成TaskResult對象，調用回調接口的方法將TaskResult對象傳遞到回調方法中。

????????executorService.execute(futureTask);


????????}
????????UNSAFE.putOrderedInt(this,?stateOffset,?EXCEPTIONAL);?//?final?state
????????????????setException(ex);
????????????set(result);

看來(lái)，這里還要看下report()方法啊，點(diǎn)進(jìn)去看下report()方法的實(shí)現，如下所示。

接下來(lái)，定義了其他幾個(gè)成員變量，如下所示.

????????//?runner?must?be?non-null?until?state?is?settled?to


????????while?(state?==?INTERRUPTING)
????????}

可以看到，handlePossibleCancellationInterrupt()方法的實(shí)現比較簡(jiǎn)單，當任務(wù)的狀態(tài)為INTERRUPTING時(shí)，使用while()循環(huán)，條件為當前任務(wù)狀態(tài)為INTERRUPTING，將當前線(xiàn)程占用的CPU資源釋放，也就是說(shuō)，當任務(wù)運行完成后，釋放線(xiàn)程所占用的資源。

????????return;

????????//?state?must?be?re-read?after?nulling?runner?to?prevent
public?boolean?cancel(boolean?mayInterruptIfRunning)?{
if?(!(state?==?NEW?&&
????????????????else?if?(!UNSAFE.compareAndSwapObject(this,?waitersOffset,
????V?get(long?timeout,?TimeUnit?unit)
????}
/**
public?boolean?isDone()?{

????????????????if?(next?==?null)

waiters：WaitNode類(lèi)型的變量，表示等待線(xiàn)程的堆棧，在FutureTask的實(shí)現中，會(huì )通過(guò)CAS結合此堆棧交換任務(wù)的運行狀態(tài)。

????????node.thread?=?null;
?????*?任務(wù)結果數據
??????????mayInterruptIfRunning???INTERRUPTING?:?CANCELLED)))
else?if?(timed)?{
if?(state?!=?NEW?||
????if?(q?!=?null)
????public?static?void?main(String[]?args)?throws?ExecutionException,?InterruptedException?{
NEW?->?CANCELLED
????this.callable?=?callable;
/**
????????new?Thread(taskExecutor).start();

可以看到，report()方法的實(shí)現比較簡(jiǎn)單，首先，將outcome數據賦值給x變量，接下來(lái)，主要是判斷接收到的任務(wù)狀態(tài)，如果狀態(tài)為NORMAL，則將x強轉為泛型類(lèi)型返回；當任務(wù)的狀態(tài)大于或者等于CANCELLED，也就是任務(wù)已經(jīng)取消，則拋出CancellationException異常，其他情況則拋出ExecutionException異常。

????????if?(s?>?COMPLETING)?{
????????}
????????System.out.println(future.get());
????????});
?*?@description?測試回調
?????*?任務(wù)狀態(tài)
????????????}?catch?(Throwable?ex)?{
?????????????????????????????????????null,?Thread.currentThread()))

本文有點(diǎn)長(cháng)，但是滿(mǎn)滿(mǎn)的干貨，以實(shí)際案例的形式分析了兩種異步模型，并從源碼角度深度解析Future接口和FutureTask類(lèi)，希望大家踏下心來(lái)，打開(kāi)你的IDE，跟著(zhù)文章看源碼，相信你一定收獲不??！

????????????public?String?call()?throws?Exception?{
????????return?taskResult;
????????//?leaked?interrupts

????try?{????//?in?case?call?to?interrupt?throws?exception

????????V?result;
protected?void?done()?{?}
這里需要注意一個(gè)細節：只有任務(wù)未啟動(dòng)，或者在完成之前被取消，才會(huì )返回true，表示任務(wù)已經(jīng)被成功取消。其他情況都會(huì )返回false。

????????????setException(ex);
}
????return?state?>=?CANCELLED;
????????????for?(;;)?{
protected?void?setException(Throwable?t)?{
?*?@author?binghe
????Thread.yield();

這里，發(fā)現變更任務(wù)狀態(tài)使用的是UNSAFE.putOrderedInt()方法，這個(gè)方法是個(gè)什么鬼呢？點(diǎn)進(jìn)去看一下，如下所示。

????????????????else?if?(pred?!=?null)?{

}
????runner?=?null;
private?static?final?long?stateOffset;
?*?@description?測試Future獲取異步結果

運行結果如下所示。

private?void?removeWaiter(WaitNode?node)?{

當任務(wù)完成時(shí)，直接返回任務(wù)的結果數據；當任務(wù)未完成時(shí)，等待任務(wù)完成并返回任務(wù)的結果數據。

finally?{
????int?s?=?state;
????WaitNode?q?=?null;

接下來(lái)，程序會(huì )進(jìn)入finally代碼塊中，如下所示。

????int?s?=?state;
?*?@version?1.0.0
????????Callable<V>?c?=?callable;
????????throw?new?CancellationException();
????}
????boolean?isDone();

沒(méi)參數的get()方法為當任務(wù)未運行完成時(shí)，會(huì )阻塞，直到返回任務(wù)結果。有參數的get()方法為當任務(wù)未運行完成，并且等待時(shí)間超出了超時(shí)時(shí)間，會(huì )TimeoutException異常。

一、兩種異步模型

在Java的并發(fā)編程中，大體上會(huì )分為兩種異步編程模型，一類(lèi)是直接以異步的形式來(lái)并行運行其他的任務(wù)，不需要返回任務(wù)的結果數據。一類(lèi)是以異步的形式運行其他任務(wù)，需要返回結果。

????try?{

?????*/
????????else?if?(!queued)
????????????throw?new?InterruptedException();
????????????????Thread?t?=?runner;
?????????????????????????????????null,?Thread.currentThread()))
接下來(lái)，cancel(boolean)方法會(huì )進(jìn)入finally代碼塊，如下所示。

????for?(WaitNode?q;?(q?=?waiters)?!=?null;)?{

runner：運行Callable的線(xiàn)程，運行期間會(huì )使用CAS保證線(xiàn)程安全，這里大家只需要知道CAS是Java保證線(xiàn)程安全的一種方式，后續文章中會(huì )深度分析CAS如何保證線(xiàn)程安全。

????}
????????????try?{
????????}?finally?{?//?final?state
return?report(s);

接下來(lái)，拆解awaitDone()方法。在awaitDone()方法中，最重要的就是for自旋循環(huán)，在循環(huán)中首先判斷當前線(xiàn)程是否被中斷，如果已經(jīng)被中斷，則調用removeWaiter()將當前線(xiàn)程從堆棧中移除，并且拋出InterruptedException異常，如下所示。

????????Callable<V>?c?=?callable;
????????ExecutorService?executorService?=?Executors.newSingleThreadExecutor();

可以看到，done()方法是一個(gè)空的方法體，交由子類(lèi)來(lái)實(shí)現具體的業(yè)務(wù)邏輯。
}
這個(gè)接口比較簡(jiǎn)單，run()方法就是運行任務(wù)時(shí)調用的方法。

????????????return?s;

結合線(xiàn)程池的使用示例如下。

????????????Thread.yield();
????????if?(mayInterruptIfRunning)?{

????????s?=?state;

當任務(wù)完成時(shí)，直接返回任務(wù)的結果數據；當任務(wù)未完成時(shí)，等待任務(wù)完成，并設置了超時(shí)等待時(shí)間。在超時(shí)時(shí)間內任務(wù)完成，則返回結果；否則，拋出TimeoutException異常。

????????if?(s?>=?INTERRUPTING)
????@Override
?*?@author?binghe
????????int?s?=?state;
?*?@author?binghe
????????q.thread?=?null;
????????????public?String?call()?throws?Exception?{

????????runner?=?null;
?*?@author?binghe

runAndReset()方法的邏輯與run()差不多，只是runAndReset()方法會(huì )在finally代碼塊中將任務(wù)狀態(tài)重置為NEW。runAndReset()方法的源代碼如下所示，就不重復說(shuō)明了。

}
public?void?run()?{
????????????}
????boolean?queued?=?false;
//?Unsafe?mechanics

定義任務(wù)結果數據的封裝類(lèi)

public?class?TaskHandler?implements?TaskCallable<TaskResult>?{

至于finishCompletion()方法，前面已經(jīng)分析過(guò)。

}
????????????????result?=?null;
else?if?(s?==?COMPLETING)
}

????????????(k.getDeclaredField("state"));
}
????????!UNSAFE.compareAndSwapObject(this,?runnerOffset,
????????if?(c?!=?null?&&?state?==?NEW)?{
????????//?runner?must?be?non-null?until?state?is?settled?to

get(long, TimeUnit)

????????????????????if?(pred.thread?==?null)?//?check?for?race
public?class?FutureTest?{
????????????????Thread?t?=?q.thread;
????}

判斷任務(wù)在完成之前是否被取消，如果在任務(wù)完成之前被取消，則返回true；否則，返回false。
????return?true;
????????????}
else?if?(q?==?null)
????????????????WaitNode?next?=?q.next;

運行結果如下所示。

兩個(gè)get()方法的主要邏輯差不多，一個(gè)沒(méi)有超時(shí)設置，一個(gè)有超時(shí)設置，這里說(shuō)一下主要邏輯。判斷任務(wù)的當前狀態(tài)是否小于或者等于COMPLETING，也就是說(shuō)，任務(wù)是NEW狀態(tài)或者COMPLETING，調用awaitDone()方法，看下awaitDone()方法的實(shí)現，如下所示。

?*?@version?1.0.0
}
????????runner?=?null;
import?java.util.concurrent.*;
????????????result?=?null;
1.無(wú)返回結果的異步模型

無(wú)返回結果的異步任務(wù)，可以直接將任務(wù)丟進(jìn)線(xiàn)程或線(xiàn)程池中運行，此時(shí)，無(wú)法直接獲得任務(wù)的執行結果數據，一種方式是可以使用回調方法來(lái)獲取任務(wù)的運行結果。

}
TaskResult{taskStatus=1,?taskMessage='測試回調任務(wù)',?taskResult='異步回調成功'}
????if?(s?==?NORMAL)
?*/
//TODO?拿到結果數據后進(jìn)一步處理
????try?{
?*?@description?定義回調接口
????????TaskResult?result?=?new?TaskResult();
????if?(callable?==?null)
private?void?finishCompletion()?{

到這里，整個(gè)大的框架算是完成了，接下來(lái)，就是測試看能否獲取到異步任務(wù)的結果了。

????????this.taskCallable?=?taskCallable;
????????????????return?"測試FutureTask獲取異步結果";
????????????handlePossibleCancellationInterrupt(s);
????????????result?=?c.call();
????????????????if?(t?!=?null)
}

????throw?new?ExecutionException((Throwable)x);
????Object?x?=?outcome;
????//?assert?state?>?COMPLETING;
}
????}
????}?finally?{

使用Future接口獲取異步結果

????????????????",?taskMessage='"?+?taskMessage?+?'\''?+
????}
?*/
????private?static?final?long?serialVersionUID?=?8678277072402730062L;
（2）構造方法

接下來(lái)，是FutureTask的兩個(gè)構造方法，比較簡(jiǎn)單，如下所示。

????V?get()?throws?InterruptedException,?ExecutionException;
二、深度解析Future接口 1.Future接口