Java异步编程之CompletableFutur

雨点打透心脏的1/2处 2023-09-24 20:24 59阅读 0赞

## 异步任务 ##

### Future获取异步任务结果 ###

利用 Java 并发包提供的 Future 可以很容易获得异步任务的执行结果，无论异步任务是通过线程池 ThreadPoolExecutor 执行的，还是通过手工创建子线程来执行的。利用多线程可以快速将一些串行的任务并行化，从而提高性能；如果任务之间有依赖关系，比如当前任务依赖前一个任务的执行结果，这种问题基本上都可以用 Future 来解决。

其类结构如下：

publicinterfaceFuture<V> {
 booleancancel(boolean mayInterruptIfRunning);
 
 booleanisCancelled();
 
 booleanisDone();
 
 V get()throws InterruptedException, ExecutionException;
 
 V get(long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException;
 }
 复制代码

方法解释如下：

<table style=""> 
 <tbody> 
 <tr> 
 <td style="width:369px;vertical-align:top;height:51px;">Future</td> 
 <td style="width:369px;vertical-align:top;height:51px;">备注</td> 
 </tr> 
 <tr> 
 <td style="width:369px;vertical-align:top;height:51px;">cancel(boolean)</td> 
 <td style="width:369px;vertical-align:top;height:51px;">尝试取消异步任务的执行。如果任务已经执行完成、已经被取消、因为某种原因不能被取消，则返回false；如果任务正在执行，并且mayInterruptIfRunning为true，那么会调用interrupt()尝试打断任务。该方法返回结果后，isDone()总会返回true</td> 
 </tr> 
 <tr> 
 <td style="width:369px;vertical-align:top;height:51px;">isCancelled()</td> 
 <td style="width:369px;vertical-align:top;height:51px;">如果在任务完成前被取消，返回true</td> 
 </tr> 
 <tr> 
 <td style="width:369px;vertical-align:top;height:51px;">isDone()</td> 
 <td style="width:369px;vertical-align:top;height:51px;">如果任务完成则返回true。任务完成包括正常结束、任务被取消、任务发生异常，都返回true</td> 
 </tr> 
 <tr> 
 <td style="width:369px;vertical-align:top;height:51px;">get()</td> 
 <td style="width:369px;vertical-align:top;height:51px;">获取异步任务执行结果，如果没有返回，则阻塞等待</td> 
 </tr> 
 <tr> 
 <td style="width:369px;vertical-align:top;height:51px;">get(long timeout, TimeUnit unit)</td> 
 <td style="width:369px;vertical-align:top;height:51px;">在给定的时间内等待获取异步任务结果，如果超时还未获取到结果，则会抛出TimeoutException</td> 
 </tr> 
 </tbody> 
</table>

但它本身只是一个接口，还需要看Future的具体实现类：

![format_png][]

红色框起来的就是常见的实现类，就不一一展开赘述了，这里以FutureTask的使用为例：

//初始化一个线程池
 ExecutorService executor = = Executors.newSingleThreadExecutor();
 //new 一个Callable并传入FutureTask
 FutureTask<String> future =new FutureTask<>(newCallable<String>() {
 publicStringcall() {
 //do somethingreturn result;
 }});
 executor.execute(future);
 //在异步任务执行期间可以做一些其他的事情displayOtherThings();
 //通过future.get()得到异步任务执行结果String result=future.get();
 复制代码

## CompletableFuture异步编程 ##

### CompletableFuture的线程池之坑 ###

**CompletableFuture又是如何创建新线程的**？答案是**ForkJoinPool.commonPool()** ，我们熟悉的老朋友又回来了，还是它。当需要新的线程时，CompletableFuture会从commonPool中获取线程，相关源码如下：

publicstatic CompletableFuture<Void> runAsync(Runnable runnable) {
 return asyncRunStage(asyncPool, runnable);
 }
 privatestaticfinalExecutorasyncPool= useCommonPool ? ForkJoinPool.commonPool() : newThreadPerTaskExecutor();
 复制代码

我们已经知道了commonPool的潜在风险，**在生产环境中使用无异于给自己挖坑**。**换句话说，当我决定使用CompletableFuture的时候，默认就是我们要创建自己的线程池**。不要偷懒，更不要存在侥幸心理。

这里补充一点：**CompletableFuture是否使用默认线程池的依据，和机器的CPU核心数有关。**

**当CPU核心数减1大于1时，才会使用默认的线程池（ForkJoinPool），否则将会为每个CompletableFuture的任务创建一个新线程去执行**。

即，CompletableFuture的默认线程池，只有在**双核以上的机器**内才会使用。在双核及以下的机器中，会为每个任务创建一个新线程，**等于没有使用线程池，且有资源耗尽的风险**。

#### 注意事项 ####

1.  如果所有 CompletableFuture 共享一个线程池，那么一旦有任务执行一些很慢的 I/O 操作，就会导致线程池中所有线程都阻塞在 I/O 操作上，从而造成**线程饥饿**，进而影响整个系统的性能。所以，强烈建议你要根据不同的业务类型创建不同的线程池，以避免互相干扰。

1.  记得考虑异常处理。

1.  公共的ForkJoinPool线程池，这个线程池默认创建的线程数是CPU的核数-1（也可以通过JVM option：\*\*-\*\*Djava.util.concurrent.ForkJoinPool.common.parallelism 来设置ForkJoinPool线程池的线程数）

### 创建 CompletableFuture 对象 ###

CompletableFuture的核心方法总共分为四类，而这四类方法又分为两种模式：**同步和异步**。

<table style=""> 
 <tbody> 
 <tr> 
 <td style="width:184px;vertical-align:top;height:51px;">类型</td> 
 <td style="width:184px;vertical-align:top;height:51px;">接收参数</td> 
 <td style="width:184px;vertical-align:top;height:51px;">返回结果</td> 
 <td style="width:184px;vertical-align:top;height:51px;">支持异步</td> 
 </tr> 
 <tr> 
 <td style="width:184px;vertical-align:top;height:51px;">Supply</td> 
 <td style="width:184px;vertical-align:top;height:51px;">✘</td> 
 <td style="width:184px;vertical-align:top;height:51px;">✔︎</td> 
 <td style="width:184px;vertical-align:top;height:51px;">✔︎</td> 
 </tr> 
 <tr> 
 <td style="width:184px;vertical-align:top;height:51px;">Run</td> 
 <td style="width:184px;vertical-align:top;height:51px;">✘</td> 
 <td style="width:184px;vertical-align:top;height:51px;">✘</td> 
 <td style="width:184px;vertical-align:top;height:51px;">✔︎</td> 
 </tr> 
 <tr> 
 <td style="width:184px;vertical-align:top;height:51px;">Accept</td> 
 <td style="width:184px;vertical-align:top;height:51px;">✔︎</td> 
 <td style="width:184px;vertical-align:top;height:51px;">✘</td> 
 <td style="width:184px;vertical-align:top;height:51px;">✔︎</td> 
 </tr> 
 <tr> 
 <td style="width:184px;vertical-align:top;height:51px;">Apply</td> 
 <td style="width:184px;vertical-align:top;height:51px;">✔︎</td> 
 <td style="width:184px;vertical-align:top;height:51px;">✔︎</td> 
 <td style="width:184px;vertical-align:top;height:51px;">✔︎</td> 
 </tr> 
 </tbody> 
</table>

上述接种类型的方法一般都有三个变种方法：**同步**、**异步**和**指定线程池**。比如， **thenApply()** 的三个变种方法如下所示：

CompletableFuture thenApply(Function<? super T,? extends U> fn)
 CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn)
 CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)
 复制代码

