4. 消息队列采集异常排查：消息积压、重复消费、序列化失败问题

消息队列这东西，用好了是数据高速公路，用不好就是堵车现场。我在数据采集系统里跟 Kafka、RocketMQ 打交道好几年了，说实话，踩过的坑比吃过的盐还多。今天咱们就聊聊三个最让人头疼的问题：消息积压、重复消费、序列化失败。

4.1 消息积压：到底是谁在拖后腿？

消息积压，说白了就是生产速度大于消费速度。我遇到过最夸张的一次，Kafka 里积压了上亿条消息，消费者 lag 高达几百万。为什么会这样？

我个人习惯，遇到积压先看消费者 lag 监控。Kafka 自带的 kafka-consumer-groups 命令就能搞定：

# 查看消费者组 lag
kafka-consumer-groups --bootstrap-server localhost:9092 \
  --group my-consumer-group --describe

# 输出示例
GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
my-group        data-topic      0          1000            1500            500
my-group        data-topic      1          1200            1800            600

你看，每个分区的 lag 一目了然。如果某个分区 lag 特别高，那可能就是分区分配不均。

4.2 重复消费：数据怎么多了一份？

重复消费，这是消息队列的「原罪」之一。你想想看，为了保证消息不丢，很多系统都会在消费成功后手动提交 offset。但如果提交失败了，下次重启就会重新消费。

我遇到过最坑的一次：消费者处理完消息后，在提交 offset 之前进程挂了。重启后，同一批消息又被消费了一遍。结果数据库里多了好几万条重复数据。

解决重复消费，常用的方法有：

• 业务幂等：在数据库里加唯一索引，重复插入直接报错
• 去重表：用 Redis 或数据库记录已消费的消息 ID
• 事务消息：RocketMQ 支持，但性能会下降

代码示例，用 Redis 做去重：

// 消费前检查是否已处理
String messageId = record.key();
Boolean isProcessed = redisTemplate.opsForValue()
    .setIfAbsent("msg:" + messageId, "1", 1, TimeUnit.HOURS);

if (Boolean.FALSE.equals(isProcessed)) {
    // 已处理过，跳过
    log.warn("重复消息，跳过: {}", messageId);
    return;
}

// 处理业务逻辑
processMessage(record.value());

// 提交 offset
consumer.commitSync();

嗯，这里要注意：Redis 的过期时间要设置合理。太短可能漏掉重复，太长又浪费内存。我一般设 1 小时，够用了。

4.3 序列化失败：数据格式不对，解析不了

序列化失败，说白了就是生产者和消费者用的协议不一致。我见过最离谱的：生产者用 JSON 序列化，消费者却用 Avro 反序列化，结果一堆乱码。

常见的序列化问题：

问题类型	典型表现	排查方法
字段不匹配	反序列化时字段缺失或类型不对	对比生产者和消费者的 schema
编码问题	中文乱码，特殊字符解析失败	检查字符编码，统一用 UTF-8
版本兼容	升级了序列化库，但消费者没升级	检查依赖版本，做好兼容性测试
数据损坏	消息在传输过程中被截断或篡改	用工具解析原始字节流

我个人习惯，在生产者端加一个「自检」逻辑：

// 生产者发送前，先自己反序列化一次
byte[] data = serializer.serialize(message);
try {
    Object obj = deserializer.deserialize(data);
    // 自检通过，再发送
    producer.send(new ProducerRecord<>(topic, data));
} catch (Exception e) {
    log.error("序列化自检失败，消息丢弃: {}", message);
    // 记录到死信队列
    deadLetterQueue.send(message);
}

4.4 知识体系总览

下面这张图，是我梳理的消息队列异常排查的核心逻辑。你看一眼，心里就有谱了。

你看，这三个问题其实是有内在联系的。消息积压可能导致重复消费（因为超时重试），序列化失败也可能引发积压（消费线程卡住）。所以排查的时候，别孤立地看问题。

嗯，这些坑我都踩过。希望你能少走弯路。