PAYATHON 2026

REST API 与事件驱动编排中的 event joins 如何实现

技术老齐

结论

不要让 API Gateway 在内存中一直等待两个 Kafka 事件，也不要把跨服务的 event join 设计成一次必须同步完成的 HTTP 请求。

更可靠的方案是：

  1. 为每次请求生成全局唯一的 operationId。
  2. 持久化流程状态，分别记录 account 和 billing 两个步骤的完成情况。
  3. 由独立的 workflow/orchestrator 服务完成 join。流程存在明确依赖时，显式编排通常比纯 choreography 更容易理解。
  4. REST API 默认返回 202 Accepted，第三方通过状态查询接口获取结果，也可以接收 webhook 通知。
  5. 如果业务希望尽可能同步返回，可以设置一个很短且有明确上限的等待窗口。等待超时后仍返回 202，后台流程继续执行。

第三方 API 使用异步操作资源很常见。难点并不在 polling，而在持久化状态、超时、重复事件和失败补偿。这些问题不会因为改用 webhook 或同步等待而消失。

推荐的流程

可以把流程调整为：

Client
  │ POST /users
  ▼
Workflow Service
  │ 保存 operation
  │ 发布 CreateAccount
  ▼
Account Service
  │ 创建 account
  │ 发布 AccountCreated
  ▼
Workflow Service
  │ 发布 CreateBillingUser
  ▼
Billing Service
  │ 创建 billing user
  │ 发布 BillingUserCreated
  ▼
Workflow Service
  │ 将 operation 标记为 SUCCEEDED
  ▼
Client 查询状态，或接收 webhook

billing-service 不必同时等待原始的 UserCreationRequested 和 AccountCreated。可以让 AccountCreated 携带 billing 所需的最少信息，也可以由 workflow 服务收到该事件后，再发送内容完整的 CreateBillingUser command。

依赖关系可以直接写成：

AccountCreated -> CreateBillingUser

这比让 billing-service 维护两个无序事件的 join 状态更清楚，失败、重试和超时也更容易处理。

如果多个下游步骤可以并行，workflow 状态可以分别记录：

accountStatus = SUCCEEDED
billingStatus = SUCCEEDED

只有两个步骤都成功，整体操作才进入 SUCCEEDED。

REST API 设计

创建请求可以返回一个操作资源：

POST /users HTTP/1.1
Content-Type: application/json
Idempotency-Key: 86c70db1-7adb-46ab-a15b-59512e45c1de

{
  "name": "Alice",
  "email": "alice@example.com"
}

响应：

HTTP/1.1 202 Accepted
Location: /operations/9bca6ba4-6216-4b70-96cf-141172282a20
Retry-After: 2
Content-Type: application/json

{
  "operationId": "9bca6ba4-6216-4b70-96cf-141172282a20",
  "status": "PENDING"
}

查询接口：

GET /operations/9bca6ba4-6216-4b70-96cf-141172282a20

处理期间可以返回：

{
  "operationId": "9bca6ba4-6216-4b70-96cf-141172282a20",
  "status": "RUNNING",
  "steps": {
    "account": "SUCCEEDED",
    "billing": "RUNNING"
  }
}

完成后返回：

{
  "operationId": "9bca6ba4-6216-4b70-96cf-141172282a20",
  "status": "SUCCEEDED",
  "result": {
    "userId": "usr-123",
    "accountId": "acc-456",
    "billingId": "bill-789"
  }
}

如果第三方不希望 polling，可以允许其在创建请求中提交 webhook 地址。workflow 完成后发送通知，客户端仍然可以通过状态接口核对最终结果。Webhook 需要有签名、重试机制和唯一事件标识，接收方也应根据事件 ID 去重。

持久化 event join

流程状态不能只保存在 CompletableFuture、本地 Map 或 API Gateway 的内存中。实例重启或扩缩容时，状态可能丢失。Kafka 消息如果由另一实例消费，当前正在等待的 HTTP 请求也无法拿到结果。

可以使用下面这样的简化状态模型：

CREATE TABLE user_creation_operation (
    operation_id       VARCHAR(64) PRIMARY KEY,
    idempotency_key    VARCHAR(128) UNIQUE,
    account_status     VARCHAR(32) NOT NULL,
    billing_status     VARCHAR(32) NOT NULL,
    operation_status   VARCHAR(32) NOT NULL,
    account_id         VARCHAR(64),
    billing_id         VARCHAR(64),
    error_code         VARCHAR(64),
    created_at         TIMESTAMP NOT NULL,
    updated_at         TIMESTAMP NOT NULL
);

所有 command 和 event 都应携带同一个关联标识：

public record AccountCreated(
        UUID eventId,
        UUID operationId,
        String accountId,
        Instant occurredAt
) {}

在 Spring Cloud Stream 中，可以用函数式 consumer 更新流程状态：

@Configuration
class WorkflowConsumers {

    @Bean
    Consumer<AccountCreated> accountCreated(
            OperationService operationService,
            StreamBridge streamBridge) {

        return event -> {
            boolean changed = operationService.markAccountSucceeded(
                    event.eventId(),
                    event.operationId(),
                    event.accountId()
            );

            if (changed) {
                CreateBillingUser command =
                        operationService.createBillingCommand(event.operationId());

                streamBridge.send("createBillingUser-out-0", command);
            }
        };
    }

    @Bean
    Consumer<BillingUserCreated> billingUserCreated(
            OperationService operationService) {

        return event -> operationService.markBillingSucceeded(
                event.eventId(),
                event.operationId(),
                event.billingId()
        );
    }
}

