【Java异步编程】CompletableFuture基础（1）：创建不同线程的子任务、子任务链式调用与异常处理
文章目录
- 1. 三种实现接口
- 2. 链式调用：保证链的顺序性与异步性
- 3. CompletableFuture创建CompletionStage子任务
- 4. 处理异常
- a. 创建回调钩子
- b. 调用handle()方法统一处理异常和结果
- 5. 如何选择线程池：不同的业务选择不同的线程池
CompletableFuture是JDK 1.8引入的实现类，该类实现了Future和CompletionStage两个接口。该类的实例作为一个异步任务，可以在自己异步执行完成之后触发一些其他的异步任务，从而达到异步回调的效果。
CompletionStage代表异步计算过程中的某一个阶段，一个阶段完成以后可能会进入另一个阶段。一个阶段可以理解为一个子任务，每一个子任务会包装一个Java函数式接口实例，表示该子任务所要执行的操作。
1. 三种实现接口
每个CompletionStage子任务所包装的可以是一个Function、Consumer或者Runnable函数式接口实例。
这三个常用的函数式接口的特点如下：
| 被包装接口 | 功能描述 |
|---|---|
| Function | Function接口的特点是：有输入、有输出。包装了Function实例的CompletionStage子任务需要一个输入参数，并会产生一个输出结果到下一步。 |
| Runnable | Runnable接口的特点是：无输入、无输出。包装了Runnable实例的CompletionStage子任务既不需要任何输入参数，又不会产生任何输出。 |
| Consumer | Consumer接口的特点是：有输入、无输出。包装了Consumer实例的CompletionStage子任务需要一个输入参数，但不会产生任何输出。 |
2. 链式调用：保证链的顺序性与异步性
多个CompletionStage构成了一条任务流水线，一个环节执行完成了可以将结果移交给下一个环节（子任务）。多个CompletionStage子任务之间可以使用链式调用。
下面是一个顺序调用的例子：
使用
CompletionStage及其方法构建了一个异步任务链，thenApply用于对前一个阶段的结果进行计算并传递结果，thenAccept用于消费前一个阶段的结果并执行操作，thenRun用于执行无输入输出的操作。
oneStage
//被thenApply包装CompletionStage子任务,由输入输出
.thenApply(x -> square(x))
//消耗上游输出，但是没有输出
.thenAccept(y -> System.out.println(y))
//不消耗上一个子任务的输出又不产生结果
.thenRun(() -> System.out.println())
这种链式操作可以方便地将多个异步操作连接起来，同时保证了操作的顺序性和异步性，提高了代码的可维护性和并发性能。
接下来是一个异步调用的例子：
在这个例子中，task2 和 task3 都依赖于 task1 完成后执行，但它们可能并行执行，也就是说，task2 和 task3 的执行顺序是不确定的，它们不一定会按照 thenRunAsync 的顺序执行。
CompletableFuture<Void> task1 = CompletableFuture.runAsync(() -> {
System.out.println("Task 1");
});
CompletableFuture<Void> task2 = task1.thenRunAsync(() -> {
System.out.println("Task 2");
});
CompletableFuture<Void> task3 = task1.thenRunAsync(() -> {
System.out.println("Task 3");
});
3. CompletableFuture创建CompletionStage子任务
CompletableFuture定义了一组方法用于创建CompletionStage子任务（或者阶段性任务），基础的方法如下：
//子任务包装一个Runnable实例，并调用ForkJoinPool.commonPool()线程池来执行
public static CompletableFuture<Void> runAsync(Runnable runnable)
//子任务包装一个Runnable实例，并调用指定的executor线程池来执行
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
//子任务包装一个Supplier实例，并调用ForkJoinPool.commonPool()线程池来执行
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
//子任务包装一个Supplier实例，并使用指定的executor线程池来执行
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
其中主要注意的信息是：
- Supplier 表示一个无参数但有返回值的函数，Runnable表示无惨无返回值的函数
- 在使用CompletableFuture创建CompletionStage子任务时，如果没有指定Executor线程池，在默认情况下CompletionStage会使用公共的ForkJoinPool线程池。
- 它们都会交给线程池执行，get方法会堵塞主线程等待执行结果。
给一个例子：
//无返回值异步调用
@Test
public void runAsyncDemo() throws Exception {
CompletableFuture<Void> future = CompletableFuture.runAsync(() ->
{
sleepSeconds(1);//模拟执行1秒
Print.tco("run end ...");
});
//等待异步任务执行完成,最多等待2秒
future.get(2, TimeUnit.SECONDS);
}
//有返回值异步调用
@Test
public void supplyAsyncDemo() throws Exception {
CompletableFuture<Long> future = CompletableFuture.supplyAsync(() ->
{
long start = System.currentTimeMillis();
sleepSeconds(1);//模拟执行1秒
Print.tco("run end ...");
return System.currentTimeMillis() - start;
});
//等待异步任务执行完成,现时等待2秒
long time = future.get(2, TimeUnit.SECONDS);
Print.tco("异步执行耗时（秒） = " + time / 1000);
}
4. 处理异常
a. 创建回调钩子
可以为CompletionStage子任务设置特定的回调钩子，当计算结果完成或者抛出异常的时候，执行这些特定的回调钩子。
//设置子任务完成时的回调钩子
public CompletableFuture<T> whenComplete(
BiConsumer<? super T,? super Throwable> action)
//设置子任务完成时的回调钩子，可能不在同一线程执行
public CompletableFuture<T> whenCompleteAsync(
BiConsumer<? super T,? super Throwable> action)
//设置子任务完成时的回调钩子，提交给线程池executor执行
public CompletableFuture<T> whenCompleteAsync(
BiConsumer<? super T,? super Throwable> action,
Executor executor)
//设置异常处理的回调钩子
public CompletableFuture<T> exceptionally(
Function<Throwable,? extends T> fn)
@Test
public void whenCompleteDemo() throws Exception {
CompletableFuture<Void> future = CompletableFuture.runAsync(() ->
{
sleepSeconds(1);//模拟执行1秒
Print.tco("抛出异常！");
throw new RuntimeException("发生异常");
//Print.tco("run end ...");
});
//设置执行完成后的回调钩子
future.whenComplete(new BiConsumer<Void, Throwable>() {
@Override
public void accept(Void t, Throwable action) {
Print.tco("执行完成！");
}
});
//设置发生异常后的回调钩子
future.exceptionally(new Function<Throwable, Void>() {
@Override
public Void apply(Throwable t) {
Print.tco("执行失败！" + t.getMessage());
return null;
}
});
future.get();
}
[ForkJoinPool.commonPool-worker-9]：抛出异常！
[main]：执行完成！
[ForkJoinPool.commonPool-worker-9]：执行失败！java.lang.RuntimeException: 发生异常
java.util.concurrent.ExecutionException: java.lang.RuntimeException: 发生异常
有如下几个注意点：
- 调用cancel()方法取消CompletableFuture时，任务被视为异常完成，completeExceptionally()方法所设置的异常回调钩子也会被执行。
- 如果没有设置异常回调钩子，发生内部异常时会有两种情况发生：
- 在调用get()时，如果遇到内部异常，get()方法就会抛出ExecutionException（执行异常）。
- 在调用join()和getNow(T)启动任务时，如果遇到内部异常，join()和getNow(T)方法就会抛出CompletionException。
b. 调用handle()方法统一处理异常和结果
//在执行任务的同一个线程中处理异常和结果
public <U> CompletionStage<U> handle(
BiFunction<? super T, Throwable, ? extends U> fn);
//可能不在执行任务的**同一个线程**中处理异常和结果
public <U> CompletionStage<U> handleAsync(
BiFunction<? super T, Throwable, ? extends U> fn);
//在指定线程池executor中处理异常和结果
public <U> CompletionStage<U> handleAsync(
BiFunction<? super T, Throwable, ? extends U> fn,
Executor executor);
@Test
public void handleDemo() throws Exception {
CompletableFuture<Void> future = CompletableFuture.runAsync(() ->
{
sleepSeconds(1);//模拟执行1秒
Print.tco("抛出异常！");
throw new RuntimeException("发生异常");
//Print.tco("run end ...");
});
//设置执行完成后的回调钩子
future.handle(new BiFunction<Void, Throwable, Void>() {
@Override
public Void apply(Void input, Throwable throwable) {
if (throwable == null) {
Print.tcfo("没有发生异常！");
} else {
Print.tcfo("sorry,发生了异常！");
}
return null;
}
});
future.get();
}
//
//[ForkJoinPool.commonPool-worker-1]：抛出异常！
//[ForkJoinPool.commonPool-worker-1|CompletableFutureDemo$3.apply]: sorry,发生了异常！
5. 如何选择线程池：不同的业务选择不同的线程池
默认情况下，通过静态方法runAsync()、supplyAsync()创建的CompletableFuture任务会使用公共的ForkJoinPool线程池，默认的线程数是CPU的核数。当然，它的线程数可以通过以下JVM参数设置
option:-Djava.util.concurrent.ForkJoinPool.common.parallelism
如果所有CompletableFuture共享一个线程池，那么一旦有任务执行一些很慢的IO操作，就会导致线程池中的所有线程都阻塞在IO操作上，造成线程饥饿，进而影响整个系统的性能。所以，强烈建议大家根据不同的业务类型创建不同的线程池，以避免互相干扰。