點擊上方☝SpringForAll社區輕鬆關注！
本文來源：http://rrd.me/g6P3V
在Java中異步編程，不一定非要使用rxJava, Java本身的庫中的CompletableFuture可以很好的應對大部分的場景。
」原文: 20 Examples of Using Java’s CompletableFuture, 作者 Mahmoud Anouti。
這篇文章介紹 Java 8 的 CompletionStage API和它的標準庫的實現 CompletableFuture。API通過例子的方式演示了它的行為，每個例子演示一到兩個行為。
既然CompletableFuture類實現了CompletionStage接口，首先我們需要理解這個接口的契約。它代表了一個特定的計算的階段，可以同步或者異步的被完成。你可以把它看成一個計算流水線上的一個單元，最終會產生一個最終結果，這意味着幾個CompletionStage可以串聯起來，一個完成的階段可以觸發下一階段的執行，接着觸發下一次，接着……
除了實現CompletionStage接口， CompletableFuture也實現了future接口, 代表一個未完成的異步事件。CompletableFuture提供了方法，能夠顯式地完成這個future,所以它叫CompletableFuture。
1、 創建一個完成的CompletableFuture
最簡單的例子就是使用一個預定義的結果創建一個完成的CompletableFuture,通常我們會在計算的開始階段使用它。
staticvoidcompletedFutureExample(){CompletableFuturecf=CompletableFuture.completedFuture("message");assertTrue(cf.isDone());assertEquals("message",cf.getNow(null));}getNow(null)方法在future完成的情況下會返回結果，就比如上面這個例子，否則返回null (傳入的參數)。
2、運行一個簡單的異步階段
這個例子創建一個一個異步執行的階段：
staticvoidrunAsyncExample(){CompletableFuturecf=CompletableFuture.runAsync(()->{assertTrue(Thread.currentThread().isDaemon());randomSleep();});assertFalse(cf.isDone());sleepEnough();assertTrue(cf.isDone());}通過這個例子可以學到兩件事情：
CompletableFuture的方法如果以Async結尾，它會異步的執行(沒有指定executor的情況下)， 異步執行通過ForkJoinPool實現， 它使用守護線程去執行任務。注意這是CompletableFuture的特性， 其它CompletionStage可以override這個默認的行為。
3、在前一個階段上應用函數
下面這個例子使用前面 #1 的完成的CompletableFuture， #1返回結果為字符串message,然後應用一個函數把它變成大寫字母。
staticvoidthenApplyExample(){CompletableFuturecf=CompletableFuture.completedFuture("message").thenApply(s->{assertFalse(Thread.currentThread().isDaemon());returns.toUpperCase();});assertEquals("MESSAGE",cf.getNow(null));}注意thenApply方法名稱代表的行為。
then意味着這個階段的動作發生當前的階段正常完成之後。本例中，當前節點完成，返回字符串message。
Apply意味着返回的階段將會對結果前一階段的結果應用一個函數。
函數的執行會被阻塞，這意味着getNow()只有打斜操作被完成後才返回。
4、在前一個階段上異步應用函數
通過調用異步方法(方法後邊加Async後綴)，串聯起來的CompletableFuture可以異步地執行（使用ForkJoinPool.commonPool()）。
staticvoidthenApplyAsyncExample(){CompletableFuturecf=CompletableFuture.completedFuture("message").thenApplyAsync(s->{assertTrue(Thread.currentThread().isDaemon());randomSleep();returns.toUpperCase();});assertNull(cf.getNow(null));assertEquals("MESSAGE",cf.join());}5、使用定製的Executor在前一個階段上異步應用函數
異步方法一個非常有用的特性就是能夠提供一個Executor來異步地執行CompletableFuture。這個例子演示了如何使用一個固定大小的線程池來應用大寫函數。
staticExecutorServiceexecutor=Executors.newFixedThreadPool(3,newThreadFactory(){intcount=1;@OverridepublicThreadnewThread(Runnablerunnable){returnnewThread(runnable,"custom-executor-"+count++);}});staticvoidthenApplyAsyncWithExecutorExample(){CompletableFuturecf=CompletableFuture.completedFuture("message").thenApplyAsync(s->{assertTrue(Thread.currentThread().getName().startsWith("custom-executor-"));assertFalse(Thread.currentThread().isDaemon());randomSleep();returns.toUpperCase();},executor);assertNull(cf.getNow(null));assertEquals("MESSAGE",cf.join());}6、消費前一階段的結果
如果下一階段接收了當前階段的結果，但是在計算的時候不需要返回值(它的返回類型是void)， 那麼它可以不應用一個函數，而是一個消費者， 調用方法也變成了thenAccept:
staticvoidthenAcceptExample(){StringBuilderresult=newStringBuilder();CompletableFuture.completedFuture("thenAcceptmessage").thenAccept(s->result.append(s));assertTrue("Resultwasempty",result.length()>0);}本例中消費者同步地執行，所以我們不需要在CompletableFuture調用join方法。
7、異步地消費遷移階段的結果
同樣，可以使用thenAcceptAsync方法， 串聯的CompletableFuture可以異步地執行。
staticvoidthenAcceptAsyncExample(){StringBuilderresult=newStringBuilder();CompletableFuturecf=CompletableFuture.completedFuture("thenAcceptAsyncmessage").thenAcceptAsync(s->result.append(s));cf.join();assertTrue("Resultwasempty",result.length()>0);}8、完成計算異常
現在我們來看一下異步操作如何顯式地返回異常，用來指示計算失敗。我們簡化這個例子，操作處理一個字符串，把它轉換成答謝，我們模擬延遲一秒。
我們使用thenApplyAsync(Function, Executor)方法，第一個參數傳入大寫函數， executor是一個delayed executor,在執行前會延遲一秒。
staticvoidcompleteExceptionallyExample(){CompletableFuturecf=CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,CompletableFuture.delayedExecutor(1,TimeUnit.SECONDS));CompletableFutureexceptionHandler=cf.handle((s,th)->{return(th!=null)?"messageuponcancel":"";});cf.completeExceptionally(newRuntimeException("completedexceptionally"));assertTrue("Wasnotcompletedexceptionally",cf.isCompletedExceptionally());try{cf.join();fail("Shouldhavethrownanexception");}catch(CompletionExceptionex){//justfortestingassertEquals("completedexceptionally",ex.getCause().getMessage());}assertEquals("messageuponcancel",exceptionHandler.join());}讓我們看一下細節。
首先我們創建了一個CompletableFuture, 完成後返回一個字符串message,接着我們調用thenApplyAsync方法，它返回一個CompletableFuture。這個方法在第一個函數完成後，異步地應用轉大寫字母函數。
這個例子還演示了如何通過delayedExecutor(timeout, timeUnit)延遲執行一個異步任務。
我們創建了一個分離的handler階段：exceptionHandler， 它處理異常異常，在異常情況下返回message upon cancel。
下一步我們顯式地用異常完成第二個階段。在階段上調用join方法，它會執行大寫轉換，然後拋出CompletionException（正常的join會等待1秒，然後得到大寫的字符串。不過我們的例子還沒等它執行就完成了異常）， 然後它觸發了handler階段。
9、取消計算
和完成異常類似，我們可以調用cancel(boolean mayInterruptIfRunning)取消計算。對於CompletableFuture類，布爾參數並沒有被使用，這是因為它並沒有使用中斷去取消操作，相反，cancel等價於completeExceptionally(new CancellationException())。
staticvoidcancelExample(){CompletableFuturecf=CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,CompletableFuture.delayedExecutor(1,TimeUnit.SECONDS));CompletableFuturecf2=cf.exceptionally(throwable->"canceledmessage");assertTrue("Wasnotcanceled",cf.cancel(true));assertTrue("Wasnotcompletedexceptionally",cf.isCompletedExceptionally());assertEquals("canceledmessage",cf2.join());}10、在兩個完成的階段其中之一上應用函數
下面的例子創建了CompletableFuture, applyToEither處理兩個階段， 在其中之一上應用函數(包保證哪一個被執行)。本例中的兩個階段一個是應用大寫轉換在原始的字符串上， 另一個階段是應用小些轉換。
staticvoidapplyToEitherExample(){Stringoriginal="Message";CompletableFuturecf1=CompletableFuture.completedFuture(original).thenApplyAsync(s->delayedUpperCase(s));CompletableFuturecf2=cf1.applyToEither(CompletableFuture.completedFuture(original).thenApplyAsync(s->delayedLowerCase(s)),s->s+"fromapplyToEither");assertTrue(cf2.join().endsWith("fromapplyToEither"));}11、在兩個完成的階段其中之一上調用消費函數
和前一個例子很類似了，只不過我們調用的是消費者函數 (Function變成Consumer):
staticvoidacceptEitherExample(){Stringoriginal="Message";StringBuilderresult=newStringBuilder();CompletableFuturecf=CompletableFuture.completedFuture(original).thenApplyAsync(s->delayedUpperCase(s)).acceptEither(CompletableFuture.completedFuture(original).thenApplyAsync(s->delayedLowerCase(s)),s->result.append(s).append("acceptEither"));cf.join();assertTrue("Resultwasempty",result.toString().endsWith("acceptEither"));}12、在兩個階段都執行完後運行一個 Runnable
這個例子演示了依賴的CompletableFuture如果等待兩個階段完成後執行了一個Runnable。注意下面所有的階段都是同步執行的，第一個階段執行大寫轉換，第二個階段執行小寫轉換。
staticvoidrunAfterBothExample(){Stringoriginal="Message";StringBuilderresult=newStringBuilder();CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).runAfterBoth(CompletableFuture.completedFuture(original).thenApply(String::toLowerCase),()->result.append("done"));assertTrue("Resultwasempty",result.length()>0);}13、 使用BiConsumer處理兩個階段的結果
上面的例子還可以通過BiConsumer來實現:
staticvoidthenAcceptBothExample(){Stringoriginal="Message";StringBuilderresult=newStringBuilder();CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).thenAcceptBoth(CompletableFuture.completedFuture(original).thenApply(String::toLowerCase),(s1,s2)->result.append(s1+s2));assertEquals("MESSAGEmessage",result.toString());}14、使用BiFunction處理兩個階段的結果
如果CompletableFuture依賴兩個前面階段的結果， 它複合兩個階段的結果再返回一個結果，我們就可以使用thenCombine()函數。整個流水線是同步的，所以getNow()會得到最終的結果，它把大寫和小寫字符串連接起來。
staticvoidthenCombineExample(){Stringoriginal="Message";CompletableFuturecf=CompletableFuture.completedFuture(original).thenApply(s->delayedUpperCase(s)).thenCombine(CompletableFuture.completedFuture(original).thenApply(s->delayedLowerCase(s)),(s1,s2)->s1+s2);assertEquals("MESSAGEmessage",cf.getNow(null));}15、異步使用BiFunction處理兩個階段的結果
類似上面的例子，但是有一點不同：依賴的前兩個階段異步地執行，所以thenCombine()也異步地執行，即時它沒有Async後綴。
Javadoc中有注釋：
「Actions supplied for dependent completions of non-async methods may be performed by the thread that completes the current CompletableFuture, or by any other caller of a completion method
」所以我們需要join方法等待結果的完成。
staticvoidthenCombineAsyncExample(){Stringoriginal="Message";CompletableFuturecf=CompletableFuture.completedFuture(original).thenApplyAsync(s->delayedUpperCase(s)).thenCombine(CompletableFuture.completedFuture(original).thenApplyAsync(s->delayedLowerCase(s)),(s1,s2)->s1+s2);assertEquals("MESSAGEmessage",cf.join());}16、組合 CompletableFuture
我們可以使用thenCompose()完成上面兩個例子。這個方法等待第一個階段的完成(大寫轉換)， 它的結果傳給一個指定的返回CompletableFuture函數，它的結果就是返回的CompletableFuture的結果。
有點拗口，但是我們看例子來理解。函數需要一個大寫字符串做參數，然後返回一個CompletableFuture, 這個CompletableFuture會轉換字符串變成小寫然後連接在大寫字符串的後面。
staticvoidthenComposeExample(){Stringoriginal="Message";CompletableFuturecf=CompletableFuture.completedFuture(original).thenApply(s->delayedUpperCase(s)).thenCompose(upper->CompletableFuture.completedFuture(original).thenApply(s->delayedLowerCase(s)).thenApply(s->upper+s));assertEquals("MESSAGEmessage",cf.join());}17、當幾個階段中的一個完成，創建一個完成的階段
下面的例子演示了當任意一個CompletableFuture完成後， 創建一個完成的CompletableFuture.
待處理的階段首先創建， 每個階段都是轉換一個字符串為大寫。因為本例中這些階段都是同步地執行(thenApply), 從anyOf中創建的CompletableFuture會立即完成，這樣所有的階段都已完成，我們使用whenComplete(BiConsumer<? super Object, ? super Throwable> action)處理完成的結果。
staticvoidanyOfExample(){StringBuilderresult=newStringBuilder();Listmessages=Arrays.asList("a","b","c");List<CompletableFuture>futures=messages.stream().map(msg->CompletableFuture.completedFuture(msg).thenApply(s->delayedUpperCase(s))).collect(Collectors.toList());CompletableFuture.anyOf(futures.toArray(newCompletableFuture[futures.size()])).whenComplete((res,th)->{if(th==null){assertTrue(isUpperCase((String)res));result.append(res);}});assertTrue("Resultwasempty",result.length()>0);}18、當所有的階段都完成後創建一個階段
上一個例子是當任意一個階段完成後接着處理，接下來的兩個例子演示當所有的階段完成後才繼續處理, 同步地方式和異步地方式兩種。
staticvoidallOfExample(){StringBuilderresult=newStringBuilder();Listmessages=Arrays.asList("a","b","c");List<CompletableFuture>futures=messages.stream().map(msg->CompletableFuture.completedFuture(msg).thenApply(s->delayedUpperCase(s))).collect(Collectors.toList());CompletableFuture.allOf(futures.toArray(newCompletableFuture[futures.size()])).whenComplete((v,th)->{futures.forEach(cf->assertTrue(isUpperCase(cf.getNow(null))));result.append("done");});assertTrue("Resultwasempty",result.length()>0);}19、當所有的階段都完成後異步地創建一個階段 使用thenApplyAsync()替換那些單個的CompletableFutures的方法，allOf()會在通用池中的線程中異步地執行。所以我們需要調用join方法等待它完成。
staticvoidallOfAsyncExample(){StringBuilderresult=newStringBuilder();Listmessages=Arrays.asList("a","b","c");List<CompletableFuture>futures=messages.stream().map(msg->CompletableFuture.completedFuture(msg).thenApplyAsync(s->delayedUpperCase(s))).collect(Collectors.toList());CompletableFutureallOf=CompletableFuture.allOf(futures.toArray(newCompletableFuture[futures.size()])).whenComplete((v,th)->{futures.forEach(cf->assertTrue(isUpperCase(cf.getNow(null))));result.append("done");});allOf.join();assertTrue("Resultwasempty",result.length()>0);}20、真實的例子
Now that the functionality of CompletionStage and specifically CompletableFuture is explored, the below example applies them in a practical scenario:
現在你已經了解了CompletionStage 和 CompletableFuture 的一些函數的功能，下面的例子是一個實踐場景：
因為每個汽車的實例都是獨立的，得到每個汽車的評分都可以異步地執行，這會提高系統的性能(延遲)，而且，等待所有的汽車評分被處理使用的是allOf方法，而不是手工的線程等待(Thread#join() 或 a CountDownLatch)。
這些例子可以幫助你更好的理解相關的API,你可以在github上得到所有的例子的代碼。
其它參考文檔
●如何使用Redis實現頁面UV統計-HyperLogLog實現詳解
●Redis 6.0 客戶端緩存特性及實踐
●推薦一個適用於SpringBoot項目的輕量級HTTP客戶端框架，快來試試它!
關注公眾號，回復「spring」有驚喜！！！
如果資源對你有幫助的話
留言列表