Java8CompletableFuture詳解剖析

1、Java8 CompletableFuture詳解Java 8 來了， 是時候?qū)W一下新的東西了。 Java 7 和 Java 6 只不過是稍作修改的版本， 而 Java 8 將會發(fā)生重大的改進(jìn)?；蛟S是 Java 8太大了吧？今天我會給你徹底地解釋JDK 8中的新的抽象 CompletableFuture 。眾所周知， Java 8不到一年就會發(fā)布， 因此這篇文章是基于 JDK 8 build 88 with lambda support 的。 CompletableFuture extends Future 提供了方法，一元操作符 和促進(jìn)異步性以及事件驅(qū)動編程模型，它并不止步于舊版本的Java

2、 中。如果你打開 JavaDocof CompletableFuture 你一定會感到震驚。大約有五十種方法(！ )，而且它們中的一些非常 有意思而且不好理解，例如：復(fù)制代碼 代碼如下 :public CompletableFuture thenCombineAsync(CompletableFuture other,BiFunction fn,Executor executor)不必?fù)?dān)心，繼續(xù)讀下去。 CompletableFuture 收集了所有 ListenableFuture in Guava 和 SettableFuture 的特征。此外，內(nèi)置的 lambda 表達(dá)式使它更接近于 S

3、cala/Akka futures 。這聽 起來好得令人難以置信，但是請繼續(xù)讀下去。 CompletableFuture 有兩個主要的方面優(yōu)于 ol 中 的 Future 異 步 回 調(diào) / 轉(zhuǎn) 換 ， 這 能 使 得 從 任 何 時 刻 的 任 何 線 程 都 可 以 設(shè) 置 CompletableFuture 的值。一、提取、修改包裝的值通常 futures 代表其它線程中運(yùn)行的代碼， 但事實并非總是如此。 有時你想要創(chuàng)造一個 Future 來表示你知道將會發(fā)生什么，例如 JMS message arrival。所以你有 Future 但是未來并沒有潛 在的異步工作。你只是想在未來JMS消

4、息到達(dá)時簡單地完成(解決) ，這是由一個事件驅(qū)動的。在這種情況下，你可以簡單地創(chuàng)建 CompletableFuture 來返還給你的客戶端，只要你認(rèn) 為你的結(jié)果是可用的，僅僅通過 complete() 就能解鎖所有等待 Future 的客戶端。首先你可以簡單地創(chuàng)建新的 CompletableFuture 并且給你的客戶端： 復(fù)制代碼 代碼如下 :public CompletableFuture ask() final CompletableFuture future = new CompletableFuture(); /.return future;注意這個 future 和 Callabl

5、e 沒有任何聯(lián)系，沒有線程池也不是異步工作。如果現(xiàn)在客戶端代 碼調(diào)用 ask().get() 它將永遠(yuǎn)阻塞。如果寄存器完成回調(diào)，它們就永遠(yuǎn)不會生效了。所以關(guān)鍵 是什么？現(xiàn)在你可以說：復(fù)制代碼 代碼如下 :plete(42)此時此刻所有客戶端 Future.get() 將得到字符串的結(jié)果，同時完成回調(diào)以后將會立即生效。 當(dāng)你想代表 Future 的任務(wù)時是非常方便的，而且沒有必要去計算一些執(zhí)行線程的任務(wù)上。CompletableFplete() 只能調(diào)用一次，后續(xù)調(diào)用將被忽略。但也有一個后門叫做 CompletableFuture.obtrudeValue( )覆蓋一個新 Future 之前的價

6、值，請小心使用。有時你想要看到信號發(fā)生故障的情況，如你所知 Future 對象可以處理它所包含的結(jié)果或異 常。如果你想進(jìn)一步傳遞一些異常，可以用 CompletableFpleteExceptionally(ex) ( 或 者用 obtrudeException(ex) 這樣更強(qiáng)大的方法覆蓋前面的異常 )。 completeExceptionally() 也能 解鎖所有等待的客戶端， 但這一次從 get() 拋出異常。說到 get()，也有 CompletableFuture.join() 方法在錯誤處理方面有著細(xì)微的變動。但總體上，它們都是一樣的。最后也有 CompletableFuture

