异步通信与事件驱动

为什么需要异步通信

上一篇讲了 gRPC 同步调用--A 发请求,等 B 响应.这个模型有三个隐含假设:

1. B 必须在线(时间耦合)
2. B 必须在 deadline 内响应(性能耦合)
3. A 必须知道 B 的地址(空间耦合)

当这些假设不成立时,同步调用就是错误的工具:

消息队列基础模型

核心概念

两种基本模式

┌──────────────┐
│ 点对点(Queue)   │
├──────────────┤
│  • Producer   │
│  • Consumer A │
│  • Consumer B │
└──────────────┘
``` ```text
┌────────────────┐
│ 发布/订阅(Topic)   │
├────────────────┤
│  • Producer     │
│  • 订阅者 A(通知服务)  │
│  • 订阅者 B(统计服务)  │
│  • 订阅者 C(积分服务)  │
└────────────────┘

Kafka 深入

Apache Kafka 是微服务领域最主流的消息系统--高吞吐,持久化,支持回溯消费.

架构概览

┌───────────────────────┐
│ Topic: order-events   │
├───────────────────────┤
│  • Partition 0         │
│  • Partition 1         │
│  • Partition 2         │
└───────────────────────┘
┌────────────────────────────────┐
│ Consumer Group: notify-group   │
├────────────────────────────────┤
│  • Consumer 0                   │
│  • Consumer 1                   │
│  • Consumer 2                   │
└────────────────────────────────┘

核心设计决策:

• Partition:Topic 被分割成多个分区,每个分区是一个有序的,不可变的消息序列
• 消息顺序:只保证分区内有序,不保证跨分区有序
• Consumer Group:同一 Group 内的消费者瓜分 Partition,实现负载均衡;不同 Group 各自独立消费全量数据
• 持久化:消息写入磁盘,支持按 offset 回溯重新消费

投递保证

Go Producer / Consumer 示例

// Producer(使用 confluent-kafka-go)
producer, _ := kafka.NewProducer(&kafka.ConfigMap{
    "bootstrap.servers": "localhost:9092",
    "acks":              "all",  // 等待所有副本确认
})

event := OrderCreatedEvent{OrderID: "ord-123", UserID: "u-1"}
payload, _ := json.Marshal(event)

producer.Produce(&kafka.Message{
    TopicPartition: kafka.TopicPartition{Topic: &topic},
    Key:            []byte(event.OrderID), // 保证同一订单的事件有序
    Value:          payload,
    Headers: []kafka.Header{
        {Key: "event-type", Value: []byte("order.created")},
        {Key: "trace-id", Value: []byte(traceID)},
    },
}, nil)

// Consumer
consumer, _ := kafka.NewConsumer(&kafka.ConfigMap{
    "bootstrap.servers": "localhost:9092",
    "group.id":          "notification-service",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": false, // 手动 commit,保证 at-least-once
})
consumer.Subscribe("order-events", nil)

for {
    msg, err := consumer.ReadMessage(time.Second)
    if err != nil { continue }

    // 处理消息
    if err := handleMessage(msg); err != nil {
        // 处理失败:记录日志,后续重试或发送到死信队列
        sendToDeadLetterQueue(msg, err)
    }

    // 处理成功后手动提交 offset
    consumer.CommitMessage(msg)
}

RabbitMQ 对比

RabbitMQ 和 Kafka 面向不同的设计目标:

事件驱动架构(EDA)

事件的三种类型

事件 Schema 设计

一个生产级事件的标准结构:

{
  "event_id": "evt-a1b2c3",
  "event_type": "order.created",
  "source": "order-service",
  "timestamp": "2024-03-15T10:30:00Z",
  "trace_id": "trace-xyz",
  "version": "1.0",
  "data": {
    "order_id": "ord-123",
    "user_id": "u-456",
    "items": [...],
    "total_cents": 29900
  }
}

设计原则:

• event_id:全局唯一,用于幂等去重
• event_type:使用 领域.动作 命名(过去时),如 order.created,payment.completed
• source:产生事件的服务名
• version:schema 版本,支持消费者区分处理
• trace_id:串联分布式追踪链路

幂等消费

At Least Once 意味着同一条消息可能被投递多次(网络抖动,消费者重启,rebalance).消费者必须做到"处理 N 次和处理 1 次效果相同"--这就是幂等性.

幂等实现策略

幂等表实现示例

func handleMessage(ctx context.Context, msg *Event) error {
    // 1. 检查是否已处理(幂等表)
    exists, err := db.ExecContext(ctx,
        "INSERT INTO processed_events (event_id, processed_at) VALUES ($1, NOW()) ON CONFLICT DO NOTHING",
        msg.EventID,
    )
    if err != nil { return err }
    if exists.RowsAffected() == 0 {
        // 已处理过,跳过
        log.Info("duplicate event, skipping", "event_id", msg.EventID)
        return nil
    }

    // 2. 执行业务逻辑(在同一个事务中!)
    return processOrderCreated(ctx, msg.Data)
}

发件箱模式(Transactional Outbox)

一个经典问题:服务需要"写数据库 + 发消息"两个操作.它们不在同一个事务里--如何保证一致性?

双写问题

场景:订单服务创建订单后发送事件

方案 A:先写 DB 再发消息
  1. INSERT INTO orders ... ✓
  2. kafka.Produce("order.created") ✗  ← 发送失败!
  结果:订单已创建,但事件丢失,下游永远不知道

方案 B:先发消息再写 DB
  1. kafka.Produce("order.created") ✓
  2. INSERT INTO orders ... ✗  ← 写入失败!
  结果:事件已发出,但订单不存在,下游处理一个幽灵事件