第一个同步，第二个异步，第三个支持指定线程池

### 如何理解 CompletionStage 接口 ###

任务是有时序关系的，比如有**串行关系、并行关系、汇聚关系**等。如下图

![format_png 1][]

还要注意参数类型：

* 参数 fn 的类型是接口 Function<T, R>，代表这个方法既能接收参数也支持返回值

*  参数 consumer 的类型是接口 Consumer，代表这个方法虽然支持接收参数，但却不支持返回值

*  参数 action 的类型是 Runnable，代表这个方法既不能接收参数也不支持返回值。

上述参数类型的方法一般都有三个变种方法：同步、异步和指定线程池。方法后缀带 Async 代表的是异步执行

#### 描述串行关系 ####

CompletionStage 接口里面描述串行关系，主要是 thenApply、thenAccept、thenRun 和 thenCompose 这四个系列的接口。

CompletionStage<R> thenApply(fn);
 CompletionStage<R> thenApplyAsync(fn);
 CompletionStage<Void> thenAccept(consumer);
 CompletionStage<Void> thenAcceptAsync(consumer);
 CompletionStage<Void> thenRun(action);
 CompletionStage<Void> thenRunAsync(action);
 CompletionStage<R> thenCompose(fn);
 CompletionStage<R> thenComposeAsync(fn);
 复制代码

