document.addeventlistener方法不执行_CompletableFuture 的执行模型

忘是亡心i 2023-01-01 10:50 156阅读 0赞

许多现代的 Java 多线程程序都广泛使用了 `CompletableFuture` 抽象作为描述异步任务的方法，Apache Flink 和 Pravega 等项目中都能看到这个 Java 8 才被引进的新的并发工具的身影。由于 `CompletableFuture` 混用了 Future/Promise 并发模型中的 Future 和 Promise，即它既是 Future 又是 Promise，并且出于下文提到的原因拥有数十个命名各异的方法，又支持高度定制化的执行策略，甫一接触这个工具的同学可能会对推理异步方法的执行会有困难。本文从方法命名、执行模型和异常处理三个方向切入，旨在帮助理解和使用 `CompletableFuture`，并在阅读代码的时候能够分析 `CompletableFuture` 所描述的任务的执行过程。

### 方法命名 ###

`CompletableFuture` 用于描述和组合异步任务的方法被抽象为 `CompletionStage` 接口，`CompletionStage` 代表了一个可能被异步执行的计算。`CompletionStage` 拥有超过三十个方法，不过正如下面所做的归类方式，我们会看到其实所能表现的组合逻辑并不多，甚至还有缺失，需要这么多个方法的原因来源于 Java 语言的若干局限性。

由于异步计算可能失败或者说抛出异常，所以组合 `CompletionStage` 的方法广泛地说可以分为处理异步计算正常完成的逻辑和异常完成的逻辑。

### 正常完成的异步计算 ###

我们先看到正常完成异步计算的逻辑。总的来说处理正常完成的异步计算的逻辑又分为三种

*  一种是 then 的逻辑，即前一个计算完成的时候调度后一个计算
 *  一种是 both 的逻辑，即等待两个计算都完成之后执行下一个计算，只要能组合一个和另一个，我们就可以无限复用这个 +1 的逻辑组合任意多的计算
 *  另一种是 either 的逻辑，即等待两个计算的其中一个完成之后执行下一个计算。注意这样的计算可以说是非确定性的。因为被组合的两个计算中先触发下一个计算执行的那个会被作为前一个计算，而这两个前置的计算到底哪一个先完成是不可预知的

从依赖关系的层面来说分成三种类型，我们换个角度从组合函数来看，也分为若干种类型

*  apply 字样的方式意味着组合方式是 `Function`，即接受前一个计算的结果，应用函数之后返回一个新的结果
 *  accept 字样的方式意味着组合方式是 `Consumer`，即接受前一个计算的结果，执行消费后不返回有意义的值
 *  run 字样的方式意味着组合方式是 `Runnable`，即忽略前一个计算的结果，仅等待它完成后执行动作

从枚举的角度讲，我们在参数和返回值上均存在有或无两种状态，那么看起来还缺少一个忽略前一个计算的结果，返回一个值的组合方式。我们知道在 Java 8 的函数式编程支持抽象中，这被称为 `Supplier`。在 `CompletionStage` 中，由于组合的前一个计算一定有结果（如果是 `Consumer` 或 `Runnable`，会返回一个标注为 `Void` 的 `null`），所以我们可以使用 `Function` 并在函数体内忽略入参来模拟 `Supplier` 的语义。当然，一个自然的想法是我们也可以用 `Consumer` 来模拟 `Runnable` 的语义，不过这可能是考虑到现有代码存在大量实现了 `Runnable` 这个 1.0 就有的接口，为了简单的复用旧有代码而添加的语法糖。毕竟就算我们加了一个 `thenSupply` 也不会影响现有接口，所以在这个方面的问题更像是实现上的体现出随意性的细节。

此外，这种区分还体现了 Java 对单元值（例如 Scala 中的 `Unit`）支持的缺乏。因为返回值是 `void` 就必须特殊处理一个新的方法而不能在最后返回一个单元值，甚至 `Void` 类型没有单例，而是用 null 底类型模拟。推开来说，`Stream` 库中针对 void 和基础类型的特殊处理也是受到了这个局限性的影响。

这两种方法区分的方式做个乘积，在和下节将要讲到的两种不同执行模型的三种签名方式做个乘积，`CompletionStage` 的方法数量一下就达到了 3 x 3 x 3 = 27 种！不同的执行模型实际上可以表示为两种方法甚至结合 Java 9 获取 default executor 的方式调整成一种方法。这一点上的方法数的膨胀就源于 Java 不支持类似 Python 的可选参数导致的同名同义函数为了实现调用方的可选参数进行多次重载。我们在这里给出一个小结的表格，看看上面这两种区分方法做乘积后如何对应到具体的方法名称上。

![373344e2e095276f29b787a5d8c485b6.png][]

