文章目录
- 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操作上，造成线程饥饿，进而影响整个系统的性能。所以，强烈建议大家根据不同的业务类型创建不同的线程池，以避免互相干扰。
