20 - 性能优化与最佳实践

本章总结 Watermill + Kafka 生产环境的性能优化策略和常见反模式。

Kafka Producer 优化

参数	推荐值	说明
batch.size	16384 (16KB)	批量发送大小，增大可提高吞吐
linger.ms	5-10	发送前等待时间，让更多消息聚集到一个 batch
compression.type	snappy 或 lz4	压缩减少网络和磁盘开销
acks	all（默认）	最强持久性保证
max.in.flight.requests.per.connection	5	控制并发度

saramaConfig := kafka.DefaultSaramaSyncPublisherConfig()
saramaConfig.Producer.Compression = sarama.CompressionSnappy
saramaConfig.Producer.Flush.Bytes = 32768
pub, _ := kafka.NewPublisher(kafka.PublisherConfig{
    Brokers:               brokers,
    OverwriteSaramaConfig: saramaConfig,
}, logger)

Handler 优化

避免阻塞操作

Handler 在 HandlerMaxConcurrency 个 goroutine 中执行（默认 1）。阻塞操作（如慢 HTTP 调用）会阻塞整个消费管道。

// 反模式：同步调用外部 API
func handler(msg *message.Message) ([]*message.Message, error) {
    resp, _ := http.Get("http://slow-api/process") // 阻塞 5s
    return nil, nil
}

// 推荐：异步处理 + 快速 ACK
func handler(msg *message.Message) ([]*message.Message, error) {
    go processAsync(msg.Payload)  // 异步处理
    return nil, nil               // 立即 ACK
}

注意：异步 ACK 意味着 offset 已提交，如果异步处理失败无法重试。仅在"尽力而为"场景使用。

批量处理

将多条消息聚合为批次处理，减少数据库 IO 次数：

func batchHandler(msgs []*message.Message, sub message.Subscriber, pub message.Publisher) error {
    ids := make([]string, len(msgs))
    for i, m := range msgs { ids[i] = string(m.Payload) }
    db.Where("id IN ?", ids).Update("status", "processed") // 一次更新
    return nil
}

Watermill 的 PoisonQueue 中间件天然支持批量。

GoChannel vs Kafka 性能对比

维度	GoChannel	Kafka
吞吐	~100K msg/s（单进程）	~1M msg/s（集群）
延迟	< 1ms	1-10ms
持久化	无	磁盘持久化
水平扩展	仅单进程	集群线性扩展
适用场景	单元测试、本地开发	生产环境

GoChannel 在测试中替代 Kafka，运行 go test 无需外部依赖，执行速度远优于集成测试。

常见反模式

1. 在 Handler 中启动后台 Goroutine 但不控制生命周期

// 反模式
func handler(msg *message.Message) ([]*message.Message, error) {
    go func() {
        time.Sleep(10 * time.Minute)  // goroutine 泄漏
        doSomething()
    }()
    return nil, nil
}

如果必须异步，使用 errgroup 或 worker pool 控制并发和生命周期。

2. 过大的消息 Payload

Kafka 默认消息大小限制 1MB。大消息降低吞吐、增加延迟。建议：

• Payload 保持 1KB 以内
• 大对象传递引用（如 S3 URL）而非内容
• 压缩后再发送

3. 在 Router 中注册过多 Handler

单进程注册 20+ 个 Handler 时，建议拆分服务。Router 内部对每个 Handler 启动独立 goroutine，过多 Handler 增加调度开销。

4. 忽略 Metadata

msg.Metadata 是携带横切关注点的最佳载体（trace ID、user ID、tenant ID），不污染业务 Payload。

msg.Metadata.Set("correlation_id", reqID)
msg.Metadata.Set("tenant_id", tenantID)

5. 重试中间件配置不当

• 无指数退避：Multiplier = 1 导致固定间隔重试，瞬时故障可能压垮下游
• 无限重试：需配合 PoisonQueue，否则消息永远卡在重试循环

6. 在 Kafka Subscriber 中设置过小的 Poll 间隔

默认 max.poll.interval.ms = 300000（5 分钟）。如果单条消息处理超过此值，Kafka 触发 rebalance。确保 Handler 执行时间 + 重试总时间 < max.poll.interval.ms。

压测建议

# 使用 kafka-producer-perf-test 压测 Producer
docker exec watermill-kafka kafka-producer-perf-test \
  --topic order.events \
  --num-records 100000 \
  --record-size 100 \
  --throughput 1000 \
  --producer-props bootstrap.servers=localhost:9092

# Go 基准测试
go test -bench=. -benchtime=10s ./...

压测时关注：消息丢失率（应为 0）、P99 延迟、Consumer lag 增长趋势。