前面两篇文章已经整理了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<Throwable,? extends T> 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来实现。
