使用JAVACompletableFuture的20例子

https://segmentfault.com/a/1190000013452165?utm_source&＃61;index-hottest

前言

Java 5 并发库主要关注于异步任务的处理&＃xff0c;它采用了这样一种模式&＃xff0c;producer 线程创建任务并且利用阻塞队列将其传递给任务的 consumer。这种模型在 Java 7 和 8 中进一步发展&＃xff0c;并且开始支持另外一种风格的任务执行&＃xff0c;那就是将任务的数据集分解为子集&＃xff0c;每个子集都可以由独立且同质的子任务来负责处理。

这种风格的基础库也就是 fork/join 框架&＃xff0c;它允许程序员规定数据集该如何进行分割&＃xff0c;并且支持将子任务提交到默认的标准线程池中&＃xff0c;也就是"通用的"ForkJoinPool。Java 8 中&＃xff0c;fork/join 并行功能借助并行流的机制变得更加具有可用性。但是&＃xff0c;不是所有的问题都适合这种风格的并行处理&＃xff1a;所处理的元素必须是独立的&＃xff0c;数据集要足够大&＃xff0c;并且在并行加速方面&＃xff0c;每个元素的处理成本要足够高&＃xff0c;这样才能补偿建立 fork/join 框架所消耗的成本。CompletableFuture 类则是 Java 8 在并行流方面的创新。

准备知识

异步计算

所谓异步调用其实就是实现一个可无需等待被调用函数的返回值而让操作继续运行的方法。在 Java 语言中&＃xff0c;简单的讲就是另启一个线程来完成调用中的部分计算&＃xff0c;使调用继续运行或返回&＃xff0c;而不需要等待计算结果。但调用者仍需要取线程的计算结果。

回调函数

回调函数比较通用的解释是&＃xff0c;它是一个通过函数指针调用的函数。如果你把函数的指针&＃xff08;地址&＃xff09;作为参数传递给另一个函数&＃xff0c;当这个指针被用为调用它所指向的函数时&＃xff0c;我们就说这是回调函数。回调函数不是由该函数的实现方直接调用&＃xff0c;而是在特定的事件或条件发生时由另外一方调用的&＃xff0c;用于对该事件或条件进行响应。

回调函数的机制&＃xff1a;

&＃xff08;1&＃xff09;定义一个回调函数&＃xff1b;

&＃xff08;2&＃xff09;提供函数实现的一方在初始化时候&＃xff0c;将回调函数的函数指针注册给调用者&＃xff1b;

&＃xff08;3&＃xff09;当特定的事件或条件发生的时候&＃xff0c;调用者使用函数指针调用回调函数对事件进行处理。

回调函数通常与原始调用者处于同一层次&＃xff0c;如图 1 所示&＃xff1a;

图 1 回调函数示例图

Future 接口介绍

JDK5 新增了 Future 接口&＃xff0c;用于描述一个异步计算的结果。虽然 Future 以及相关使用方法提供了异步执行任务的能力&＃xff0c;但是对于结果的获取却是很不方便&＃xff0c;只能通过阻塞或者轮询的方式得到任务的结果。阻塞的方式显然和我们的异步编程的初衷相违背&＃xff0c;轮询的方式又会耗费无谓的 CPU 资源&＃xff0c;而且也不能及时地得到计算结果&＃xff0c;为什么不能用观察者设计模式呢&＃xff1f;即当计算结果完成及时通知监听者。

有一些开源框架实现了我们的设想&＃xff0c;例如 Netty 的 ChannelFuture 类扩展了 Future 接口&＃xff0c;通过提供 addListener 方法实现支持回调方式的异步编程。Netty 中所有的 I/O 操作都是异步的,这意味着任何的 I/O 调用都将立即返回&＃xff0c;而不保证这些被请求的 I/O 操作在调用结束的时候已经完成。取而代之地&＃xff0c;你会得到一个返回的 ChannelFuture 实例&＃xff0c;这个实例将给你一些关于 I/O 操作结果或者状态的信息。当一个 I/O 操作开始的时候&＃xff0c;一个新的 Future 对象就会被创建。在开始的时候&＃xff0c;新的 Future 是未完成的状态&＃xff0d;&＃xff0d;它既非成功、失败&＃xff0c;也非被取消&＃xff0c;因为 I/O 操作还没有结束。如果 I/O 操作以成功、失败或者被取消中的任何一种状态结束了&＃xff0c;那么这个 Future 将会被标记为已完成&＃xff0c;并包含更多详细的信息&＃xff08;例如&＃xff1a;失败的原因&＃xff09;。请注意&＃xff0c;即使是失败和被取消的状态&＃xff0c;也是属于已完成的状态。阻塞方式的示例代码如清单 1 所示。

清单 1 阻塞方式示例代码

上面代码使用的是 awaitUninterruptibly 方法&＃xff0c;源代码如清单 2 所示。

清单 2 awaitUninterruptibly 源代码

清单 3 异步非阻塞方式示例代码

