Java异步编程之CompletableFuture

忘是亡心i 2023-09-24 20:25 125阅读 0赞

# 异步任务 #

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

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

其类结构如下：

public interface Future<V> {
      boolean cancel(boolean mayInterruptIfRunning);
      
      boolean isCancelled();
    
      boolean isDone();
    
      V get() throws InterruptedException, ExecutionException;
    
      V get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException;
     }
    复制代码

方法解释如下：

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

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

![format_png][]

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

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

# CompletableFuture异步编程 #

## CompletableFuture的线程池之坑 ##

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

public static CompletableFuture<Void> runAsync(Runnable runnable) {
        return asyncRunStage(asyncPool, runnable);
    }
    private static final Executor asyncPool = useCommonPool ? ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
    复制代码

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

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

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

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

### 注意事项 ###

1.  如果所有 CompletableFuture 共享一个线程池，那么一旦有任务执行一些很慢的 I/O 操作，就会导致线程池中所有线程都阻塞在 I/O 操作上，从而造成**线程饥饿**，进而影响整个系统的性能。所以，强烈建议你要根据不同的业务类型创建不同的线程池，以避免互相干扰。
2.  记得考虑异常处理。
3.  公共的ForkJoinPool线程池，这个线程池默认创建的线程数是CPU的核数-1（也可以通过JVM option：\*\*-\*\*Djava.util.concurrent.ForkJoinPool.common.parallelism 来设置ForkJoinPool线程池的线程数）

## 创建 CompletableFuture 对象 ##

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

<table> 
 <thead> 
  <tr> 
   <th><strong>类型</strong></th> 
   <th><strong>接收参数</strong></th> 
   <th><strong>返回结果</strong></th> 
   <th><strong>支持异步</strong></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><strong>Supply</strong></td> 
   <td>✘</td> 
   <td>✔︎</td> 
   <td>✔︎</td> 
  </tr> 
  <tr> 
   <td><strong>Run</strong></td> 
   <td>✘</td> 
   <td>✘</td> 
   <td>✔︎</td> 
  </tr> 
  <tr> 
   <td><strong>Accept</strong></td> 
   <td>✔︎</td> 
   <td>✘</td> 
   <td>✔︎</td> 
  </tr> 
  <tr> 
   <td><strong>Apply</strong></td> 
   <td>✔︎</td> 
   <td>✔︎</td> 
   <td>✔︎</td> 
  </tr> 
 </tbody> 
</table>

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

<U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
    <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
    <U> CompletableFuture<U> 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 方法

public class ApplyToEitherTest {
        public static void main(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());
                    return 1;
                }
                return 0;
            });
    复制代码

# 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`来解决。

[format_png]: https://img-blog.csdnimg.cn/img_convert/cffc0aaca1e6f6f1a8fde5059afe2ab0.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/a4c64f8f4bc3593a589cadb4699c3981.webp?x-oss-process=image/format,png