Java8CompletableFuture詳解剖析

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

2、 中。如果你打開 JavaDocof CompletableFuture 你一定會(huì)感到震驚。大約有五十種方法(！ )，而且它們中的一些非常 有意思而且不好理解，例如：復(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 。這聽 起來好得令人難以置信，但是請(qǐng)繼續(xù)讀下去。 CompletableFuture 有兩個(gè)主要的方面優(yōu)于 ol 中 的 Future 異 步 回 調(diào) / 轉(zhuǎn) 換 ， 這 能 使 得 從 任 何 時(shí) 刻 的 任 何 線 程 都 可 以 設(shè) 置 CompletableFuture 的值。一、提取、修改包裝的值通常 futures 代表其它線程中運(yùn)行的代碼， 但事實(shí)并非總是如此。 有時(shí)你想要?jiǎng)?chuàng)造一個(gè) Future 來表示你知道將會(huì)發(fā)生什么，例如 JMS message arrival。所以你有 Future 但是未來并沒有潛 在的異步工作。你只是想在未來JMS消

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

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

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

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

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

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

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

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

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

17、sage();exceptionally() 接受一個(gè)函數(shù)時(shí)， 將調(diào)用原始 future 來拋出一個(gè)異常。 我們會(huì)有機(jī)會(huì)將此異常 轉(zhuǎn)換為和 Future 類型的兼容的一些值來進(jìn)行恢復(fù)。 safe 進(jìn)一步的轉(zhuǎn)換將不再產(chǎn)生一個(gè)異常 而是從提供功能的函數(shù)返回一個(gè) String 值。一個(gè)更加靈活的方法是 handle() 接受一個(gè)函數(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é)果和異常都非空，這是個(gè)一站式全方位的策略。六、一起結(jié)合兩個(gè) CompletableFuture異步處理過程之一的 CompletableFuture 非常不錯(cuò)但是當(dāng)多個(gè)這樣的 futures 以各種方式組合 在一起時(shí)確實(shí)顯示了它的強(qiáng)大。七、結(jié)合(鏈接)這兩個(gè) futures ( thenCompose() ) 有時(shí)你想運(yùn)行一些 future 的值(當(dāng)它準(zhǔn)備好了)，但這個(gè)函數(shù)也返回了 future 。CompletableFuture 足夠靈活地明白我們的函數(shù)結(jié)果現(xiàn)在應(yīng)該作為頂級(jí)的 future ，對(duì)比 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è)重要的方法允許構(gòu)建健壯的和異步的管道，沒有阻塞和等待的中間步 驟。八、兩個(gè) futures 的轉(zhuǎn)換值 (th

21、enCombine()當(dāng) thenCompose() 用于鏈接一個(gè) future 時(shí)依賴另一個(gè) thenCombine ，當(dāng)他們都完成之后就結(jié) 合兩個(gè)獨(dú)立的 futures ：復(fù)制代碼 代碼如下 : CompletableFuture thenCombine(CompletableFuture other, BiFunction fn) Async 變量也是可用的，假設(shè)你有兩個(gè) CompletableFuture ，一個(gè)加載 Customer 另一個(gè)加 載最近的 Shop。他們彼此完全獨(dú)立， 但是當(dāng)他們完成時(shí)， 您想要使用它們的值來計(jì)算 Route。 這是一個(gè)可剝奪的例子：復(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) /.請(qǐng)注意，在 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)備好了，它會(huì)運(yùn)行我們提供的函數(shù)來結(jié)合所有的結(jié)果 (findRoute() 。當(dāng) 兩個(gè)基本的 futures 完成并且 findRoute() 也完成時(shí)，這樣 routeFuture 將會(huì)完成。九、等待所有的 Co

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

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

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

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

28、tion)作為一個(gè)例子， 你有兩個(gè)系統(tǒng)可以集成。 一個(gè)具有較小的平均響應(yīng)時(shí)間但是擁有高的標(biāo)準(zhǔn)差， 另一個(gè)一般情況下較慢， 但是更加容易預(yù)測(cè)。 為了兩全其美（性能和可預(yù)測(cè)性）你可以在同 一時(shí)間調(diào)用兩個(gè)系統(tǒng)并等著誰先完成。 通常這會(huì)是第一個(gè)系統(tǒng)， 但是在進(jìn)度變得緩慢時(shí)， 第 二個(gè)系統(tǒng)就可以在可接受的時(shí)間內(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)換第一個(gè)系統(tǒng)applyToEither() 算是 acceptEither() 的前輩了。 當(dāng)兩個(gè) futures 快要完成時(shí)，后者只是簡單地調(diào) 用一些代碼片段， applyToEither()將會(huì)返回一個(gè)新的 future 。當(dāng)這兩個(gè)最初的 futures 完成時(shí)， 新的 future 也會(huì)完成。 API 有點(diǎn)類似于 ( Async 變量也是可用的 )：復(fù)制代碼 代碼如下 : CompletableFuture applyToEither(CompletableFuture other, Function fn)這個(gè)額外的 fn 功能在第一個(gè) future 被調(diào)用時(shí)能完成。我不確定這個(gè)專業(yè)化方法的目的是什 么，畢竟一個(gè)人可以簡單地使用： fast.applyToEither(predicta