配置GroupID后segmentio/kafka-go消费者仍收到重复消息如何解决

• offset提交策略问题：你使用的kafka-go库默认开启自动异步提交offset，若消息处理耗时超过kafka的session.timeout.ms或max.poll.interval.ms阈值，kafka会判定消费者已下线触发分区重平衡，未提交offset的消息会被分配给新的消费者重新消费，导致重复接收。
• StartOffset配置问题：你配置了StartOffset: kafka.FirstOffset，若消费者组为首次启动、或消费进度已过期被kafka清理，所有消费者都会从最早偏移量开始拉取，消费进度同步不及时就会出现重复消费。
• 重平衡过程异常：消费者组触发重平衡时，旧消费者可能还没来得及提交已处理完成的消息offset，分区就被分配给新的消费者，新消费者会从旧offset位置重新拉取消息导致重复。
• 缺少业务侧幂等校验：如果消息处理逻辑执行失败，没有手动调整offset也没有做重复校验，kafka会判定该消息未被消费，后续会再次推送该消息。

• 改为手动同步提交offset：关闭自动提交，确保消息业务逻辑处理完成后再主动提交offset，配置示例如下：

// 初始化消费者时关闭自动提交
consumer := kafka.NewReader(kafka.ReaderConfig{
    Brokers:         kafkaBrokers,
    Topic:           "my-topic",
    StartOffset:     kafka.FirstOffset,
    MinBytes:        1,
    MaxBytes:        100e6,
    GroupID:         "my-group",
    CommitInterval:  0, // 0表示关闭自动提交，需要手动提交offset
    MaxWait:         5 * time.Second, // 原24小时的配置太长，容易触发超时，建议调整为秒级
    ReadLagInterval: 1 * time.Second,
    Dialer:          dialer,
    QueueCapacity:   queueCapacity * 2,
})

// 消费逻辑示例
msg, err := consumer.ReadMessage(ctx)
if err != nil {
    // 处理读取错误
    return
}
// 执行业务处理逻辑
if err = processMessage(msg); err != nil {
    // 处理业务错误，可选择重试或者记录后跳过提交
    return
}
// 业务处理成功后手动提交offset
if err = consumer.CommitMessages(ctx, msg); err != nil {
    // 处理offset提交失败逻辑
}

• 调整超时适配业务耗时：根据业务最大处理耗时，合理设置session.timeout.ms、max.poll.interval.ms、HeartbeatInterval参数，避免消费者因处理超时被误判下线，减少不必要的重平衡。
• 调整StartOffset配置：如果不需要每次都从头消费历史消息，将StartOffset改为kafka.LastOffset，仅新消费者组首次启动时从最新消息开始消费，避免重复拉取历史消息。
• 增加业务侧幂等校验：在消费逻辑侧增加唯一标识校验，比如根据消息key或消息内的唯一业务ID，在数据库中记录已处理的消息标识，重复消息直接跳过处理，这是兜底解决重复消费的最稳妥方案。

内容的提问来源于stack exchange，提问作者Vikram Biwal