使用 Vert.X Future/Promise 编写异步代码_在线工具

Future 和 Promise 是 Vert.X 4.0中的重要角色，贯穿了整个 Vert.X 框架。掌握 Future/Promise 的用法，是用好 Vert.X、编写高质量异步代码的基础。本文从 Future/Promise 的概念出发，介绍这两者的定义以及如何理解其定义；然后介绍 Promise 和 Future 相关的 API，结合若干实例介绍如何编写异步代码。

1. 概念

Future 和 Promise 是一个宽泛的概念，很多编程语言都有对这二者的实现。在 Java 中，JDK 有对 Future 的实现，不同的 Java 框架（如：Netty）也有各自的实现。

Vert.X 也实现了 Future 和 Promise，并且有自己的定义：Future 表示某种已经发生或未发生的行为的结果，Promise 表示某种已经发生或未发生的行为的写入端。

A future represents the result of an action that may, or may not, have occurred yet.
A promise represents the writable side of an action that may, or may not, have occurred yet.

这段来自于 Vert.X 源码中的描述比较抽象，网络上对这二者有多种不同的描述。自认为“把 Promise 当做消息的生产者，把 Future 当做消息的消费者”这一描述最好理解。例如下面一段代码，promise 通过 complete() 方法来生产一条消息，future 通过 onSuccess() 来消费一条消息。

Promise<String> promise = Promise.promise();
Future<String> future = promise.future();
promise.complete("Hello"); // 生产消息
future.onSuccess(msg -> System.out.println(msg)); // 消费消息

从 Java 语法上看，Promise 和 Future 是两个接口，它们有共同的实现类，在运行时 Fromise 和与之关联的 Future 是同一个对象。一个 Promise 对象整个生命周期只能够生产一条消息，意味着与之关联的 Future 只能够消费一条消息，但可以多次消费这条消息。

2. Promise

Promise 是一个泛型接口，泛型表示要写入（生产）的消息的类型。

2.1 获取 Promise 实例

获取 Promise 实例是编写 Future/Promise 异步代码的第一步，用户可以通过 Promise 接口提供的静态工厂方法promise()来获取一个 Promise 实例。如下代码获取了一个实例，它可以写入一条 String 消息。

Promise<String> promise = Promise.promise();

2.2 Promise 写入（生产）异步消息

Promise 作为行为的写入端，可以通过 complete 方法来写入一条消息。

promise.complete("Hello");

当然，在发生异常时，也可以通过 fail 方法写入一个异常。如下代码，从 path 中读取一个字符串，若读取成功，则 promise 写入读取到的内容；若读取失败，则写入捕获到的异常。

try {
    String str = Files.readString(path);
    promise.complete(str); // 写入一条消息
} catch(IOException e) {
    promise.fail(e); // 写入一个异常
}

一个 promise 对象只能写入一次消息，要么写入一条正常消息，要么写入一个异常。通常情况下，往一个已经写入过内容的 promise 的对象中继续写入，会抛出 IllegalStateException。除非使用以 "try" 开头的 API 来写入消息，这类 API 通过返回 boolean 值来判断写入是否成功，不抛出 IllegaStateException。

在 Promise 内部，写入消息实质上是把消息或者异常设置到 Promise 实现类的对象中，并通知与 Promise 关联的 Future 来处理消息或异常。

写入（生产）消息的 API 清单如下：

3. Future

Vert.X 官方定义 Future 为已发生或还未发生行为的结果。从另一角度看，相对于生产端的 Promise，Future 又是消息的消费端。Future 是一个比 Promise 更加复杂的抽象，出场率也更高；很多时候会隐去 Promise，只使用 Future 来完成异步操作。

3.1 Future 的状态与结果

Future 有两个状态，已完成（已完成 completed）或未完成两种状态。而按照 Promise 写入的结果类型，又可以将已完成状态分为成功完成和失败完成。Future 提供了 3 个 API 来查询状态。

当 Promise 写入结果时，对应的 Future 状态会发生改变，状态图如下所示。

当状态是成功完成（SUCCEED）时，可以通过 future.result() 来读取一个正常消息；当状态是失败完成（FAILED）时，可以通过 future.cause() 来读取异常。

以上查询状态和读取结果的方法来自于 Future 的父接口 AsyncResult，这是 Future 能够表示异步结果的原因。

3.2 SucceedFuture 与 FailedFuture

与刚创建时状态不确定的普通 Future 不同，SucceedFuture 和 FailedFuture 在创建出来的时候状态就已经确定。Future 提供了静态工厂方法来获取它们的实例，这一类 Future 对象没有与之关联的 Promise，因为消息已经存在，不需要 Promise 来生产消息。

