前面兩篇文章已經(jīng)整理了CompletableFuture大部分的特性，本文會整理完CompletableFuture余下的特性，以及將它跟RxJava進行比較。

3.6 Either

Either 表示的是兩個CompletableFuture，當(dāng)其中任意一個CompletableFuture計算完成的時候就會執(zhí)行。

方法名描述

acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action)	當(dāng)任意一個CompletableFuture完成的時候，action這個消費者就會被執(zhí)行。
acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action)	當(dāng)任意一個CompletableFuture完成的時候，action這個消費者就會被執(zhí)行。使用ForkJoinPool
acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor)	當(dāng)任意一個CompletableFuture完成的時候，action這個消費者就會被執(zhí)行。使用指定的線程池

Random random = new Random();CompletableFuture<String> future1 = CompletableFuture.supplyAsync(()->{try {Thread.sleep(random.nextInt(1000));} catch (InterruptedException e) {e.printStackTrace();}return "from future1";});CompletableFuture<String> future2 = CompletableFuture.supplyAsync(()->{try {Thread.sleep(random.nextInt(1000));} catch (InterruptedException e) {e.printStackTrace();}return "from future2";});CompletableFuture<Void> future = future1.acceptEither(future2,str->System.out.println("The future is "+str));try {future.get();} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();}復(fù)制代碼

執(zhí)行結(jié)果：The future is from future1 或者 The future is from future2。
因為future1和future2，執(zhí)行的順序是隨機的。

applyToEither 跟 acceptEither 類似。

方法名描述

applyToEither(CompletionStage<? extends T> other, Function<? super T,U> fn)	當(dāng)任意一個CompletableFuture完成的時候，fn會被執(zhí)行，它的返回值會當(dāng)作新的CompletableFuture<U>的計算結(jié)果。
applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn)	當(dāng)任意一個CompletableFuture完成的時候，fn會被執(zhí)行，它的返回值會當(dāng)作新的CompletableFuture<U>的計算結(jié)果。使用ForkJoinPool
applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn, Executor executor)	當(dāng)任意一個CompletableFuture完成的時候，fn會被執(zhí)行，它的返回值會當(dāng)作新的CompletableFuture<U>的計算結(jié)果。使用指定的線程池

Random random = new Random();CompletableFuture<String> future1 = CompletableFuture.supplyAsync(()->{try {Thread.sleep(random.nextInt(1000));} catch (InterruptedException e) {e.printStackTrace();}return "from future1";});CompletableFuture<String> future2 = CompletableFuture.supplyAsync(()->{try {Thread.sleep(random.nextInt(1000));} catch (InterruptedException e) {e.printStackTrace();}return "from future2";});CompletableFuture<String> future = future1.applyToEither(future2,str->"The future is "+str);try {System.out.println(future.get());} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();}復(fù)制代碼

執(zhí)行結(jié)果也跟上面的程序類似。

3.7 其他方法

allOf、anyOf是CompletableFuture的靜態(tài)方法。

3.7.1 allOf

方法名描述

allOf(CompletableFuture<?>... cfs)	在所有Future對象完成后結(jié)束，并返回一個future。

allOf()方法所返回的CompletableFuture，并不能組合前面多個CompletableFuture的計算結(jié)果。于是我們借助Java 8的Stream來組合多個future的結(jié)果。

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "tony");CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "cafei");CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> "aaron");CompletableFuture.allOf(future1, future2, future3).thenApply(v ->Stream.of(future1, future2, future3).map(CompletableFuture::join).collect(Collectors.joining(" "))).thenAccept(System.out::print);復(fù)制代碼

執(zhí)行結(jié)果：

tony cafei aaron復(fù)制代碼

3.7.2 anyOf

方法名描述

anyOf(CompletableFuture<?>... cfs)	在任何一個Future對象結(jié)束后結(jié)束，并返回一個future。

Random rand = new Random();CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(rand.nextInt(1000));} catch (InterruptedException e) {e.printStackTrace();}return "from future1";});CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(rand.nextInt(1000));} catch (InterruptedException e) {e.printStackTrace();}return "from future2";});CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(rand.nextInt(1000));} catch (InterruptedException e) {e.printStackTrace();}return "from future3";});CompletableFuture<Object> future = CompletableFuture.anyOf(future1,future2,future3);try {System.out.println(future.get());} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();}復(fù)制代碼

