一网打尽异步神器CompletableFuture

三友的java日记

发布于 2022-07-27 14:08:49

6380

发布于 2022-07-27 14:08:49

文章被收录于专栏：原创干货

大家好，我是三友。

最近一直畅游在RocketMQ的源码中，发现在RocketMQ中很多地方都使用到了CompletableFuture，所以今天就跟大家来聊一聊JDK1.8提供的异步神器CompletableFuture，并且最后会结合RocketMQ源码分析一下CompletableFuture的使用。

Future接口以及它的局限性

我们都知道，Java中创建线程的方式主要有两种方式，继承Thread或者实现Runnable接口。但是这两种都是有一个共同的缺点，那就是都无法获取到线程执行的结果，也就是没有返回值。于是在JDK1.5 以后为了解决这种没有返回值的问题，提供了Callable和Future接口以及Future对应的实现类FutureTask，通过FutureTask的就可以获取到异步执行的结果。

于是乎，我们想要开启异步线程，执行任务，获取结果，就可以这么实现。

 FutureTask<String> futureTask = new FutureTask<>(() -> "三友");
 new Thread(futureTask).start();
 System.out.println(futureTask.get());

或者使用线程池的方式

ExecutorService executorService = Executors.newSingleThreadExecutor();
Future<String> future = executorService.submit(() -> "三友");
System.out.println(future.get());
executorService.shutdown();

线程池底层也是将提交的Callable的实现先封装成FutureTask，然后通过execute方法来提交任务，执行异步逻辑。

Future接口的局限性

虽然通过Future接口的get方法可以获取任务异步执行的结果，但是get方法会阻塞主线程，也就是异步任务没有完成，主线程会一直阻塞，直到任务结束。

Future也提供了isDone方法来查看异步线程任务执行是否完成，如果完成，就可以获取任务的执行结果，代码如下。

ExecutorService executorService = Executors.newSingleThreadExecutor();
Future<String> future = executorService.submit(() -> "三友");
while (!future.isDone()) {
  //任务有没有完成，没有就继续循环判断
}
System.out.println(future.get());
executorService.shutdown();

但是这种轮询查看异步线程任务执行状态，也是非常消耗cpu资源。

同时对于一些复杂的异步操作任务的处理，可能需要各种同步组件来一起完成。

所以，通过上面的介绍可以看出，Future在使用的过程中还是有很强的局限性，所以为了解决这种局限性，在JDK1.8的时候，Doug Lea 大神为我们提供了一种更为强大的类CompletableFuture。

什么是CompletableFuture？

CompletableFuture在JDK1.8提供了一种更加强大的异步编程的api。它实现了Future接口，也就是Future的功能特性CompletableFuture也有；除此之外，它也实现了CompletionStage接口，CompletionStage接口定义了任务编排的方法，执行某一阶段，可以向下执行后续阶段。

CompletableFuture相比于Future最大的改进就是提供了类似观察者模式的回调监听的功能，也就是当上一阶段任务执行结束之后，可以回调你指定的下一阶段任务，而不需要阻塞获取结果之后来处理结果。

CompletableFuture常见api详解

CompletableFuture的方法api多，但主要可以分为以下几类。

1、实例化CompletableFuture

构造方法创建

CompletableFuture<String> completableFuture = new CompletableFuture<>();
System.out.println(completableFuture.get());

此时如果有其它线程执行如下代码，就能执行打印出三友

completableFuture.complete("三友")

静态方法创建

除了使用构造方法构造，CompletableFuture还提供了静态方法来创建

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier);
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor);

public static CompletableFuture<Void> runAsync(Runnable runnable);
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor);

supply 和 run 的主要区别就是 supply 可以有返回值，run 没有返回值。至于另一个参数Executor 就是用来执行异步任务的线程池，如果不传Executor 的话，默认是ForkJoinPool这个线程池的实现。

一旦通过静态方法来构造，会立马开启异步线程执行Supplier或者Runnable提交的任务。

CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "三友");
System.out.println(completableFuture.get());

一旦任务执行完成，就可以打印返回值，这里的使用方法跟Future是一样的。

所以对比两个两种实例化的方法，使用静态方法的和使用构造方法主要区别就是，使用构造方法需要其它线程主动调用complete来表示任务执行完成，因为很简单，因为在构造的时候没有执行异步的任务，所以需要其它线程主动调用complete来表示任务执行完成。

