23 - 常见问题与排查

本章汇总 Watermill 使用中的常见问题、调试方法和故障排除技巧。

常见问题 (FAQ)

Q1: 消息发出去但消费不到？

排查步骤：

1. 确认 topic 名称一致（Publisher 和 Subscriber 的 topic 字符串完全匹配，注意大小写）
2. GoChannel：先 Subscribe 再 Publish，否则消息丢失（无订阅者时丢弃）
3. Kafka：检查 consumer group offset 是否已提交（可能是重复运行脚本）
4. 检查 Router 是否已启动：<-router.Running() 阻塞直到就绪

Q2: Handler 返回 error 但没重试？

原因：未添加 Retry 中间件。

// 必须显式添加
router.AddMiddleware(middleware.Retry{...}.Middleware)

没有 Retry 中间件时，error 只触发 NACK（Kafka 不提交 offset），消息在 Router 内部不会被自动重试。

Q3: GoChannel 消息顺序不对？

GoChannel 不保证消息顺序。当 OutputChannelBuffer > 0 且有多个 Publish goroutine 时，channel 内消息顺序不确定。设置 OutputChannelBuffer = 0 可强制同步投递（但阻塞 Publish）。

Q4: Kafka consumer lag 持续增长？

原因：消费速度跟不上生产速度。

解决方案：

• 增加 Handler 并发数：HandlerMaxConcurrency > 1
• 增加消费者实例数（同 group）：Kafka 自动 rebalance，partition 分配给新实例
• 优化 Handler 处理逻辑：减少 DB IO、异步化非关键操作
• 增加 partition 数：需要重新创建 topic

Q5: 消息重复消费？

Kafka at-least-once 语义必然导致重复。解决方案参见第 15 章幂等消费部分。关键：将幂等逻辑构建到 Handler 业务逻辑中，而非依赖消息系统保证唯一性。

调试技巧

日志级别

开发环境设置详细日志：

logger := watermill.NewStdLogger(true, true)
// 第一个 bool = debug（详细日志）
// 第二个 bool = trace（极详细日志，包括消息内容）

在 ecommerce/cmd/*/main.go 中将 logger 设置为 watermill.NewStdLogger(false, false) 以减少生产日志量，开发时改为 true, true。

消息追踪

通过 Metadata 注入 correlation ID：

msg.Metadata.Set("correlation_id", uuid.New().String())

然后在日志中输出：

log.Info("处理消息", zap.String("correlation_id", msg.Metadata.Get("correlation_id")))

Kafka 命令行工具

# 实时消费 topic（调试用）
docker exec watermill-kafka kafka-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic order.events --from-beginning

# 查看消费者组 lag
docker exec watermill-kafka kafka-consumer-groups \
  --bootstrap-server localhost:9092 \
  --group order-service --describe

# 重置 offset 到最早（重新消费所有消息）
docker exec watermill-kafka kafka-consumer-groups \
  --bootstrap-server localhost:9092 \
  --group order-service --topic order.events \
  --reset-offsets --to-earliest --execute

常见错误场景

场景 1：Kafka 连接超时

kafka: client has run out of available brokers

• 检查 docker-compose ps 确认 Kafka 已启动
• 确认端口 9092 未被防火墙阻挡
• 检查配置文件中 broker 地址是否正确（Docker 内用 kafka:9092，宿主机用 localhost:9092）

场景 2：消息序列化错误

proto: cannot parse invalid wire-format data

• 确认 Publisher 使用 proto.Marshal() 而非 json.Marshal()
• 确认 Subscriber 使用 proto.Unmarshal() 而非 json.Unmarshal()
• 检查 Protobuf 定义是否一致（Publisher 和 Subscriber 使用同一版本 .proto 文件）

场景 3：数据库死锁（库存扣减）

Error 1213: Deadlock found when trying to get lock

• 乐观锁冲突是正常的并发现象
• Retry 中间件会自动重试（第 10、16 章）
• 若持续出现，检查是否有跨表的不一致锁定顺序

场景 4：Router 启动后立即退出

Router closed with error: context canceled

• 检查 ctx 是否过早 cancel
• 确认 router.Run(ctx) 在 goroutine 中运行
• 确认 <-router.Running() 用于等待就绪

社区资源

• Watermill GitHub: https://github.com/ThreeDotsLabs/watermill
• 官方示例: https://github.com/ThreeDotsLabs/watermill/tree/master/_examples
• Kafka 适配器: https://github.com/ThreeDotsLabs/watermill-kafka
• Go Kafka 客户端 (Sarama): https://github.com/IBM/sarama