可以明显的看出&＃xff0c;在异步模式下&＃xff0c;上面这段代码没有阻塞&＃xff0c;在执行 connect 操作后直接执行到 printTime("异步时间&＃xff1a; ")&＃xff0c;随后 connect 完成&＃xff0c;Future 的监听函数输出 connect 操作完成。

非阻塞则是添加监听类 ChannelFutureListener&＃xff0c;通过覆盖 ChannelFutureListener 的 operationComplete 执行业务逻辑。

清单 4 异步非阻塞方式示例代码

这部分代码的逻辑很简单&＃xff0c;就是注册回调函数&＃xff0c;当操作完成后自动调用回调函数&＃xff0c;就达到了异步的效果。

CompletableFuture 类介绍 

Java 8 中, 新增加了一个包含 50 个方法左右的类--CompletableFuture&＃xff0c;它提供了非常强大的 Future 的扩展功能&＃xff0c;可以帮助我们简化异步编程的复杂性&＃xff0c;并且提供了函数式编程的能力&＃xff0c;可以通过回调的方式处理计算结果&＃xff0c;也提供了转换和组合 CompletableFuture 的方法。

对于阻塞或者轮询方式&＃xff0c;依然可以通过 CompletableFuture 类的 CompletionStage 和 Future 接口方式支持。

CompletableFuture 类声明了 CompletionStage 接口&＃xff0c;CompletionStage 接口实际上提供了同步或异步运行计算的舞台&＃xff0c;所以我们可以通过实现多个 CompletionStage 命令&＃xff0c;并且将这些命令串联在一起的方式实现多个命令之间的触发。

我们可以通过 CompletableFuture.supplyAsync(this::sendMsg); 这么一行代码创建一个简单的异步计算。在这行代码中&＃xff0c;supplyAsync 支持异步地执行我们指定的方法&＃xff0c;这个例子中的异步执行方法是 sendMsg。当然&＃xff0c;我们也可以使用 Executor 执行异步程序&＃xff0c;默认是 ForkJoinPool.commonPool()。

我们也可以在异步计算结束之后指定回调函数&＃xff0c;例如 CompletableFuture.supplyAsync(this::sendMsg) .thenAccept(this::notify);这行代码中的 thenAccept 被用于增加回调函数&＃xff0c;在我们的示例中 notify 就成了异步计算的消费者&＃xff0c;它会处理计算结果。

因为CompletableFuture是CompletionInterface接口的实现&＃xff0c;所以我们首先要了解该接口的契约。它代表某个同步或异步计算的一个阶段。你可以把它理解为是一个为了产生有价值最终结果的计算的流水线上的一个单元。这意味着多个ComletionStage指令可以链接起来从而一个阶段的完成可以触发下一个阶段的执行。

除了实现了CompletionStage接口&＃xff0c;Completion还继承了Future&＃xff0c;这个接口用于实现一个未开始的异步事件。因为能够显式的完成Future&＃xff0c;所以取名为CompletableFuture。

1.新建一个完成的CompletableFuture

这个简单的示例中创建了一个已经完成的预先设置好结果的CompletableFuture。通常作为计算的起点阶段。

1. static void completedFutureExample() {

2. CompletableFuture cf &＃61; CompletableFuture.completedFuture("message");

3. assertTrue(cf.isDone());

4. assertEquals("message", cf.getNow(null));

5. }

getNow方法会返回完成后的结果&＃xff08;这里就是message&＃xff09;&＃xff0c;如果还未完成&＃xff0c;则返回传入的默认值null。

2.运行一个简单的异步stage

下面的例子解释了如何创建一个异步运行Runnable的stage。

1. static void runAsyncExample() {

2. CompletableFuture cf &＃61; CompletableFuture.runAsync(() -> {

3. assertTrue(Thread.currentThread().isDaemon());

4. randomSleep();

5. });

6. assertFalse(cf.isDone());

7. sleepEnough();

8. assertTrue(cf.isDone());

9. }

这个例子想要说明两个事情&＃xff1a;

1. CompletableFuture中以Async为结尾的方法将会异步执行
2. 默认情况下&＃xff08;即指没有传入Executor的情况下&＃xff09;&＃xff0c;异步执行会使用ForkJoinPool实现&＃xff0c;该线程池使用一个后台线程来执行Runnable任务。注意这只是特定于CompletableFuture实现&＃xff0c;其它的CompletableStage实现可以重写该默认行为。

3.将方法作用于前一个Stage

下面的例子引用了第一个例子中已经完成的CompletableFuture&＃xff0c;它将引用生成的字符串结果并将该字符串大写。

1. static void thenApplyExample() {

2. CompletableFuture cf &＃61; CompletableFuture.completedFuture("message").thenApply(s -> {

3. assertFalse(Thread.currentThread().isDaemon());

4. return s.toUpperCase();

5. });

6. assertEquals("MESSAGE", cf.getNow(null));

7. }

这里的关键词是thenApply&＃xff1a;

1. then是指在当前阶段正常执行完成后&＃xff08;正常执行是指没有抛出异常&＃xff09;进行的操作。在本例中&＃xff0c;当前阶段已经完成并得到值message。
2. Apply是指将一个Function作用于之前阶段得出的结果