2、获取任务执行结果

public T get();
public T get(long timeout, TimeUnit unit);
public T getNow(T valueIfAbsent);
public T join();

get()和get(long timeout, TimeUnit unit)是实现了Future接口的功能，两者主要区别就是get()会一直阻塞直到获取到结果，get(long timeout, TimeUnit unit)值可以指定超时时间，当到了指定的时间还未获取到任务，就会抛出TimeoutException异常。

getNow(T valueIfAbsent)：就是获取任务的执行结果，但不会产生阻塞。如果任务还没执行完成，那么就会返回你传入的 valueIfAbsent 参数值，如果执行完成了，就会返回任务执行的结果。

join()：跟get()的主要区别就是，get()会抛出检查时异常，join()不会。

3、主动触发任务完成

public boolean complete(T value);
public boolean completeExceptionally(Throwable ex);

complete：主动触发当前异步任务的完成。调用此方法时如果你的任务已经完成，那么方法就会返回false；如果任务没完成，就会返回true，并且其它线程获取到的任务的结果就是complete的参数值。

completeExceptionally：跟complete的作用差不多，complete是正常结束任务，返回结果，而completeExceptionally就是触发任务执行的异常。

4、对任务执行结果进行下一步处理

只能接收任务正常执行后的回调

public <U> CompletionStage<U> thenApply(Function<? super T,? extends U> fn);
public CompletableFuture<Void> thenRun(Runnable action);
public CompletionStage<Void> thenAccept(Consumer<? super T> action);

这类回调的特点就是，当任务正常执行完成，没有异常的时候就会回调。

thenApply：可以拿到上一步任务执行的结果进行处理，并且返回处理的结果 thenRun：拿不到上一步任务执行的结果，但会执行Runnable接口的实现 thenAccept：可以拿到上一步任务执行的结果进行处理，但不需要返回处理的结果

thenApply示例：

CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> 10)
                .thenApply(v -> ("上一步的执行的结果为：" + v));
System.out.println(completableFuture.join());

执行结果：

上一步的执行的结果为：10

thenRun示例：

CompletableFuture<Void> completableFuture = CompletableFuture.supplyAsync(() -> 10)
      .thenRun(() -> System.out.println("上一步执行完成"));

执行结果：

上一步执行完成

thenAccept示例：

CompletableFuture<Void> completableFuture = CompletableFuture.supplyAsync(() -> 10)
      .thenAccept(v -> System.out.println("上一步执行完成，结果为：" + v));

执行结果：

上一步执行完成，结果为：10

thenApply有异常示例：

CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
    //模拟异常
    int i = 1 / 0;
    return 10;
}).thenApply(v -> ("上一步的执行的结果为：" + v));
System.out.println(completableFuture.join());

执行结果：

Exception in thread "main" java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
 at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
 at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
 at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1606)

当有异常时是不会回调的

只能接收任务处理异常后的回调

public CompletionStage<T> exceptionally(Function<Throwable, ? extends T> fn);

当上面的任务执行过程中出现异常的时候，会回调exceptionally方法指定的回调，但是如果没有出现异常，是不会回调的。

exceptionally能够将异常给吞了，并且fn的返回值会返回回去。

其实这个exceptionally方法有点像降级的味道。当出现异常的时候，走到这个回调，可以返回一个默认值回去。

没有异常情况下：

CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
    return 100;
}).exceptionally(e -> {
    System.out.println("出现异常了，返回默认值");
    return 110;
});
System.out.println(completableFuture.join());

执行结果：

有异常情况下：

CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
    int i = 1 / 0;
    return 100;
}).exceptionally(e -> {
    System.out.println("出现异常了，返回默认值");
    return 110;
});
System.out.println(completableFuture.join());

执行结果：

出现异常了，返回默认值
110

能同时接收任务执行正常和异常的回调

public <U> CompletionStage<U> handle(BiFunction<? super T, Throwable, ? extends U> fn);
public CompletionStage<T> whenComplete(BiConsumer<? super T, ? super Throwable> actin);

不论前面的任务执行成功还是失败都会回调的这类方法指定的回调方法。

handle : 跟exceptionally有点像，但是exceptionally是出现异常才会回调，两者都有返回值，都能吞了异常，但是handle正常情况下也能回调。

