15 - 消息顺序与去重
在分布式消息系统中，消息顺序性和去重是两个核心挑战。Kafka 提供了分区级别的顺序保证，但需要正确的配置和使用模式。
Kafka 分区与顺序
Kafka 的主题（Topic）被划分为多个分区（Partition），每个分区内部严格有序。关键规则：
同一 Key → 同一分区 → 顺序消费
不同 Key → 可能不同分区 → 无序在电商场景中，order_id 是天然的分区键。同一个订单的所有事件（OrderCreated、InventoryReserved、PaymentCompleted）使用相同的 order_id 作为 Key，保证这些事件在同一个 partition 内顺序处理。
Watermill 的 DefaultMarshaler 使用 msg.UUID 作为 Key（随机值），不保证顺序。要实现按 order_id 分区，需要自定义 Marshaler（参见第 14 章）。
消费者顺序保证
即使 producer 按序发布，consumer 端也可能乱序。原因：
- 多线程消费：Router 的
HandlerMaxConcurrency > 1时，同一 partition 的消息被多个 goroutine 并发处理 - 重试：retry 中间件使失败消息滞后于后续消息
保证顺序的方案：
// 设置并发数为 1，串行处理同一 partition 的消息
router, _ := message.NewRouter(message.RouterConfig{
HandlerMaxConcurrency: 1,
}, logger)或者使用 Kafka 的原生分区消费保证（单个 partition 同一时刻只被一个 consumer 线程处理）。
幂等消费的必要性
Kafka 默认提供 at-least-once 语义——消息可能被多次投递。常见重复场景：
| 场景 | 原因 |
|---|---|
| Consumer crash | 处理完消息但 offset 未提交，重启后重复消费 |
| Rebalance | 短暂失联触发 rebalance，已分配的 partition 重新投递 |
| 网络重试 | Producer 没收到 ack 但消息已写入，重试导致重复 |
因此，所有 Handler 必须是幂等的——同一条消息处理多次产生相同的最终结果。
幂等实现策略
1. 消息 UUID 去重
最简单的方式是利用 Watermill 消息的 UUID 做去重：
CREATE TABLE idempotency_keys (
message_uuid VARCHAR(36) PRIMARY KEY,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);Handler 处理前检查 message_uuid 是否已存在：
func (h *Handler) Handle(msg *message.Message) error {
exists, _ := h.repo.ExistsIdempotencyKey(ctx, msg.UUID)
if exists {
return nil // 已处理过，直接返回
}
// 处理业务逻辑...
h.repo.SaveIdempotencyKey(ctx, msg.UUID)
return nil
}2. 业务键去重
对于状态变更类操作（如订单状态更新），可以通过"版本号"或"状态机"实现幂等：
// 只有状态为 pending 才能改为 confirmed
func (r *orderRepo) Confirm(ctx context.Context, orderID string) error {
result := r.db.Model(&Order{}).
Where("order_id = ? AND status = ?", orderID, "pending").
Update("status", "confirmed")
if result.RowsAffected == 0 {
return nil // 已经是 confirmed 或不存在，幂等返回
}
return result.Error
}3. 数据库唯一约束
对于插入类操作，利用 MySQL 唯一索引：
// order 表有 order_id 唯一索引
// 重复插入自动失败，配合 ON DUPLICATE KEY UPDATE
db.Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "order_id"}},
DoNothing: true,
}).Create(&order)去重存储的清理
去重表会不断增长，需要定期清理。常见策略：
- TTL 清理：只保留最近 N 天的去重键（与 Kafka 消息保留期一致）
- 分区表：按日期分区，每日 DROP 旧分区
- 布隆过滤器：内存去重 + DB 兜底，减少数据库查询
电商项目的去重实践
当前电商项目未显式实现去重（简化版本），但架构已为去重做好准备：
- 订单状态更新通过条件更新（
WHERE status = ?）天然幂等 - 库存扣减通过乐观锁（version 字段）防止重复扣减
- 每条消息携带 UUID，可随时接入去重存储
生产环境中建议在 Retry 中间件内部或 Router 级别加入去重逻辑。