markAccountSucceeded 和 markBillingSucceeded 应在数据库事务中完成这些工作：

  • 根据 eventId 判断事件是否已经处理。
  • 更新相应步骤的状态和结果。
  • 检查所有必要步骤是否完成。
  • 满足 join 条件后，把整体状态更新为 SUCCEEDED。
  • 已完成的操作再次收到相同事件时，不产生副作用。

示意逻辑如下：

@Transactional
public void markBillingSucceeded(
        UUID eventId,
        UUID operationId,
        String billingId) {

    if (processedEventRepository.existsById(eventId)) {
        return;
    }

    Operation operation =
            operationRepository.findByIdForUpdate(operationId)
                    .orElseThrow();

    operation.completeBilling(billingId);

    if (operation.accountSucceeded()
            && operation.billingSucceeded()) {
        operation.markSucceeded();
    }

    processedEventRepository.save(new ProcessedEvent(eventId));
    operationRepository.save(operation);
}

实际项目还要通过数据库行锁、乐观锁或条件更新，防止两个事件并发到达时出现丢失更新。

可靠发布消息

下面的写法存在一致性窗口：

operationRepository.save(operation);
streamBridge.send("commands-out-0", command);

如果数据库已经提交，但进程在发送消息前崩溃，流程会永久停住。反过来，如果消息已经发出，数据库事务却回滚了，同样会造成状态不一致。

通常可以使用 Transactional Outbox：

  1. 在同一个数据库事务中更新 operation，并写入 outbox 表。
  2. 由独立发布器读取 outbox，再把消息发送到 Kafka。
  3. 发送成功后标记对应的 outbox 记录。
  4. 消费方通过 inbox 或 processed-events 表，根据 eventId 去重。

Kafka 的 exactly-once processing 很适合 Kafka 内部的 consume-process-produce，但它不能自动把普通数据库事务与 Kafka 写入合并成一个原子事务。只要流程涉及业务数据库，就需要单独处理这个一致性边界。

必须同步返回时

如果通常只需几百毫秒就能处理完成，而且第三方明确要求同步响应，可以采用有限等待：

@PostMapping("/users")
public DeferredResult<ResponseEntity<?>> create(
        @RequestBody CreateUserRequest request) {

    UUID operationId = workflowService.start(request);
    DeferredResult<ResponseEntity<?>> result =
            new DeferredResult<>(3000L);

    completionRegistry.await(operationId)
            .whenComplete((operation, error) -> {
                if (error != null) {
                    result.setErrorResult(error);
                } else {
                    result.setResult(ResponseEntity.ok(operation.result()));
                }
            });

    result.onTimeout(() -> result.setResult(
            ResponseEntity.accepted()
                    .location(URI.create("/operations/" + operationId))
                    .body(new OperationAccepted(operationId, "RUNNING"))
    ));

    return result;
}

这里的等待注册表只能用来加快响应，不能作为流程状态的唯一存储。如果 Gateway 在等待期间重启，客户端可能收到连接错误，但后台 operation 仍会继续。客户端可以使用相同的 Idempotency-Key 重试，并取得原操作的结果。

使用这种模式时，还需要设置：

  • 明确且较短的 HTTP 超时。
  • Gateway、负载均衡器和反向代理之间一致的超时配置。
  • 最大并发等待数，防止慢请求耗尽连接和内存。
  • 超时后不撤销已经进入 Kafka 的业务流程。
  • 客户端重试时的幂等处理。

Kafka Streams 是否适合 join

如果两个事件确实相互独立、顺序不确定，而且必须按 key 聚合，可以使用 Kafka Streams。两个事件需要采用相同的 operationId 作为 Kafka record key，再通过 state store 保存中间结果。

但普通的 stream-stream join 通常需要设置时间窗口。事件超过窗口、严重乱序或发生重放时，结果可能无法满足长事务工作流的要求。如果业务流程可能持续数秒、数分钟甚至更久，还需要记录失败状态、支持人工介入、执行补偿并提供查询能力，那么持久化 workflow 状态通常更合适。

Kafka Streams 适合根据两个事件流计算结果，workflow/orchestrator 则适合管理具有生命周期的业务操作。

关键注意事项

  • 不要把业务编排状态放在 API Gateway 中。Gateway 应尽量只处理认证、限流、路由和协议转换。
  • command 表示要求执行某项操作，event 表示已经发生的事实，两者应使用不同的消息类型和 topic。
  • 所有消息都应携带 operationId 和唯一的 eventId，必要时再加入 causationId 与 correlationId。
  • Kafka 通常提供至少一次效果，因此消费者必须具备幂等性，不能假设消息只会到达一次。
  • 为每个步骤定义超时、失败和补偿策略。例如 account 已经成功，但 billing 最终失败时，需要确定是保留 account、禁用 account，还是发送删除 account 的补偿 command。
  • 状态接口应返回持久化的流程状态，不要临时查询多个下游服务后再拼装结果。
  • 使用 Idempotency-Key，避免客户端因 HTTP 超时或断线而重复创建用户。
  • 无法自动恢复的操作应进入 FAILED 或 MANUAL_REVIEW，不要无限重试。
  • operation 和事件处理记录需要保留多久，应根据审计、重试和数据保留要求确定。

对于这个场景，可靠性的基础是把”等待两个步骤完成”建模为一个可持久化、可恢复、可查询的业务流程，而不是设法让 HTTP 请求一直等到两个事件到齐。同步响应可以用来改善性能和体验，但不能承担正确性保证。

备注：内容仅供参考。