7、.getNow(valueIfAbsent) 方法沒有阻塞但是如果 Future 還沒完成將返回默認(rèn) 值，這使得當(dāng)構(gòu)建那種我們不想等太久的健壯系統(tǒng)時非常有用。最后 static 的方法是用 completedFuture(value) 來返回已經(jīng)完成 Future 的對象， 當(dāng)測試或者寫 一些適配器層時可能非常有用。二、創(chuàng)造和獲取 CompletableFuture好了，那么手動地創(chuàng)建 CompletableFuture 是我們唯一的選擇嗎？不一定。就像一般的 Futures ，我們可以關(guān)聯(lián)存在的任務(wù)，同時 CompletableFuture 使用工廠方法：復(fù)制代碼 代碼如下 : static

8、 CompletableFuture supplyAsync(Supplier supplier);static CompletableFuture supplyAsync(Supplier supplier, Executor executor);static CompletableFuture runAsync(Runnable runnable);static CompletableFuture runAsync(Runnable runnable, Executor executor);無參方法 Executor 是以 Async 結(jié)尾同時將會使用 ForkJoinPmonPool()

9、( 全局的，在 JDK8 中介紹的通用池) ，這適用于 CompletableFuture 類中的大多數(shù)的方法。 runAsync()易于 理解，注意它需要 Runnable，因此它返回 CompletableFuture 作為 Runnable 不返回任何值。如果你需要處理異步操作并返回結(jié)果，使用Supplier:復(fù)制代碼 代碼如下 :final CompletableFuture future = CompletableFuture.supplyAsync(new Supplier() Overridepublic String get() /.long running.return 42

10、;, executor);但是別忘了， Java 8里面還有 lambdas 表達(dá)式呢！復(fù)制代碼 代碼如下 :finalCompletableFuture future = CompletableFuture.supplyAsync() - /.long running.return 42;, executor);或者：復(fù)制代碼 代碼如下 :final CompletableFuture future =CompletableFuture.supplyAsync() - longRunningTask(params), executor);雖然這篇文章不是關(guān)于 Lambda 的，但是我會相當(dāng)頻

11、繁地使用 lambda 表達(dá)式。三、轉(zhuǎn)換和作用于 CompletableFuture(thenApply)我 說 過 CompletableFuture 優(yōu) 于 Future 但 是 你 還 不 知 道 為 什 么 嗎 ？ 簡 單 說 ， 因 為 CompletableFuture 是一個原子也是一個因子。我說的這句話沒什么幫助嗎？ Scala 和 JavaScript 都允許 future 完成時允許注冊異步回調(diào)，直到它準(zhǔn)備好我們才要等待和阻止它。 我們可以簡單地說： 運(yùn)行這個函數(shù)時就出現(xiàn)了結(jié)果。此外，我們可以疊加這些功能，把多個 future 組合在一起等。 例如如果我們從 String

12、轉(zhuǎn)為 Integer ，我們可以轉(zhuǎn)為在不關(guān)聯(lián)的前提下 從 CompletableFuture 到 CompletableFutureInteger 。這是通過 thenApply() 的方法：復(fù)制代碼 代碼如下 : CompletableFuture thenApply(Function fn); CompletableFuture thenApplyAsync(Function fn); CompletableFuture thenApplyAsync(Function fn, Executor executor);如前所述 .Async 版本提供對 CompletableFuture 的大

13、多數(shù)操作，因此我將在后面的部分 中跳過它們。 記住， 第一個方法將在 future 完成的相同線程中調(diào)用該方法， 而剩下的兩個將 在不同的線程池中異步地調(diào)用它。讓我們來看看 thenApply() 的工作流程： CompletableFuture f1 = /.CompletableFuture f2 = f1.thenApply(Integer:parseInt); CompletableFuture f3 = f2.thenApply(r - r * r * Math.PI);或在一個聲明中：復(fù)制代碼 代碼如下 :CompletableFuture f3 =f1.thenApply(Int

14、eger:parseInt).thenApply(r - r * r * Math.PI);這里，你會看到一個序列的轉(zhuǎn)換，從 String 到 Integer 再到 Double 。但最重要的是，這些轉(zhuǎn) 換既不立即執(zhí)行也不停止。這些轉(zhuǎn)換既不立即執(zhí)行也不停止。他們只是記得，當(dāng)原始 f1 完 成他們所執(zhí)行的程序。 如果某些轉(zhuǎn)換非常耗時， 你可以提供你自己的 Executor 來異步地運(yùn)行他們。注意 ,此操作相當(dāng)于 Scala中的一元 map 。四、運(yùn)行完成的代碼( thenAccept/thenRun )復(fù)制代碼 代碼如下 :CompletableFuture thenAccept(Consume

15、r block);CompletableFuture thenRun(Runnable action);在 future 的管道里有兩種典型的 “最終” 階段方法。 他們在你使用 future 的值的時候做好準(zhǔn) 備，當(dāng) thenAccept() 提供最終的值時， thenRun 執(zhí)行 Runnable，這甚至沒有方法去計算值。 例如：復(fù)制代碼 代碼如下 :future.thenAcceptAsync(dbl - log.debug(Result: , dbl), executor);log.debug(Continuing);Async變量也可用兩種方法，隱式和顯式執(zhí)行器，我不會過多強(qiáng)調(diào)這個方