#### 描述 AND 汇聚关系 ####

CompletionStage 接口里面描述 AND 汇聚关系，主要是 thenCombine、thenAcceptBoth 和 runAfterBoth 系列的接口。

both：前两阶段同时执行完毕执行下一阶段

CompletionStage<R> thenCombine(other, fn);
 CompletionStage<R> thenCombineAsync(other, fn);
 CompletionStage<Void> thenAcceptBoth(other, consumer);
 CompletionStage<Void> thenAcceptBothAsync(other, consumer);
 CompletionStage<Void> runAfterBoth(other, action);
 CompletionStage<Void> runAfterBothAsync(other, action);
 复制代码

#### 描述 OR 汇聚关系 ####

CompletionStage 接口里面描述 OR 汇聚关系，主要是 applyToEither、acceptEither 和 runAfterEither 系列的接口。

either：前两阶段任一执行完毕执行下一阶段

CompletionStage applyToEither(other, fn);
    CompletionStage applyToEitherAsync(other, fn);
    CompletionStage acceptEither(other, consumer);
    CompletionStage acceptEitherAsync(other, consumer);
    CompletionStage runAfterEither(other, action);
    CompletionStage runAfterEitherAsync(other, action);
    复制代码

案例：李四下班后坐公交回家，可以选择 111 路和 222 路，当其中有一辆公交到达，李四就选择坐这辆车回家，这种场景可以使用 applyToEither 方法

publicclassApplyToEitherTest {
 publicstaticvoidmain(String[] args) {
 PrintTool.printTimeAndThread("李四下班准备回家。。。");
 PrintTool.printTimeAndThread("李四等待111,222公交。。。");
 CompletableFuture<String> busCF = CompletableFuture.supplyAsync(() -> {
 PrintTool.printTimeAndThread("111 路公交在路上。。。");
 PrintTool.sleep(5000);
 return"906";
 }).applyToEither(CompletableFuture.supplyAsync(() -> {
 PrintTool.printTimeAndThread("222 路公交在路上。。。");
 PrintTool.sleep(4000);
 return"539";
 }), first -> first + "路公交");
 PrintTool.printTimeAndThread("李四坐上" + busCF.join());
 }
 }
 复制代码

#### allOf与anyOf ####

**allOf()** 与**anyOf()** 也是一对孪生兄弟，当我们需要对**多个Future**的运行进行组织时，就可以考虑使用它们：

*  **allOf()** ：给定一组任务，等待**所有任务**执行结束；

*  **anyOf()** ：给定一组任务，等待**其中任一**任务执行结束。

### 异常处理 ###

虽然fn、consumer、action它们的核心方法都**不允许抛出可检查异常，但是却无法限制它们抛出运行时异常。**

CompletionStage exceptionally(fn);
 CompletionStage<R> whenComplete(consumer);
 CompletionStage<R> whenCompleteAsync(consumer);
 CompletionStage<R> handle(fn);
 CompletionStage<R> handleAsync(fn);
 复制代码

#### 使用exceptionally()回调处理异常 ####

下面的示例代码展示了如何使用 exceptionally() 方法来处理异常，exceptionally() 的使用非常类似于try\{\}catch\{\}中的 catch\{\}，但是由于支持链式编程方式，所以相对更简单。

