多Kafka Producer实例运行时Consumer丢失15-20%日志问题排查

算力阿峰

2026-6-24

问题分析与解决方案

结合你的场景和代码来看，日志丢失大概率是Kafka客户端配置/代码逻辑问题，而非AWS基础设施问题，以下是具体分析和修复建议：

一、核心问题定位

1. Producer异步发送无回调与确认机制

你的Producer使用异步发送但完全未处理发送结果，也未配置关键可靠性参数：

底层依赖的librdkafka默认acks=1（仅leader副本确认），无重试配置时，发送失败的消息会直接丢失
异步发送的错误不会主动抛出，必须通过回调捕获，否则无法感知发送失败

2. Producer资源与配置限制

大量实例并发发送时，可能触发：

默认队列缓存上限过低，导致消息被丢弃
无重试策略，网络波动或集群压力大时的失败消息无法恢复

3. Consumer端的隐性问题

你的Consumer代码存在多个风险点：

未处理ReadMessage的错误，网络中断、Rebalance异常时会陷入空循环，错过消息
默认自动提交offset，若消费逻辑耗时较长，可能出现offset提前提交，重启后跳过未处理消息
未限制单次拉取消息数量，易触发Rebalance导致消息丢失或重复消费

AWS基础设施排查方向（快速排除）

若要确认不是AWS问题，可检查：

Kafka集群（如MSK）监控：UnderReplicatedPartitions、BrokerCPUUtilization、NetworkIn/Out，确认集群无压力瓶颈
EC2实例CloudWatch指标：NetworkPacketsDropIn/Out、CPUUtilization，排查实例资源是否耗尽
VPC流量日志：确认无数据包被安全组/NACL丢弃

二、代码修复建议

Producer代码优化（添加回调与可靠性配置）

func pushLogToKafka(str string, flg bool) {
    var logEntry MongoLogInfoType
    logEntry.Message = str
    logEntry.Timestamp, _ = time.Parse(time.RFC3339, time.Now().Format(time.RFC3339))
    logEntry.Flag = flg
    jsonString, err := json.Marshal(logEntry)
    if err != nil {
        log.Printf("日志序列化失败: %v", err)
        return // 避免panic导致服务崩溃
    }

    topic := config.ApiLogTopic
    deliveryChan := make(chan kafka.Event)
    
    // 发送消息并绑定回调
    err = kafkaInit.Producer.Produce(&kafka.Message{
        TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
        Value:          jsonString,
    }, deliveryChan)

    if err != nil {
        log.Printf("消息入队失败: %v", err)
        return
    }

    // 异步处理发送结果
    go func() {
        e := <-deliveryChan
        switch ev := e.(type) {
        case *kafka.Message:
            if ev.TopicPartition.Error != nil {
                log.Printf("消息投递失败: %v", ev.TopicPartition.Error)
                // 此处可添加自定义重试逻辑
            } else {
                log.Printf("消息投递成功: %v", ev.TopicPartition)
            }
        case kafka.Error:
            log.Printf("Kafka客户端错误: %v", ev)
        }
        close(deliveryChan)
    }()
}

Producer初始化时补充可靠性配置：

producer, err := kafka.NewProducer(&kafka.ConfigMap{
    "bootstrap.servers":        config.BootstrapServers,
    "acks":                     "all",        // 等待所有同步副本确认，确保消息不丢失
    "retries":                  3,           // 失败重试次数
    "retry.backoff.ms":         1000,        // 重试间隔
    "queue.buffering.max.messages": 100000, // 增大队列缓存上限
})

Consumer代码优化（错误处理与手动offset管理）

func Listen() {
    log.Println("开始消费Kafka主题")
    c, err := kafka.NewConsumer(&kafka.ConfigMap{
        "bootstrap.servers":                  config.BootstrapServers,
        "group.id":                           config.GroupId,
        "auto.offset.reset":                  config.AutoOffsetReset,
        "topic.metadata.refresh.interval.ms": config.TopicMetadataRefresh,
        "enable.auto.commit":                 false, // 关闭自动提交，手动控制offset
        "max.poll.records":                   100,   // 限制单次拉取数量，避免处理超时
    })
    if err != nil {
        log.Printf("Kafka消费者初始化失败: %v", err)
        logs := "Kafka consumer failed\n" + err.Error()
        writeLogFile(config.LogsFilePath, []byte(logs))
        return
    }
    defer c.Close() // 确保退出时关闭消费者

    // 订阅主题并处理Rebalance事件
    err = c.SubscribeTopics([]string{config.ApiLogTopic, config.SendSmsTopic}, func(_ *kafka.Consumer, event kafka.Event) error {
        switch ev := event.(type) {
        case kafka.AssignedPartitions:
            log.Printf("分配分区: %v", ev.Partitions)
            c.Assign(ev.Partitions)
        case kafka.RevokedPartitions:
            log.Printf("回收分区: %v", ev.Partitions)
            c.Unassign()
        }
        return nil
    })
    if err != nil {
        log.Printf("订阅主题失败: %v", err)
        return
    }

    for {
        msg, err := c.ReadMessage(-1)
        if err != nil {
            if err.(kafka.Error).IsEOF() {
                log.Println("已到达主题末尾")
                continue
            }
            log.Printf("读取消息错误: %v", err)
            time.Sleep(1 * time.Second)
            continue
        }

        // 解析消息
        var kafkaLog core.LogInfo
        err = json.Unmarshal(msg.Value, &kafkaLog)
        if err != nil {
            log.Printf("日志解析失败: %v, 原始消息: %s", err, string(msg.Value))
        } else {
            // 此处添加写入数据库的逻辑
            // ...
        }

        // 手动提交offset，确保消息处理完成后再提交
        _, err = c.CommitMessage(msg)
        if err != nil {
            log.Printf("提交offset失败: %v", err)
        }
    }
}

三、验证步骤

先部署优化后的Producer，观察错误日志，确认是否有发送失败的消息
检查Kafka集群的MessagesInPerSec和FailedProduceRequestsPerSec指标，验证发送成功率
部署优化后的Consumer，监控消费进度与offset提交情况
若问题仍存在，再排查AWS基础设施指标，排除集群或实例瓶颈

内容的提问来源于stack exchange，提问作者amogh latlong