16、法。 thenAccept()/thenRun() 方法并沒有發(fā)生阻塞 (即使沒有明確的 executor) 。它們像一個事件偵 聽器 /處理程序，你連接到一個 future 時，這將執(zhí)行一段時間。 ” Continuing ”消息將立即出 現(xiàn)，盡管 future 甚至沒有完成。五、單個 CompletableFuture 的錯誤處理 到目前為止，我們只討論計算的結(jié)果。那么異常呢？我們可以異步地處理它們嗎？當(dāng)然！復(fù)制代碼 代碼如下 :CompletableFuture safe = future.exceptionally(ex - We have a problem: + ex.getMes

17、sage();exceptionally() 接受一個函數(shù)時， 將調(diào)用原始 future 來拋出一個異常。 我們會有機(jī)會將此異常 轉(zhuǎn)換為和 Future 類型的兼容的一些值來進(jìn)行恢復(fù)。 safe 進(jìn)一步的轉(zhuǎn)換將不再產(chǎn)生一個異常 而是從提供功能的函數(shù)返回一個 String 值。一個更加靈活的方法是 handle() 接受一個函數(shù)，它接收正確的結(jié)果或異常： 復(fù)制代碼 代碼如下 :CompletableFuture safe = future.handle(ok, ex) - if (ok != null) return Integer.parseInt(ok); else log.warn(Pro

18、blem, ex);return -1;); handle()總是被調(diào)用，結(jié)果和異常都非空，這是個一站式全方位的策略。六、一起結(jié)合兩個 CompletableFuture異步處理過程之一的 CompletableFuture 非常不錯但是當(dāng)多個這樣的 futures 以各種方式組合 在一起時確實顯示了它的強(qiáng)大。七、結(jié)合(鏈接)這兩個 futures ( thenCompose() ) 有時你想運(yùn)行一些 future 的值(當(dāng)它準(zhǔn)備好了)，但這個函數(shù)也返回了 future 。CompletableFuture 足夠靈活地明白我們的函數(shù)結(jié)果現(xiàn)在應(yīng)該作為頂級的 future ，對比 Completa

19、bleFuture 。方法 thenCompose() 相當(dāng)于 Scala 的 flatMap ： 復(fù)制代碼 代碼如下 : CompletableFuture thenCompose(Function? super T,CompletableFuture fn); Async 變化也是可用的，在下面的事例中，仔細(xì)觀察thenApply()(map) 和 thenCompose()( flatMap )的類型和差異，當(dāng)應(yīng)用 calculateRelevance() 方法返回 CompletableFuture ：復(fù)制代碼 代碼如下 :CompletableFuture docFuture = /