Future<Void> future = Future.succeedFuture();
Future<String> future1 = Future.succeedFuture("hello");
Future<Void> future2 = Future.failedFuture(e);

3.3 Future 处理（消费）异步消息

前面提到，异步消息可能是一条正常消息，也可能是一个异常；当然，Future 本身也表示异步消息（结果）。

Future 提供了 onSuccess, onFailure, 和 onComplete 这几个 API 来处理异步消息。它们的参数都是一个处理器 Handler，也可以看做一个无返回值的函数，这些 API 内部将传入的处理器包装称为监听器，监听 Promise 写入事件。它们返回的结果都是 this，因此可以链式调用这些方法。

如下代码以异步的形式通过 HttpClient 向远程服务器发起请求，获取响应体。如果执行正常，promise 将响应体作为一条正常消息写入；如果发生异常，则写入一个异常。后面与之关联的 future 设置了 3 个处理器，当请求成功时，执行 onSuccess 设置的处理器；当请求失败时，执行 onFailure() 设置的处理器；无论请求成功还是失败，都将执行 onComplete() 设置的处理器。

Promise<String> promise = Promise.promise();
new Thread(() -> {
    try {
	    String responseBody = httpClient.request("http://www.test.com/user");        
        promise.complete(responseBody); // 写入正常消息
    } catch(Exception e) {
        promise.fail(e); // 写入异常
    }
}).start();

Future<String> future = promise.future();
future.onSuccess(body -> System.out.println("获取到 Response Body:" + body)) // 成功时执行
	  .onFailure(cause -> cause.printStacktrace()) // 失败时执行
	  .onComplete(asyncResult -> { // 无论成功失败都执行
    	if (asyncResult.succeed()) {
        	logger.info("成功获取到 Reponse Body");
	    } else {
    	    logger.error("获取 Response Body 失败。", asyncResult.cause());
    	}
	  });

与只能写入一次结果的 Promise 不同，Future 可以设置任意多个处理器，或者说可以多次消费消息。