当然，我们前面提到组合方法超过三十种，除了下面将要提到的处理计算失败的方法之外，还有一个重要的处理正常完成的计算的方法，也就是 `thenCompose` 方法。我们首先看到它的方法签名

public CompletionStage thenCompose
 (Function<? super T, ? extends CompletionStage> fn);

我们用文字来表述这个方法做的事情。它接受前一个计算的结果，返回一个返回新类型的异步计算的 `CompletionStage`，并将这个新的异步计算的 Future 作为整个方法的返回值。请注意，方法的签名非常的重要！

我们乍一看这个方法跟 `thenApply` 的区别不是特别大，因为都是接受一个参数返回一个值的模式。但是我们考虑一下，当接受的 `Function` 签名是 `Function<? super T, ? extends CompletionStage>` 的时候，`thenApply` 的返回值类型将是 `CompletionStage<CompletionStage>`！如果这个拼接反复的进行下去，我们将面临每次组合 `thenApply` 的前后都要手动处理这多出来的 `CompletionStage` 的壳的工作。毕竟我们实际关心的是类型为 `U` 的值，`CompletionStage` 这个壳只是用于表示其异步计算的性质和组合不同异步计算的类型通行证。而 `thenCompose` 方法恰好就为我们解决了这个用户层面处理壳的逻辑，帮助我们专注在逻辑代码上。可能有同学会说，我们 `thenApply` 里直接返回 `U` 类型的值就好了啊。这里需要说明的是，组合并不总是简单的 lambda，还有复用现有函数。我们编写各个函数的时候，出于独立性，返回异步计算的值的函数其签名为 `CompletableFuture<?>` 是很自然的，因此我们在组合的时候很容易组合到 `Function<? super T, ? extends CompletionStage>` 这样签名的函数。

顺带一提，在函数式编程的领域中，`thenApply` 对应了函子（Functor）的 `map` 方法，而 `thenCompose` 对应了单子（Monad）的 `flatMap`（Haskell 中称为 `bind`）方法。所以其实 Monad 也并不是很遥远很可怕的东西，而是实际开发中实实在在会接触的概念。关于 Monad 在实际开发中的应用，也可以参考这篇文章。

### 失败的异步计算 ###

这里先从方法命名上介绍 `CompletionStage` 处理失败的异步计算的概要，其中的细节将放在异常处理这一节中展开讨论。这里的命名就没有太多的规则可寻了（随意性），我们直接看到方法签名

public CompletionStage<T> whenComplete
 (BiConsumer<? super T, ? super Throwable> action);
 
 public CompletionStage handle
 (BiFunction<? super T, Throwable, ? extends U> fn);
 
 public CompletionStage<T> exceptionally
 (Function<Throwable, ? extends T> fn);

描述性地说，`whenComplete` 在方法结束后插入一个动作，并返回原来的计算结果；`handle` 在方法结束后插入一个函数，返回一个可能是新的类型的计算结果；`exceptionally` 针对失败的计算的具体异常，返回一个和原先计算结果类型相符合的值，其语义类似于从异常中恢复。其中，对于 `whenComplete` 和 `handle` 的二元入参来说，恰有一个为 `null` 一个非 `null`，这一点实际上类似于 Go 语言的异常处理。我们会在异常处理中看到这几个方法的执行细节和局限性。

最后一个方法是 `toCompletableFuture`，用于其他 `CompletionStage` 的实现和 Java 内部实现的互操作。例子可以参考 Apache Curator 的 curator-x-async 模块。

### 执行模型 ###

`CompletableFuture` 的执行模型讨论的是组合起来的计算依次在哪个线程上执行的问题。

这件事情说起来很简单，针对三种不同的签名分开来讲就可以了。但是由于没有 Async 后缀的版本和有 Async 后缀的版本仅在方法名上有一点点的区别，以及有 Async 后缀的版本还提供了一个可以不传 `Executor` 的方法，所以开发者很容易就忽略了短短不超过十个字符的差异带来的巨大的变化。

*  没有 Async 后缀的版本，后一个计算将在前一个计算完成时紧接着进行。这里有一个非常不幸的不确定性，如果前一个计算在 thread-1 上进行，执行组合方法（例如 `thenApply`），如果前一个计算在执行组合方法时还没有完成，那么完成后后一个计算就在 thread-1 上紧接着进行。如果在执行组合方法时前一个计算已经完成，那么后一个计算将在执行组合方法的线程上执行。
 *  有 Async 后缀但不传递 `Executor` 的版本，后一个计算将在前一个计算完成时由默认的 `Executor` 执行。这个 `Executor` 可能是 `ForkJoinPool#commonPool`，但如果这个线程池不支持并发也就是并发度不大于 1 的话，就会为每个计算都起一个新的 Thread 来执行。
 *  有 Async 后缀且传递 `Executor` 的版本，跟不传递的版本类似，只是使用用户传递的 `Executor` 来执行。