Function是阻塞的&＃xff0c;这意味着只有当大写操作执行完成之后才会执行getNow()方法。

4.异步的的将方法作用于前一个Stage

通过在方法后面添加Async后缀&＃xff0c;该CompletableFuture链将会异步执行&＃xff08;使用ForkJoinPool.commonPool()&＃xff09;

1. static void thenApplyAsyncExample() {

2. CompletableFuture cf &＃61; CompletableFuture.completedFuture("message").thenApplyAsync(s -> {

3. assertTrue(Thread.currentThread().isDaemon());

4. randomSleep();

5. return s.toUpperCase();

6. });

7. assertNull(cf.getNow(null));

8. assertEquals("MESSAGE", cf.join());

9. }

使用一个自定义的Executor来异步执行该方法

异步方法的一个好处是可以提供一个Executor来执行CompletableStage。这个例子展示了如何使用一个固定大小的线程池来实现大写操作。

1. static ExecutorService executor &＃61; Executors.newFixedThreadPool(3, new ThreadFactory() {

2. int count &＃61; 1;

3. &＃64;Override

4. public Thread newThread(Runnable runnable) {

5. return new Thread(runnable, "custom-executor-" &＃43; count&＃43;&＃43;);

6. }

7. });

8. static void thenApplyAsyncWithExecutorExample() {

9. CompletableFuture cf &＃61; CompletableFuture.completedFuture("message").thenApplyAsync(s -> {

10. assertTrue(Thread.currentThread().getName().startsWith("custom-executor-"));

11. assertFalse(Thread.currentThread().isDaemon());

12. randomSleep();

13. return s.toUpperCase();

14. }, executor);

15. assertNull(cf.getNow(null));

16. assertEquals("MESSAGE", cf.join());

17. }

6.消费(Consume)前一个Stage的结果

如果下一个Stage接收了当前Stage的结果但是在计算中无需返回值&＃xff08;比如其返回值为void&＃xff09;&＃xff0c;那么它将使用方法thenAccept并传入一个Consumer接口。

1. static void thenAcceptExample() {

2. StringBuilder result &＃61; new StringBuilder();

3. CompletableFuture.completedFuture("thenAccept message")

4. .thenAccept(s -> result.append(s));

5. assertTrue("Result was empty", result.length() > 0);

6. }

Consumer将会同步执行&＃xff0c;所以我们无需在返回的CompletableFuture上执行join操作。

7.异步执行Comsume

同样&＃xff0c;使用Asyn后缀实现&＃xff1a;

1. static void thenAcceptAsyncExample() {

2. StringBuilder result &＃61; new StringBuilder();

3. CompletableFuture cf &＃61; CompletableFuture.completedFuture("thenAcceptAsync message")

4. .thenAcceptAsync(s -> result.append(s));

5. cf.join();

6. assertTrue("Result was empty", result.length() > 0);

7. }

8.计算出现异常时

我们现在来模拟一个出现异常的场景。为了简洁性&＃xff0c;我们还是将一个字符串大写&＃xff0c;但是我们会模拟延时进行该操作。我们会使用thenApplyAsyn(Function, Executor)&＃xff0c;第一个参数是大写转化方法&＃xff0c;第二个参数是一个延时executor&＃xff0c;它会延时一秒钟再将操作提交给ForkJoinPool。

1. static void completeExceptionallyExample() {

2. CompletableFuture cf &＃61; CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,

3. CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));

4. CompletableFuture exceptionHandler &＃61; cf.handle((s, th) -> { return (th !&＃61; null) ? "message upon cancel" : ""; });

5. cf.completeExceptionally(new RuntimeException("completed exceptionally"));

6. assertTrue("Was not completed exceptionally", cf.isCompletedExceptionally());

7. try {

8. cf.join();

9. fail("Should have thrown an exception");

10. } catch(CompletionException ex) { // just for testing

11. assertEquals("completed exceptionally", ex.getCause().getMessage());

12. }

13. assertEquals("message upon cancel", exceptionHandler.join());

14. }

1. 首先&＃xff0c;我们新建了一个已经完成并带有返回值message的CompletableFuture对象。然后我们调用thenApplyAsync方法&＃xff0c;该方法会返回一个新的CompletableFuture。这个方法用异步的方式执行大写操作。这里还展示了如何使用delayedExecutor(timeout, timeUnit)方法来延时异步操作。
2. 然后我们创建了一个handler stage&＃xff0c;exceptionHandler&＃xff0c;这个阶段会处理一切异常并返回另一个消息message upon cancel。
3. 最后&＃xff0c;我们显式的完成第二个阶段并抛出异常&＃xff0c;它会导致进行大写操作的阶段抛出CompletionException。它还会触发handler阶段。

API补充&＃xff1a;
CompletableFuture handle(BiFunction fn)
返回一个新的CompletionStage&＃xff0c;无论之前的Stage是否正常运行完毕。传入的参数包括上一个阶段的结果和抛出异常。