20、.CompletableFutureCompletableFuture f =docFuture.thenApply(this:calculateRelevance);CompletableFuture relevanceFuture = docFuture.thenCompose(this:calculateRelevance);/.private CompletableFuture calculateRelevance(Document doc)/.thenCompose() 是一個重要的方法允許構(gòu)建健壯的和異步的管道，沒有阻塞和等待的中間步 驟。八、兩個 futures 的轉(zhuǎn)換值 (th

21、enCombine()當(dāng) thenCompose() 用于鏈接一個 future 時依賴另一個 thenCombine ，當(dāng)他們都完成之后就結(jié) 合兩個獨立的 futures ：復(fù)制代碼 代碼如下 : CompletableFuture thenCombine(CompletableFuture other, BiFunction fn) Async 變量也是可用的，假設(shè)你有兩個 CompletableFuture ，一個加載 Customer 另一個加 載最近的 Shop。他們彼此完全獨立， 但是當(dāng)他們完成時， 您想要使用它們的值來計算 Route。 這是一個可剝奪的例子：復(fù)制代碼 代碼如下

22、:CompletableFuture customerFuture = loadCustomerDetails(123);CompletableFuture shopFuture = closestShop();CompletableFuture routeFuture = customerFuture.thenCombine(shopFuture, (cust, shop) - findRoute(cust, shop);/.private Route findRoute(Customer customer, Shop shop) /.請注意，在 Java 8 中可以用 (cust, sho

23、p) - findRoute(cust, shop) 簡單地代替 this:findRoute 方 法的引用：復(fù)制代碼 代碼如下 :customerFuture.thenCombine(shopFuture, this:findRoute);你也知道，我們有 customerFuture 和 shopFuture 。那么 routeFuture 包裝它們?nèi)缓蟆暗却?它們完成。當(dāng)他們準(zhǔn)備好了，它會運(yùn)行我們提供的函數(shù)來結(jié)合所有的結(jié)果 (findRoute() 。當(dāng) 兩個基本的 futures 完成并且 findRoute() 也完成時，這樣 routeFuture 將會完成。九、等待所有的 Co

24、mpletableFutures 完成如果不是產(chǎn)生新的 CompletableFuture 連接這兩個結(jié)果，我們只是希望當(dāng)完成時得到通知， 我們可以使用 thenAcceptBoth()/runAfterBoth() 系列的方法， ( Async 變量也是可用的) 。它 們的工作方式與 thenAccept() 和 thenRun() 類似，但是是等待兩個 futures 而不是一個： 復(fù)制代碼 代碼如下 : CompletableFuture thenAcceptBoth(CompletableFuture other, BiConsumer block)CompletableFuture

25、runAfterBoth(CompletableFuture other, Runnable action)想象一下上面的例子，這不是產(chǎn)生新的 CompletableFuture ，你只是想要立刻發(fā)送一些事件 或刷新 GUI。這可以很容易地實現(xiàn)： thenAcceptBoth():復(fù)制代碼 代碼如下 :customerFuture.thenAcceptBoth(shopFuture, (cust, shop) - final Route route = findRoute(cust, shop);/refresh GUI with route);我希望我是錯的， 但也許有些人會問自己一個問題：

26、 為什么我不能簡單地阻塞這兩個 futures 呢？ 就像：復(fù)制代碼 代碼如下 :Future customerFuture = loadCustomerDetails(123); Future shopFuture = closestShop(); findRoute(customerFuture.get(), shopFuture.get();好了，你當(dāng)然可以這么做。但是最關(guān)鍵的一點是 CompletableFuture 是允許異步的，它是事 件驅(qū)動的編程模型而不是阻塞并急切地等待著結(jié)果。 所以在功能上， 上面兩部分代碼是等價 的，但后者沒有必要占用一個線程來執(zhí)行。十、等待第一個 Comp

27、letableFuture 來完成任務(wù)另一個有趣的事是 CompletableFutureAPI 可以等待第一個（與所有相反）完成的 future 。當(dāng) 你有兩個相同類型任務(wù)的結(jié)果時就顯得非常方便， 你只要關(guān)心響應(yīng)時間就行了， 沒有哪個任 務(wù)是優(yōu)先的。 API方法 （Async變量也是可用的） ：復(fù)制代碼 代碼如下 :CompletableFuture acceptEither(CompletableFuture other, Consumer block)CompletableFuture runAfterEither(CompletableFuture other, Runnable ac

28、tion)作為一個例子， 你有兩個系統(tǒng)可以集成。 一個具有較小的平均響應(yīng)時間但是擁有高的標(biāo)準(zhǔn)差， 另一個一般情況下較慢， 但是更加容易預(yù)測。 為了兩全其美（性能和可預(yù)測性）你可以在同 一時間調(diào)用兩個系統(tǒng)并等著誰先完成。 通常這會是第一個系統(tǒng)， 但是在進(jìn)度變得緩慢時， 第 二個系統(tǒng)就可以在可接受的時間內(nèi)完成：復(fù)制代碼 代碼如下 :CompletableFuture fast = fetchFast();CompletableFuture predictable = fetchPredictably(); fast.acceptEither(predictable, s - System.out.

29、println(Result: + s););s 代表了從 fetchFast()或是 fetchPredictably() 得到的 String 。我們不必知道也無需關(guān)心。十一、完整地轉(zhuǎn)換第一個系統(tǒng)applyToEither() 算是 acceptEither() 的前輩了。 當(dāng)兩個 futures 快要完成時，后者只是簡單地調(diào) 用一些代碼片段， applyToEither()將會返回一個新的 future 。當(dāng)這兩個最初的 futures 完成時， 新的 future 也會完成。 API 有點類似于 ( Async 變量也是可用的 )：復(fù)制代碼 代碼如下 : CompletableFuture applyToEither(CompletableFuture other, Function fn)這個額外的 fn 功能在第一個 future 被調(diào)用時能完成。我不確定這個專業(yè)化方法的目的是什 么，畢竟一個人可以簡單地使用： fast.applyToEither(predicta