当前位置：首页 > article >正文

记录一次Kafka重复消费的问题

article 2025/3/24 7:34:32

不讲道理，先抛问题

日志发现Kafka同一个消费者在一段时间内对同一条消息多次消费。
请添加图片描述
在这里插入图片描述

原理不讲，先上配置

# 是否自动提交offset
spring.kafka.consumer.enable-auto-commit=true
# 提交offset延时(接收到消息后多久提交offset)
spring.kafka.consumer.auto.commit.interval.ms=1000
# latest:重置为分区中最新的offset(消费分区中新产生的数据);
spring.kafka.consumer.auto-offset-reset=latest
# Rebalance 超时
spring.kafka.consumer.properties.max.poll.interval.ms=300000  # 5 分钟

三言两语，背景简介

Kafka一个生产者，一个消费者，消费同一个Topic，但是其中的某些消息处理耗时超过5分钟。

捕获问题，深度剖析

Kafka自动提交offset后因默认max.poll.interval.ms设置5分钟没有调用poll()从而发生Reblance重复消费的问题。

解决方案，横向对比

offset自动提交改为手动提交
spring.kafka.consumer.enable-auto-commit=false

@KafkaListener(topics = "generateYyVoucher-topic", groupId = "defaultConsumerGroup")
public void generateYyVoucher(ConsumerRecord<String, String> record, Acknowledgment ack) {
    try {
        ...
        // 提交 Offset
        ack.acknowledge();
    } catch (Exception e) {
        log.error("Message processing failed: ", e);
        // 如果处理失败，Offset 不会被提交
    }
}

2.增加max.poll.interval.ms延迟
spring.kafka.consumer.properties.max.poll.interval.ms=900000 # 设置为15分钟

总结：但是上述两种方案均不能从根本上解决消费者重复消费的问题！根本问题是在于Reblance消费重组原因导致！
第1种只能解决offset偏移量不会重发消费当前消息，但可能会消费上一个消息；
第2种增加延迟，当业务逻辑超过设置时间时仍然会重复消费。

重复消费，最佳解决方案

幂等消费标识（唯一标识）

        // 幂等消费标识（唯一标识），以解决Kafka自动提交offset后因默认max.poll.interval.ms设置5分钟没有调用poll()从而发生Reblance重复消费的问题
        String redisKey = "voucher:processing:" + ledgerId;
        // 检查 Redis 中是否已存在该幂等标识
        if (stringRedisTemplate.hasKey(redisKey)) {
            return;
        }
        // 设置 Redis 中的标识为正在处理中（可以设置一个有效期，比如 30 分钟）
        stringRedisTemplate.opsForValue().set(redisKey, "processing", 30, TimeUnit.MINUTES);