CompletableFuture在RocketMQ中的使用实战！

原创

程序员蜗牛

发布于 2024-02-18 09:12:24

15300

代码可运行

运行总次数：0

代码可运行

今天想跟大家来聊一聊JDK1.8提供的异步神器CompletableFuture，

最后呢我会结合RocketMQ源码分析一下CompletableFuture的使用。

Future接口以及它的局限性

 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接口的局限性

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

什么是CompletableFuture？

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);

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

2、获取任务执行结果

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

3、主动触发任务完成

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

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);

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);

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);

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);