如何在Spring Boot中防止未消费的重复消息发布到Kafka主题？

核心思路 #

要实现这个需求，核心是跟踪目标Kafka主题中当前未被消费的消息唯一标识。当上游发来重复消息时，先校验该标识是否存在于“未消费消息池”，存在则跳过发送，不存在才执行发布操作。

具体实现方案 #

1. 分布式缓存（Redis）+ 消费确认机制（推荐） #

这是最通用、易实现的方案，适合集群部署场景：

• 步骤：
  • 给每个业务消息定义全局唯一标识（比如业务ID，如你示例中的"ABC"），不要用消息内容哈希（避免内容细微差异导致误判）
  • 中间系统收到上游消息时，先调用Redis的SETNX命令（仅当key不存在时设置），将消息ID作为key，值设为"SENT"，并设置合理的过期时间（比如比消息最大重试周期长30分钟）
  • 如果SETNX返回成功（说明该消息未被发送或已被消费），则发送消息到Kafka；如果返回失败（说明消息已发送且未被消费），直接跳过
  • 业务消费者成功消费消息并手动提交offset后，删除Redis中对应的消息ID key
• Spring Boot代码示例：

  // 发送前校验
  String msgId = "ABC"; // 从上游消息中提取的唯一标识
  Boolean isNew = stringRedisTemplate.opsForValue().setIfAbsent(msgId, "SENT", 30, TimeUnit.MINUTES);
  if (Boolean.TRUE.equals(isNew)) {
      kafkaTemplate.send("target-topic", msgId, messageContent);
  } else {
      // 跳过重复发送，可记录日志
      log.info("Message {} already exists in unconsumed queue, skip sending", msgId);
  }
  
  // 消费成功后清理缓存
  @KafkaListener(topics = "target-topic", groupId = "business-consumer-group")
  public void consume(ConsumerRecord<String, String> record, Acknowledgment ack) {
      try {
          // 执行业务逻辑
          processMessage(record.value());
          // 手动提交offset
          ack.acknowledge();
          // 删除Redis中的消息ID
          stringRedisTemplate.delete(record.key());
      } catch (Exception e) {
          // 消费失败，不清理缓存，等待重试
          log.error("Consume message failed", e);
      }
  }
  

• 注意点：必须使用手动提交offset，确保消费成功后再清理缓存；过期时间要覆盖消息的最大重试时长，避免因消费者重试期间缓存过期导致重复发送。

2. 本地缓存+消费确认（仅单节点部署场景） #

如果你的中间系统是单节点部署，可以用本地缓存（如Guava Cache、Caffeine）替代Redis：

• 发送前检查缓存中是否存在消息ID，不存在则发送并加入缓存
• 消费成功后从缓存中移除ID
• 需给缓存设置过期时间，防止消费者异常挂掉导致缓存永久留存

3. 基于Kafka消费者组的未消费消息跟踪 #

这种方案无需额外缓存，但实现复杂度较高：

• 启动一个独立的消费者组，订阅目标主题，专门用于跟踪未被消费的消息
• 维护一个全局的未消费消息ID集合（用数据库或Redis），当新消息发送到主题时，将ID加入集合
• 业务消费者消费成功并提交offset后，将ID从集合中移除
• 发送新消息前，先查询该集合，存在则跳过
• 缺点：需要处理offset提交的一致性，且当主题消息量大时，跟踪集合的维护成本较高

关键建议 #

• 一定要和上游确认消息的全局唯一业务ID，这是整个方案的基础，不要依赖消息内容做判断
• 优先选择Redis方案，支持集群部署且原子性操作能保证并发场景下的正确性
• 监控未消费消息的数量（Redis的key数量或集合大小），及时排查消费堆积问题
• 处理异常场景：比如生产者发送Kafka失败时，要删除Redis中对应的key，避免后续无法正常发送该消息

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