使用anyOf()時，只要某一個future完成，就結(jié)束了。所以執(zhí)行結(jié)果可能是"from future1"、"from future2"、"from future3"中的任意一個。

anyOf 和 acceptEither、applyToEither的區(qū)別在于，后兩者只能使用在兩個future中，而anyOf可以使用在多個future中。

3.8 CompletableFuture異常處理

CompletableFuture在運行時如果遇到異常，可以使用get()并拋出異常進行處理，但這并不是一個最好的方法。CompletableFuture本身也提供了幾種方式來處理異常。

3.8.1 exceptionally

方法名描述

exceptionally(Function fn)	只有當(dāng)CompletableFuture拋出異常的時候，才會觸發(fā)這個exceptionally的計算，調(diào)用function計算值。

CompletableFuture.supplyAsync(() -> "hello world").thenApply(s -> {s = null;int length = s.length();return length;}).thenAccept(i -> System.out.println(i)).exceptionally(t -> {System.out.println("Unexpected error:" + t);return null;});復(fù)制代碼

執(zhí)行結(jié)果：

Unexpected error:java.util.concurrent.CompletionException: java.lang.NullPointerException復(fù)制代碼

對上面的代碼稍微做了一下修改，修復(fù)了空指針的異常。

CompletableFuture.supplyAsync(() -> "hello world").thenApply(s -> { // s = null;int length = s.length();return length;}).thenAccept(i -> System.out.println(i)).exceptionally(t -> {System.out.println("Unexpected error:" + t);return null;});復(fù)制代碼

執(zhí)行結(jié)果：

11復(fù)制代碼

3.8.2 whenComplete

whenComplete 在上一篇文章其實已經(jīng)介紹過了，在這里跟exceptionally的作用差不多，可以捕獲任意階段的異常。如果沒有異常的話，就執(zhí)行action。

CompletableFuture.supplyAsync(() -> "hello world").thenApply(s -> {s = null;int length = s.length();return length;}).thenAccept(i -> System.out.println(i)).whenComplete((result, throwable) -> {if (throwable != null) {System.out.println("Unexpected error:"+throwable);} else {System.out.println(result);}});復(fù)制代碼

執(zhí)行結(jié)果：

Unexpected error:java.util.concurrent.CompletionException: java.lang.NullPointerException復(fù)制代碼

跟whenComplete相似的方法是handle，handle的用法在上一篇文章中也已經(jīng)介紹過。

四. CompletableFuture VS Java8 Stream VS RxJava1 & RxJava2

CompletableFuture 有很多特性跟RxJava很像，所以將CompletableFuture、Java 8 Stream和RxJava做一個相互的比較。

composablelazyresuableasynccachedpushback pressure

CompletableFuture	支持	不支持	支持	支持	支持	支持	不支持
Stream	支持	支持	不支持	不支持	不支持	不支持	不支持
Observable(RxJava1)	支持	支持	支持	支持	支持	支持	支持
Observable(RxJava2)	支持	支持	支持	支持	支持	支持	不支持
Flowable(RxJava2)	支持	支持	支持	支持	支持	支持	支持

五. 總結(jié)

Java 8提供了一種函數(shù)風(fēng)格的異步和事件驅(qū)動編程模型CompletableFuture，它不會造成堵塞。CompletableFuture背后依靠的是fork/join框架來啟動新的線程實現(xiàn)異步與并發(fā)。當(dāng)然，我們也能通過指定線程池來做這些事情。

CompletableFuture特別是對微服務(wù)架構(gòu)而言，會有很大的作為。舉一個具體的場景，電商的商品頁面可能會涉及到商品詳情服務(wù)、商品評論服務(wù)、相關(guān)商品推薦服務(wù)等等。獲取商品的信息時（/productdetails?productid=xxx），需要調(diào)用多個服務(wù)來處理這一個請求并返回結(jié)果。這里可能會涉及到并發(fā)編程，我們完全可以使用Java 8的CompletableFuture或者RxJava來實現(xiàn)。

先前的文章：
Java8新的異步編程方式 CompletableFuture(一)
Java8新的異步編程方式 CompletableFuture(二)

總結(jié)

以上是生活随笔為你收集整理的Java8新的异步编程方式 CompletableFuture(三)的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。