future.onSuccess(result -> // 处理 1)
      .onSuccess(result -> // 处理 2);

另外，无论当前 Future 的状态是成功、失败或是未完成，都可以调用这些 API 来设置处理器（Handler）。如果 Future 是未完成状态，这些处理器会在 Promise 写入结果之后触发执行；如果 Future 是已完成状态，在设置处理器时立即执行它们。

如下是这几个 API 的说明。

3.4 Future 转换

Future 转换是指将一个 Future 对象经过处理之后转换为另一个 Future 对象。Future 转换通过 Future 的一系列成员方法来完成，这些转换 API 的参数是一个 Function， 返回值是另一个 Future 对象。

看下面一段代码，根据名字查询到对应的用户 User，然后根据用户 ID 查询账户信息 Account，再从账户中获取余额，最后打印余额。

Future<String> f1 = Future.succeedFuture("Robothy"); // 1
Future<User> f2 = f1.compose(name -> getUserByName(name)); // 2
Future<Account> f3 = f2.compose(user -> getAccountInfo(user.getId())); // 3
Future<Double> f4 = f3.map(account -> account.getBalance()); // 4
f4.onSuccess(balance -> System.out.println("Balance: " + balance)) // 5
  .onFailure(cause -> cause.printStacktrace() ); // 6

1. 代码获取了一个 SucceedFuture 对象 f1，它的结果是正常消息 "Robothy"。
2. f1 调用 compose 方法，设置一个异步函数，即该行代码中的 Lambda 表达式。函数会在 f1 正常完成的时候执行，以异步的方式根据用户名获取用户信息，并将用户信息设置到 f2 中。这行代码将 f1 转化成了 f2。
3. f2 继续调用 compose 方法，设置另一个异步函数，根据用户 ID 获取账户信息。同样，该函数在 f2 正常完成之后执行，执行结果设置到 f3 中，即：把 f2 转化为 f3。
4. f3 调用 map 方法，设置一个同步函数，从账户信息中读取余额。同步函数在 f3 正常完成之后执行，执行结果会被写入到 f4 中。
5. 为 f4 设置正常消息处理器，处理器在 f4 正常完成之后被调用，打印余额信息。
6. 为 f4 设置异常处理器，处理器在 f4 异常完成之后被调用，打印栈信息。

为了方便描述，上面代码被拆成了多个部分，采用链式调用可以使代码更加简洁。

Future<String> f1 = Future.succeedFuture("Robothy") // 1
    .compose(name -> getUserByName(name)) // 2
	.compose(user -> getAccountInfo(user.getId())) // 3
	.map(account -> account.getBalance()) // 4
	.onSuccess(balance -> System.out.println("Balance: " + balance)) // 5
 	.onFailure(cause -> cause.printStacktrace() ); // 6

上面提到的同步函数和异步函数都是 Function 类型，同步函数是同步执行，可以返回任意结果；而异步函数形式上异步执行，返回的是一个 Future。之所以说形式上异步执行，是因为实际上是否异步取决于实现。例如：getUserByName 的方法签名如下，但是它方法体的实现可以是同步的，也可以是异步的。

Future<User> getUserByName(String name);
Future<Account> getAccountInfo(String userId); // 同理

compose 设置的是一个异步函数，函数被调用之后，立即返回一个 Future，但此时异步函数返回的 Future （区别于 compose 返回的 Future）并不一定已完成。所以并不能向下面这段代码一样，直接读取结果，正确的做法是像上面一样通过 compose 将异步函数连接起来。

Future<User> future = getUserByName(name);
User user = future.result(); // 立即读取，此时 future 未完成，读取到的是 null
Future<Account> account = getAccount(user.getId()); // 抛出 NullPointerException

类似地，map 里面设置的是同步函数，函数被调用之后，返回的结果就是最终结果。getBalance 是 Account 的成员方法，它的方法签名如下：

Double getBalance();

此外，如果 getUserByName 有异常抛出，或者它返回的 Future 包含了一个异常消息，则后续的 compose 和 map 设置的函数都不会被执行。也就是说，只有 2, 6 处设置的函数和处理器会执行。同理，如果 getAccountInfo 发生异常，2,3,6 处的代码会被执行；如果调用链设置的各个函数执行正常，则 2,3,4,5 的函数或处理器会被执行。

这种执行方式与同步代码块中的 try-catch 很相似，假如 getUserByName 和 getAccoutInfo 都是同步的，则代码可以用 try-catch 表达如下。

try {
    String name = "Robothy"; // 1
    User user = getUserByName(name); // 2，注意这里是同步的，返回的是 User，不是 Future<User>
    Account account = getAccountInfo(user.getId()); // 3，返回 Account，不是 Future<Account>
    Double balance = account.getBalance(); // 4
    System.out.println("Balance: " + balance); // 5
} catch(Throwable cause) {
    cause.printStacktrace(); // 6
}

map 和 compose 会在遇到异常时会跳过执行，而 recover 与之相反，只有在遇到异常时才执行。比如把上面代码的效果修改一下：如果获取账户余额的过程中失败，则默认用户余额是 0。代码可以这样写：

Future<String> f1 = Future.succeedFuture("Robothy")
    .compose(name -> getUserByName(name))
	.compose(user -> getAccountInfo(user.getId()))
	.map(account -> account.getBalance())
    .recover(cause -> Future.succeedFuture(0.0)) // 拦截异常，把余额设置成 0.
	.onSuccess(balance -> System.out.println("Balance: " + balance))
 	.onFailure(cause -> cause.printStacktrace() );

Future 还有其他的转换操作，这里不一一赘述，下面会给出 API 清单。

Future 转换 API 和前面的 Future 结果处理 API 很相似，内部都是将参数包装成监听器，都可以链式调用，区别在于 Future 结果处理 API 返回的是 this，而转换 API 返回的是另一个 Future 对象。在实际开发中，通常在调用链的尾部使用处理 API，而在中间使用转换 API。

4. 小结

以上是 Vert.X Promise/Future 的基本概念和用法。为了方便理解，可以把 Promise 看做生产者，Future 看做消费者。生产者 Promise 即可以生产正常消息，也可以生产一个异常，消费者 Future 可以消费正常消息，也可以消费异常。一个 Promise 对象整个生命周期只能够写入一次消息，而对应的 Future 可以消费这条消息多次。

Future 作为异步结果（AsyncReuslt）时有状态，当状态是成功完成时，可以读取正常结果（消息），当状态是失败完成时，可以读取一个异常。SucceedFuture 和 FailedFuture 是两种特殊的 Future，它们在实例化之后就是已完成状态。

Future 作为消费者提供了结果处理 API 和 Future 转换 API。结果处理 API 可以设置处理器 Handler 来处理相应的结果，转换 API 可以设置函数 Function 把当前 Future 转化为另一个 Future 对象。

Promise/Future 本身比较抽象，异步编程也是个技术难点。 本文仅仅包含了 Promise/Future 最基本的用法，要熟练掌握它们的用法，写出高效、优雅的异步代码还需要进行大量的练习。