whenComplete：能接受正常或者异常的回调，并且不影响上个阶段的返回值，也就是主线程能获取到上个阶段的返回值；当出现异常时，whenComplete并不能吞了这个异常，也就是说主线程在获取执行异常任务的结果时，会抛出异常。

这里演示一下whenComplete处理异常示例情况，handle跟exceptionally对异常的处理差不多。

whenComplete处理异常示例：

CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
      int i = 1 / 0;
      return 10;
}).whenComplete((r, e) -> {
      System.out.println("whenComplete被调用了");
});
System.out.println(completableFuture.join());

执行结果：

whenComplete被调用了
Exception in thread "main" java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
 at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
 at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
 at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1606)

5、对任务结果进行合并

public <U,V> CompletionStage<V> thenCombine
        (CompletionStage<? extends U> other,
         BiFunction<? super T,? super U,? extends V> fn);

这个方法的意思是，当前任务和other任务都执行结束后，拿到这两个任务的执行结果，回调 BiFunction ，然后返回新的结果。

thenCombine的例子请往下继续看。

6、以Async结尾的方法

上面说的一些方法，比如说thenAccept方法，他有两个对应的Async结尾的方法，如下：

public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor);
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);

thenAcceptAsync跟thenAccept的主要区别就是thenAcceptAsync会重新开一个线程来执行下一阶段的任务，而thenAccept还是用上一阶段任务执行的线程执行。

两个thenAcceptAsync主要区别就是一个使用默认的线程池来执行任务，也就是ForkJoinPool，一个是使用方法参数传入的线程池来执行任务。

当然除了thenAccept方法之外，上述提到的方法还有很多带有Async结尾的对应的方法，他们的主要区别就是执行任务是否开启异步线程来执行的区别。

当然，还有一些其它的api，可以自行查看

CompletableFuture在RocketMQ中的使用

CompletableFuture在RocketMQ中的使用场景比较多，这里我举一个消息存储的场景。

在RocketMQ中，Broker接收到生产者产生的消息的时候，会将消息持久化到磁盘和同步到从节点中。持久化到磁盘和消息同步到从节点是两个独立的任务，互不干扰，可以相互独立执行。当消息持久化到磁盘和同步到从节点中任务完成之后，需要统计整个存储消息消耗的时间，所以统计整个存储消息消耗的时间是依赖前面两个任务的完成。

实现代码如下

消息存储刷盘任务和主从复制任务：

PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);
// 提交刷盘的请求
CompletableFuture<PutMessageStatus> flushResultFuture = submitFlushRequest(result, msg);
//提交主从复制的请求
CompletableFuture<PutMessageStatus> replicaResultFuture = submitReplicaRequest(result, msg);

//刷盘 和 主从复制 两个异步任务通过thenCombine联合
return flushResultFuture.thenCombine(replicaResultFuture, (flushStatus, replicaStatus) -> {
    // 当两个刷盘和主从复制任务都完成的时候，就会回调
    // 如果刷盘没有成功，那么就将消息存储的状态设置为失败
    if (flushStatus != PutMessageStatus.PUT_OK) {
        putMessageResult.setPutMessageStatus(flushStatus);
    }
    // 如果主从复制没有成功，那么就将消息存储的状态设置为失败
    if (replicaStatus != PutMessageStatus.PUT_OK) {
        putMessageResult.setPutMessageStatus(replicaStatus);
    }
    // 最终返回消息存储的结果
    return putMessageResult;
});

对上面两个合并的任务执行结果通过thenAccept方法进行监听，统计消息存储的耗时：

//消息存储的开始时间
long beginTime = this.getSystemClock().now();
// 存储消息，然后返回 CompletableFuture,也就是上面一段代码得返回值‍
CompletableFuture<PutMessageResult> putResultFuture = this.commitLog.asyncPutMessage(msg);

//监听消息存储的结果
putResultFuture.thenAccept((result) -> {
    // 消息存储完成之后会回调
    long elapsedTime = this.getSystemClock().now() - beginTime;
    if (elapsedTime > 500) {
        log.warn("putMessage not in lock elapsed time(ms)={}, bodyLength={}", elapsedTime, msg.getBody().length);
    }
    this.storeStatsService.setPutMessageEntireTimeMax(elapsedTime);

    if (null == result || !result.isOk()) {
        this.storeStatsService.getPutMessageFailedTimes().add(1);
    }
});