CompletableFuture<Integer> f0 = CompletableFuture.supplyAsync(() -> 7 / 0).thenApply(r -> r * 10)
 .exceptionally(e -> 0);复制代码

#### 使用handle()处理异常 ####

既然有 try\{\}catch\{\}，那就一定还有 try\{\}finally\{\}，whenComplete() 和 handle() 系列方法就类似于try\{\}finally\{\}中的 finally\{\}，无论是否发生异常都会执行 whenComplete() 中的回调函数 consumer 和 handle() 中的回调函数 fn。whenComplete() 和 handle() 的区别在于 whenComplete() 不支持返回结果，而 handle() 是支持返回结果的。

CompletableFuture<Integer> f0 = CompletableFuture.supplyAsync(() -> 7 / 0).thenApply(r -> r * 10)
 .handle((res, ex) -> {
 if (ex != null) {
 System.out.println("出错：" + ex.getMessage());
 return1;
 }
 return0;
 });
 复制代码

## CompletionService批量执行异步任务 ##

当需要批量提交异步任务的时候建议你使用 CompletionService。CompletionService 将线程池 Executor 和阻塞队列 BlockingQueue 的功能融合在了一起，能够让批量异步任务的管理更简单。除此之外，CompletionService 能够让异步任务的执行结果有序化，先执行完的先进入阻塞队列，利用这个特性，你可以轻松实现后续处理的有序性，避免无谓的等待，同时还可以快速实现诸如 Forking Cluster 这样的需求。

### 利用 CompletionService 实现 Forking Cluster ###

Dubbo 中有一种叫做**Forking 的集群模式**，这种集群模式下，支持**并行地调用多个查询服务，只要有一个成功返回结果，整个服务就可以返回了**。

利用 CompletionService 可以快速实现 Forking 这种集群模式，比如下面的示例代码就展示了具体是如何实现的。首先我们创建了一个线程池 executor 、一个 CompletionService 对象 cs 和一个Future<Integer>类型的列表 futures，每次通过调用 CompletionService 的 submit() 方法提交一个异步任务，会返回一个 Future 对象，我们把这些 Future 对象保存在列表 futures 中。

通过调用 cs.take().get()，我们能够拿到最快返回的任务执行结果，只要我们拿到一个正确返回的结果，就可以取消所有任务并且返回最终结果了。

// 创建线程池
 ExecutorService executor = Executors.newFixedThreadPool(3);
 // 创建 CompletionService
 CompletionService<Integer> cs = new ExecutorCompletionService<>(executor);
 // 用于保存 Future 对象
 List<Future<Integer>> futures = new ArrayList<>(3);
 // 提交异步任务，并保存 future 到 futures 
 futures.add(cs.submit(() -> geocoderByS1()));
 futures.add(cs.submit(() -> geocoderByS2()));
 futures.add(cs.submit(() -> geocoderByS3()));
 // 获取最快返回的任务执行结果
 Integer r = 0;
 try {
 // 只要有一个成功返回，则 break
 for (int i = 0; i < 3; ++i) {
 r = cs.take().get();
 // 简单地通过判空来检查是否成功返回
 if (r != null) {
 break;
 }
 }
 } finally {
 // 取消所有任务
 for (Future<Integer> f : futures)
 f.cancel(true);
 }
 // 返回结果
 return r;
 复制代码

## 总结 ##

对于简单的并行任务，你可以通过“线程池 +Future”的方案来解决；如果任务之间有聚合关系，无论是 AND 聚合还是 OR 聚合，都可以通过CompletableFuture来解决；而批量的并行任务，则可以通过CompletionService来解决。

作者：Linn

链接：[https://juejin.cn/post/720351860501792364][https_juejin.cn_post_720351860501792364]

[format_png]: https://img-blog.csdnimg.cn/img_convert/92486d869b5fafe07f5b721f59231741.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/775ab998d2ea19d83ad07915e526dd7e.webp?x-oss-process=image/format,png
[https_juejin.cn_post_720351860501792364]: https://juejin.cn/post/7203518605017923643