注意所谓有 Async 后缀的版本使用 `Executor` 来执行，这句话的意思是调用 `Executor.execute` 来执行对应的计算体。通常来说， `Executor` 的 `execute` 方法会将计算体调度到异步线程上执行。但也有可能是非常定制化的执行方式，例如 FLINK 代码中 `RpcEndpoint` 的 `MainThreadExecutor` 目前仅有的实现会将计算以 akka 中 actor 的 tell 模式发送出去；甚至有可能立即执行，例如 FLINK 代码中的 `o.a.f.runtiem.concurrent.Executors#directExecutor`。

### 异常处理 ###

如果计算永不出错，包括远端过程调用总能及时正确地返回，那么这个世界将变得更加美好。可惜现实并非如此。

异步的计算也有可能失败，我们暂时不考虑超时或死锁等活跃性的问题。如果超时抛出一个异常，那么它和其他的异常情况就没什么不同；如果无限等待，那么计算机无从得知计算是否只是一时延迟，将在遥远的将来完成。

针对抛出异常的失败情况，从第一节中我们看到 Java 的 `CompletableFuture` 提供了三个方法，`whenComplete`、`handle` 和 `exceptionally`，也介绍了它们的描述性语义。但是这几种方法实际上有诸多不足。

首先，`whenComplete` 和 `handle` 采用了类似于 Go 语言的异常处理方案。通过同时传递两个入参其中有且仅有一个为 `null` 的方式来区分计算正常完成或异常完成。这种方式是非常土味的，依赖于不可编译期检查的约定。导致这一问题的主要原因是缺少一个 Union Type 的概念。在 Scala 中，可失败的计算结果以 `Try` 类型出现。可以将 `Try` 类型的数据处理作为基础实现，把确定成功过失败作为特例来实现。其次，`handle` 和 `exceptionally` 均只支持返回值为 `U` 类型的组合函数，注意我们无法免费的组合 `CompletionStage` 类型返回值的函数并简单的脱掉多余的壳。特别是针对 `exceptionally`，这点在 Scala 上实现为 `recover` 和 `recoeverWith`，而 Java 对应的只有 `recover`，缺少了 `recoverWith`。总之，`CompletableFuture` 的实现在正常完成的逻辑上是健全的，但是错误处理上是不够重视的，缺少不少开箱即用的功能。Java 9 中加入的静态方法 `failedFuture` 是这一点的最好佐证，既然考虑到了 `completedFuture` 居然没考虑到对应的 `failedFuture`，真是难以置信。

这节正文的最后讨论一下 `CompletableFuture` 作为新的异步计算工具带来的新的异常及其处理。

`CompletableFuture` 是兼容 Java 1.5 的 `Future` 接口的，所以复用此前接口的 `get` 等方法将会在计算失败时返回 `ExecutionException`，并持有一个真正失败原因的 cause。不过 `CompletableFuture` 真正自带的异常体系是基于 `CompletionException` 的，使用 `join` 作为 `get` 的等价替换调用将抛出 `CompletionException` 替代 `ExecutionException` 作为壳的异常。这里需要指出的是，`CompletableFuture` 的 `encodeThrowable` 方法不会对 `CompletionException` 再加封装，而 `CompletionException` 是 `RuntimeException`，因此可以在 `thenCompose` 等组合子的 lambda 中用 `CompletionException` 来封装异常从而解决 `Runnable` 等接口要求实现方法不能抛出 checked exception 的问题。

### 一点故事 ###

其实，`CompletableFuture` 代表的 Future/Promise 这套并发抽象存在已久，在 Scala 的早期版本和 C\# 中早已实现，不得不说 Java 在并发支持上的演进速度慢得惊人。Java 1.5 的 `Future` 虽然能抽象出一个异步执行的计算，但是获取结果却不得不是阻塞的，可以说是骇人听闻。

至于再之前的 `InterruptException` 方法就更难顶了，可以参考 Apache ZooKeeper 里部分 Thread 的子类的方法。其主要思想是一个线程做着工作，外部产生了一个需要这个线程处理的事件（通常是状态变化），就把工作线程中断。中断的工作线程一脸懵逼地检查状态，看看有没有啥需要做的。这种机制跟 C 的错误码有点像，不过一个干燥的 `InterruptException` 只是告诉你出事了，甚至不能像错误码那样传递一点有限的信息。

[373344e2e095276f29b787a5d8c485b6.png]: /images/20221120/fdcb2394a80e4add82ae8d263418f283.png