参考:
CompletableFuture原理与实践-外卖商家端API的异步化 - 美团技术团队
CompletableFuture 详解 | JavaGuide
1.CompletableFuture介绍
CompletableFuture是由Java 8引入的,在Java8之前我们一般通过Future实现异步。
- Future用于表示异步计算的结果,只能通过阻塞或者轮询的方式获取结果,而且不支持设置回调方法,Java 8之前若要设置回调一般会使用guava的ListenableFuture,回调的引入又会导致臭名昭著的回调地狱。
- CompletableFuture对Future进行了扩展,可以通过设置回调的方式处理计算结果,同时也支持组合操作,支持进一步的编排,同时一定程度解决了回调地狱的问题。
CompletableFuture实现了两个接口(如上图所示):Future、CompletionStage。Future表示异步计算的结果,CompletionStage用于表示异步执行过程中的一个步骤(Stage),这个步骤可能是由另外一个CompletionStage触发的,随着当前步骤的完成,也可能会触发其他一系列CompletionStage的执行。从而我们可以根据实际业务对这些步骤进行多样化的编排组合,CompletionStage接口正是定义了这样的能力,我们可以通过其提供的thenAppy、thenCompose等函数式编程方法来组合编排这些步骤。
2. CompletableFuture操作
2.1 使用
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier);
// 使用自定义线程池(推荐)
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor);static CompletableFuture<Void> runAsync(Runnable runnable);
// 使用自定义线程池(推荐)
static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor);
runAsync()
方法接受的参数是 Runnable
,这是一个函数式接口,不允许返回值。
supplyAsync()
方法接受的参数是 Supplier<U>
,这也是一个函数式接口,U
是返回结果值的类型。
2.2 处理异步计算的结果
当我们获取到异步计算的结果之后,还可以对其进行进一步的处理,比较常用的方法有下面几个:
thenApply()
thenAccept()
thenRun()
whenComplete()
thenApply()
方法接受一个 Function
实例,用它来处理结果。
CompletableFuture<String> future = CompletableFuture.completedFuture("hello!").thenApply(s -> s + "world!");
assertEquals("hello!world!", future.get());
// 这次调用将被忽略。
future.thenApply(s -> s + "nice!");
assertEquals("hello!world!", future.get());
如果你不需要从回调函数中获取返回结果,可以使用 thenAccept()
或者 thenRun()
。这两个方法的区别在于 thenRun()
不能访问异步计算的结果。
CompletableFuture.completedFuture("hello!").thenApply(s -> s + "world!").thenApply(s -> s + "nice!").thenAccept(System.out::println);//hello!world!nice!CompletableFuture.completedFuture("hello!").thenApply(s -> s + "world!").thenApply(s -> s + "nice!").thenRun(() -> System.out.println("hello!"));//hello!
whenComplete()
的方法的参数是 BiConsumer<? super T, ? super Throwable>
public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) {return uniWhenCompleteStage(null, action);
}public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action) {return uniWhenCompleteStage(defaultExecutor(), action);
}
// 使用自定义线程池(推荐)
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor) {return uniWhenCompleteStage(screenExecutor(executor), action);
}
2.3 异常处理
可以通过 handle()
方法来处理任务执行过程中可能出现的抛出异常的情况。
CompletableFuture<String> future= CompletableFuture.supplyAsync(() -> {if (true) {throw new RuntimeException("Computation error!");}return "hello!";
}).handle((res, ex) -> {// res 代表返回的结果// ex 的类型为 Throwable ,代表抛出的异常return res != null ? res : "world!";
});
assertEquals("world!", future.get());
还可以通过 exceptionally()
方法来处理异常情况。
CompletableFuture<String> future= CompletableFuture.supplyAsync(() -> {if (true) {throw new RuntimeException("Computation error!");}return "hello!";
}).exceptionally(ex -> {System.out.println(ex.toString());// CompletionExceptionreturn "world!";
});
assertEquals("world!", future.get());