前面兩篇文章已經整理了CompletableFuture大部分的特性，本文會整理完CompletableFuture余下的特性，以及將它跟RxJava進行比較。
3.6 Either
Either 表示的是兩個CompletableFuture，當其中任意一個CompletableFuture計算完成的時候就會執行。
| 方法名 | 描述 |
|---|---|
| acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action) | 當任意一個CompletableFuture完成的時候，action這個消費者就會被執行。 |
| acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action) | 當任意一個CompletableFuture完成的時候，action這個消費者就會被執行。使用ForkJoinPool |
| acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor) | 當任意一個CompletableFuture完成的時候，action這個消費者就會被執行。使用指定的線程池 |
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();}復制代碼執行結果：The future is from future1 或者 The future is from future2。
因為future1和future2，執行的順序是隨機的。
applyToEither 跟 acceptEither 類似。
| 方法名 | 描述 |
|---|---|
| applyToEither(CompletionStage<? extends T> other, Function<? super T,U> fn) | 當任意一個CompletableFuture完成的時候，fn會被執行，它的返回值會當作新的CompletableFuture<U>的計算結果。 |
| applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn) | 當任意一個CompletableFuture完成的時候，fn會被執行，它的返回值會當作新的CompletableFuture<U>的計算結果。使用ForkJoinPool |
| applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn, Executor executor) | 當任意一個CompletableFuture完成的時候，fn會被執行，它的返回值會當作新的CompletableFuture<U>的計算結果。使用指定的線程池 |
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();}復制代碼執行結果也跟上面的程序類似。
3.7 其他方法
allOf、anyOf是CompletableFuture的靜態方法。
3.7.1 allOf
| 方法名 | 描述 |
|---|---|
| allOf(CompletableFuture<?>... cfs) | 在所有Future對象完成后結束，并返回一個future。 |
allOf()方法所返回的CompletableFuture，并不能組合前面多個CompletableFuture的計算結果。于是我們借助Java 8的Stream來組合多個future的結果。
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);復制代碼執行結果：
tony cafei aaron復制代碼3.7.2 anyOf
| 方法名 | 描述 |
|---|---|
| anyOf(CompletableFuture<?>... cfs) | 在任何一個Future對象結束后結束，并返回一個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();}復制代碼使用anyOf()時，只要某一個future完成，就結束了。所以執行結果可能是"from future1"、"from future2"、"from future3"中的任意一個。
anyOf 和 acceptEither、applyToEither的區別在于，后兩者只能使用在兩個future中，而anyOf可以使用在多個future中。
3.8 CompletableFuture異常處理
CompletableFuture在運行時如果遇到異常，可以使用get()并拋出異常進行處理，但這并不是一個最好的方法。CompletableFuture本身也提供了幾種方式來處理異常。
3.8.1 exceptionally
| 方法名 | 描述 |
|---|---|
| exceptionally(Function fn) | 只有當CompletableFuture拋出異常的時候，才會觸發這個exceptionally的計算，調用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;});復制代碼執行結果：
Unexpected error:java.util.concurrent.CompletionException: java.lang.NullPointerException復制代碼對上面的代碼稍微做了一下修改，修復了空指針的異常。
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;});復制代碼執行結果：
11復制代碼3.8.2 whenComplete
whenComplete 在上一篇文章其實已經介紹過了，在這里跟exceptionally的作用差不多，可以捕獲任意階段的異常。如果沒有異常的話，就執行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);}});復制代碼執行結果：
Unexpected error:java.util.concurrent.CompletionException: java.lang.NullPointerException復制代碼跟whenComplete相似的方法是handle，handle的用法在上一篇文章中也已經介紹過。
四. CompletableFuture VS Java8 Stream VS RxJava1 & RxJava2
CompletableFuture 有很多特性跟RxJava很像，所以將CompletableFuture、Java 8 Stream和RxJava做一個相互的比較。
| composable | lazy | resuable | async | cached | push | back pressure | |
|---|---|---|---|---|---|---|---|
| CompletableFuture | 支持 | 不支持 | 支持 | 支持 | 支持 | 支持 | 不支持 |
| Stream | 支持 | 支持 | 不支持 | 不支持 | 不支持 | 不支持 | 不支持 |
| Observable(RxJava1) | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 |
| Observable(RxJava2) | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 | 不支持 |
| Flowable(RxJava2) | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 |
五. 總結
Java 8提供了一種函數風格的異步和事件驅動編程模型CompletableFuture，它不會造成堵塞。CompletableFuture背后依靠的是fork/join框架來啟動新的線程實現異步與并發。當然，我們也能通過指定線程池來做這些事情。
CompletableFuture特別是對微服務架構而言，會有很大的作為。舉一個具體的場景，電商的商品頁面可能會涉及到商品詳情服務、商品評論服務、相關商品推薦服務等等。獲取商品的信息時（/productdetails?productid=xxx），需要調用多個服務來處理這一個請求并返回結果。這里可能會涉及到并發編程，我們完全可以使用Java 8的CompletableFuture或者RxJava來實現。
先前的文章：
Java8新的異步編程方式 CompletableFuture(一)
Java8新的異步編程方式 CompletableFuture(二)
