Vert.x 实战（二）：Event Bus、Future 与订单状态编排

第一篇完成了工程和 Verticle 生命周期。本篇把 HTTP、订单与库存连接成一条可以成功、拒绝或超时的异步链路。

用 request/reply 表达订单命令

创建订单必须得到明确结果，因此使用 Event Bus 的 request/reply，而不是不等待结果的 send 或广播用的 publish：

1 2	public static final String CREATE_ADDRESS = "orders.create"; public static final String FIND_ADDRESS = "orders.find";

订单消费者先校验输入：

vertx.eventBus().<JsonObject>consumer(CREATE_ADDRESS, message -> {
  JsonObject request = message.body();
  String validationError = validate(request);
  if (validationError != null) {
    message.fail(400, validationError);
    return;
  }
  // 预留库存并创建订单
});

这里使用 JsonObject 是为了让第一版代码容易观察，不表示大型项目应该到处传无类型 JSON。字段增多后，可以使用领域对象和 MessageCodec，也可以保留 JSON 协议并增加 Schema 校验。

把业务失败转换为 HTTP 状态

消费者的 message.fail(400, ...) 在请求侧表现为 ReplyException。HTTP 层读取 failure code，而不是把所有失败都改成 500：

private static void handleFailure(RoutingContext ctx, Throwable error) {
  int status = error instanceof ReplyException reply
      ? reply.failureCode()
      : 500;
  respond(ctx, status, new JsonObject().put("error", error.getMessage()));
}

业务校验失败返回 400，库存冲突返回 409，内部依赖不可用返回 503。状态码决定客户端是修正请求、提示用户还是稍后重试。

在创建订单前预留库存

部署顺序先满足内部依赖，再开放 HTTP：

1
2
3

vertx.deployVerticle(new InventoryVerticle())
    .compose(ignored -> vertx.deployVerticle(new OrderVerticle()))
    .compose(ignored -> vertx.deployVerticle(new HttpVerticle()));

compose 表示下一步依赖上一步成功。任何部署失败都会终止链路，不会开放一个缺少消费者的入口。

库存请求设置一秒等待边界：

DeliveryOptions options = new DeliveryOptions().setSendTimeout(1_000);
vertx.eventBus().<JsonObject>request(
        InventoryVerticle.RESERVE_ADDRESS, request, options)
    .onSuccess(reservation -> {
      String id = UUID.randomUUID().toString();
      JsonObject order = request.copy()
          .put("id", id)
          .put("status", "CREATED")
          .put("reservation", reservation.body());
      orders.put(id, order);
      message.reply(order);
    })
    .onFailure(error ->
        message.fail(failureCode(error), error.getMessage()));

一秒只是教学值。生产参数应来自请求总预算、下游延迟分位和业务容忍度。没有超时的异步请求并不会更可靠，只会让等待变得不明显。

Future 不提供分布式事务

当前顺序是先扣内存库存，再保存内存订单。如果库存成功后进程退出，会出现库存已扣而订单未保存。Future 只编排异步结果，不提供跨资源原子性。

真实系统需要按约束选择一致性方案：

同一数据库内使用本地事务；
订单落库后通过 Outbox 发布事件；
库存预留设置过期时间，以确认或取消完成 Saga；
请求携带幂等键，避免超时重试重复扣减。

有状态 Verticle 不能盲目扩容

InventoryVerticle 使用普通 Map 保存库存。同一个消费者的 Handler 在关联 Context 上调度，读取与扣减不会被该消费者另一条消息从中间插入。

但部署四个实例会产生四份 Map。Event Bus 会轮询消费者，实例之间却无法看到彼此的扣减，总库存会被错误放大。真实库存需要数据库条件更新、按 SKU 一致性分片或其他明确的一致性策略。

设计有状态 Verticle 时需要回答：状态由谁拥有、哪些消息可以修改它、实例数量变化是否改变语义。当前示例明确存在重启丢失、无法多实例、没有预留过期和无法审计等限制。

运行与验证

1
2
3

curl -i -X POST http://localhost:8080/orders \
  -H 'content-type: application/json' \
  -d '{"customerId":"customer-1","sku":"book","quantity":2}'

正常请求返回 201、订单 ID、CREATED 和剩余库存。请求 4 个初始库存只有 3 个的 keyboard，返回 409。自动化测试同时覆盖创建并查询订单以及库存不足，避免只验证成功路径。

这一版仍没有数据库、幂等键和跨资源补偿，不能被称为生产订单系统；它的价值是让消息协议、失败传播和状态边界都可以实际运行和测试。