Vert.x 源码阅读 (1) - Future 和 Promise

Question

这是 Vert.x 项目源码阅读笔记的第一篇，主要讲一下最基础的 Future 和 Promise 的概念。 Future 和 Promise 实际上是异步编程中非常基础的两个抽象，大部分程序员想必都非常熟悉，我这里主要是整理为主。

Vert.x 中的 Future 和 Promise

AsyncResult 接口

Vert.x 中的 Future 扩展了 AsyncResult 。 AsyncResult 有点像一个 Monad ，它包装了一个结果，或者是一个错误。注意 AsyncResult 这个接口并不包含结果还没有准备就绪的状态，这个状态是由 Future 暴露出来的。

Future 接口

Future 逻辑上表达了一个异步计算的结果。它扩展了 AsyncResult 接口，主要提供了两类方法：

1. 判断是否已经完成： isComplete 。
2. 注册回调方法，例如 onComplete 。

等待结果的一方直接面对的，应该是一个 Future 对象。这个对象对于结果使用方来说，是只读的。使用方可以不断的 poll 它直到计算完成，结果 ready。同时使用也可用基于这个 Future 创建一个新的 Future 对象。

Promise 接口

Promise 有点难解释，我觉得我们可以把它理解成 Future 的可写的另一端。实际完成异步计算的结果提供方，在干完活之后，将结果写入这个 Promise 。 Promise 会负责通知自己对应的 Future 结果已经准备就绪了。

Golang 中的 Channel

我感觉 Golang 中的 channel 是一个有点相似的概念。 Future 和 Promise 是 channel 的两端。这也意味着，我们可以用 Future 和 Promise 来做线程间同步，虽然我们不应该怎么做，而应该根据你的情况使用更加适合的，专门用于同步的抽象。

一般的使用场景

在一般情况下， Promise 和 Future 的使用往往符合这样的模式。我们假设存在 A 和 B 两个实体，A 需要 B 执行一个任务，然后将结果告诉自己。为了达到这个目的，一般会经过一下几个步骤：

1. A 触发整个流程，往往是通过调用 B 提供的某个接口，这个接口一般都返回一个 Future 对象。
2. B 的接口中创建了一对 Promise 和 Future 。 Future 会被返回给 A。
3. A 获得返回的 Future 之后一般会在这个 Future 上注册回调函数，表示收到结果之后要做什么。
4. B 在结果计算出来之后调用 Promise::complete 方法，而这个 complete 方法会调用 A 在 Future 上注册的函数。

样例

为了加深理解，这里给出几段代码，用来品品这两个抽象的使用方式。为了方便理解，这里去掉了和 context 相关的代码。

Future 的 map 方法

public Future<O> map(Function<I, O> transformer) {
  Promise<O> ret = Promise.promise();
  this.setHandler(asyncResult -> {
    if (asyncResult.succeed()) {
      ret.complete(transformer.apply(asyncResult.result()));
    } else {
      ret.fail(asyncResult.cause())
    }
  });
  return ret.future();
}

通过 map 的代码，我们可以清楚地发现， Promise 和 Future 非常类似于一个 channel 。我们总是留住一根管子的一端，把另一端交给下游。自己干完活就对着管子吼一声。

CompositeFuture 的实现

这里为了方便理解做了一点简化，实际上 CompositeFuture 是一个接口，真正的实现在 CompositeFutureImpl 中。

public class CompositeFuture implements Future<CompositeFuture> {
  // 自己包含的子 Future 的数量
  private int count;
  // 用来表示自己所有子 Future 完成的 Promise
  private Promise promise = Promise.promise();
  
  public static CompositFuture all(List<Future<?>> futures) {
    CompositFutureImpl composit = new CompositFutureImpl(futures);

    for (Future<?> future : futures) {
      future.setHandler(result -> {
        if (result.succeed()) {
          composit.count ++;
        }
        if (count == composit.len) {
          composit.succeed();
        }
      })
    }
  }
}

CompositFuture 的实现思路也很简单，就是在每一个子 Future 上注册一个回调来通知自己一个子 Future 已经完成。如果所有的子 Future 都完成了，就可以把自己标记成完成。