事件桥接

用事件总线（Kafka/NATS/EventBridge）连接服务，实现解耦可回放通信。

为什么

服务间点对点 HTTP 调用创建脆弱耦合图：服务 A 宕→B 失败→C 超时→D 返回过期数据。事件总线断此链——服务发布事件并订阅所关心的，不知谁生产什么。总线还给你可回放：消费者 bug 修后，重处理最近 7 天事件而非请用户重做。这是'邮件服务宕了所以丢你订单'和'服务恢复时邮件会发'的架构差异。代价是运营总线，但托管选项（EventBridge、Upstash）使其可忽略。

何时使用

有 ≥3 服务须对同一业务事件反应时。约 1-2 天。单体或 2 服务可跳过。

如何做

1. 按规模选总线：NATS（轻量、<1ms）、Kafka（持久、可回放）、EventBridge（AWS 托管）
   - 多数 SaaS：Upstash Kafka 或 AWS EventBridge——无基础设施运营
   - 低延迟：NATS 加 JetStream 持久

2. 在 packages/events 包（monorepo）或 schema registry 定义共享事件 schema
   - 用 zod 或 Avro；每事件用 schema_version 字段版本化
   - 滚动部署期间消费者须处理 N 和 N-1 版本

3. 生产者用主题 + 事件类型 + 关联 ID 发布
   - 主题："orders"；事件类型："OrderCreated"；correlationId 供追踪
   - 发布从生产者角度是发后不管（至少一次投递）

4. 消费者用消费者组 + 幂等 handler 订阅
   - 消费者组：并行消费者分担负载，每事件处理一次
   - 幂等：按事件 ID 去重（同 webhook 幂等）

5. 启用回放：事件保留 7-30 天
   - 消费者 bug 修后，重置消费者 offset 重处理
   - 这是杀手功能——点对点 HTTP 做不到

示例

// 生产者（订单服务）
import { Kafka } from "kafkajs";
const kafka = new Kafka({ brokers: [env.KAFKA_BROKER] });
const producer = kafka.producer();
await producer.connect();
await producer.send({
  topic: "orders",
  messages: [{
    key: order.id,
    value: JSON.stringify({
      type: "OrderCreated", schemaVersion: 1, correlationId: crypto.randomUUID(),
      data: { orderId: order.id, userId: order.userId, total: order.total },
    }),
  }],
});

// 消费者（邮件服务）——幂等、消费者组
const consumer = kafka.consumer({ groupId: "email-service" });
await consumer.subscribe({ topic: "orders", fromBeginning: false });
await consumer.run({
  eachMessage: async ({ message }) => {
    const event = JSON.parse(message.value.toString());
    if (await alreadySent(event.data.orderId)) return;  // 幂等
    await sendOrderConfirmation(event.data);
  },
});

常见问题

⚠请求 handler 里同步发布事件，总线抖动给每用户请求加 200ms。
⚠不版本化事件 schema，生产者部署破所有消费者直到他们也部署。
⚠所有事件用单主题，消费者滤 100 条无关消息找自己的。
⚠因'总线投递一次'跳过幂等——Kafka 是至少一次，重平衡时重复。

最佳实践

✓DB 事务提交后发布事件（outbox 模式）——永不之前，否则为回滚写发布事件。
✓用事务 outbox 模式：同 TX 写事件到 outbox 表，relay 发布到总线。
✓设每消费者延迟告警——消费者落后指示 bug 或扩容需求。
✓事件小（<100KB）；大载荷放对象存储，事件里用 URL 引用。