前面兩篇文章已經整理了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做一個相互的比較。

composablelazyresuableasynccachedpushback 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(二)

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。