无论先做哪个,都可能出现不一致.这就是双写问题(Dual Write Problem).

发件箱方案

核心思想:不直接发消息,而是把待发送的消息写入本地数据库的 outbox 表(与业务操作在同一个事务中).一个独立的进程轮询 outbox 表,将消息投递到消息队列.

sequenceDiagram
  participant APP as 订单服务
  participant DB as 数据库
  participant RELAY as Outbox Relay
  participant MQ as Kafka

  APP->>DB: BEGIN TRANSACTION
  APP->>DB: INSERT INTO orders (...)
  APP->>DB: INSERT INTO outbox (event_type, payload)
  APP->>DB: COMMIT
  Note over DB: 数据库事务保证原子性

  loop 轮询 / CDC
    RELAY->>DB: SELECT * FROM outbox WHERE status='pending'
    RELAY->>MQ: Produce message
    RELAY->>DB: UPDATE outbox SET status='sent'
  end

-- outbox 表结构
CREATE TABLE outbox (
    id          BIGSERIAL PRIMARY KEY,
    event_id    UUID NOT NULL UNIQUE,
    event_type  VARCHAR(100) NOT NULL,
    payload     JSONB NOT NULL,
    status      VARCHAR(20) DEFAULT 'pending',
    created_at  TIMESTAMP DEFAULT NOW(),
    sent_at     TIMESTAMP
);

-- 业务代码:在同一个事务中写业务表 + outbox
BEGIN;
INSERT INTO orders (id, user_id, total) VALUES ('ord-123', 'u-1', 299);
INSERT INTO outbox (event_id, event_type, payload)
VALUES ('evt-abc', 'order.created', '{"order_id":"ord-123","user_id":"u-1","total":299}');
COMMIT;

投递语义:at-least-once

发件箱模式保证消息不丢,但代价是可能重:Relay "已经投递到 MQ,但还没来得及把 outbox 标记为 sent" 时崩溃,重启后会重发同一条消息.因此它的投递语义是 at-least-once(至少一次),而非 exactly-once.

CDC 替代方案

轮询 outbox 表有延迟和性能问题.更优方案是使用 CDC(Change Data Capture)--监听数据库的 binlog / WAL,实时将变更发送到 Kafka:

• Debezium:最流行的 CDC 工具,支持 MySQL / PostgreSQL / MongoDB
• 直接监听 outbox 表的 INSERT 事件,零轮询,低延迟
• 也可以监听业务表本身的变更--连 outbox 表都不用了

事件溯源(Event Sourcing)

传统方式存储实体的"当前状态"(一行记录).事件溯源存储实体经历的"所有事件序列"--当前状态通过回放事件推导得出.

核心概念

传统 CRUD:
  orders 表:{id: "ord-1", status: "shipped", total: 299, ...}
  → 只看到最终状态,中间发生了什么?不知道

事件溯源:
  Event 1: OrderCreated {order_id: "ord-1", items: [...], total: 299}
  Event 2: PaymentReceived {order_id: "ord-1", amount: 299, method: "wechat"}
  Event 3: OrderShipped {order_id: "ord-1", tracking: "SF123456"}
  → 当前状态 = 回放所有事件得出
  → 完整审计轨迹,可以回到任意历史时间点

适用场景与代价

死信队列(Dead Letter Queue)

消费者处理消息失败后怎么办?无限重试会卡住整个队列(Head-of-Line Blocking).标准做法:重试 N 次后转入死信队列(DLQ).

[RETRY] → [主队列] → [Consumer]
[DLQ] → [告警 + 人工处理]

DLQ 消息的处理:

• 人工排查原因后修复 bug,再从 DLQ 重新投递
• 配置自动重试策略(如 1 小时后再试一次)
• 持续失败的消息标记为永久失败,记录到监控报表

消息顺序性

并非所有场景都需要严格有序.分清楚:

背压与流控

当消费者处理速度跟不上生产者时:

• Kafka 天然支持背压:消费者拉模型(pull),处理完一批再拉下一批.消息堆积在 Broker 的磁盘上(Kafka 的持久化日志设计让堆积是安全的)
• RabbitMQ:推模型需要手动设置 prefetch count 限制同时处理的消息数

生产环境的应对:

1. 监控消费延迟(Consumer Lag):当前 offset 与最新 offset 的差值,是最重要的 Kafka 监控指标
2. 水平扩容消费者:增加 Consumer Group 中的实例数(不超过 Partition 数)
3. 限流上游:如果堆积持续增长,考虑对 Producer 限流

事件驱动与 Saga 模式(预告)

异步消息天然适合编排跨服务的业务流程.例如"创建订单"涉及库存扣减,支付,物流--每一步都是一个事件触发下一步:

订单服务: OrderCreated →
  库存服务: InventoryReserved →
    支付服务: PaymentCompleted →
      物流服务: ShipmentCreated

任何一步失败 → 发送补偿事件 → 回滚前序操作

这就是 Saga 模式的核心思想--第 07 篇会深入讲解.

生产最佳实践

• 消息不要太大:单条消息建议 < 1MB.大文件放对象存储,消息里只带 URL
• Schema 注册表:用 Confluent Schema Registry 管理 Avro / Protobuf schema,防止生产者随意改格式
• 消费者要快:消息处理逻辑应尽量简单,耗时操作异步化(消费者内部再用 goroutine pool)
• 监控三大指标:Consumer Lag(消费延迟),消息处理延时 P99,死信队列积压量
• 消息要带 trace_id:否则跨服务的链路追踪在异步环节断裂
• 版本兼容:事件 schema 只做加法,新增字段不影响旧消费者(和 Protobuf 同理)