跨服务器运行相同Kafka消费者组重复消费问题及Scala解决方案咨询

问题根源分析 #

你的现有配置存在几个关键问题：

• StreamsConfig.EXACTLY_ONCE仅适用于Kafka Streams框架，普通消费者无法通过这个配置实现精确一次语义。
• 开启ENABLE_AUTO_COMMIT_CONFIG=true自动提交偏移量，当消费者宕机时，已消费但未提交偏移量的消息会被重新分配给存活的消费者，导致重复消费。
• 生产者端的幂等配置（ENABLE_IDEMPOTENCE_CONFIG等）是解决生产端重复发送的问题，无法直接解决消费端的重复消费。

解决方案步骤 #

1. 关闭自动提交，手动管理偏移量 #

关闭自动提交逻辑，仅在确认消息处理完成后手动提交偏移量，确保只有成功处理的消息才会记录消费位置。

2. 实现消费端幂等性 #

即使偏移量提交出现异常，也要保证同一消息多次消费的业务结果与单次消费一致。可通过消息的唯一标识（如业务ID、消息Key）实现去重。

3. 优化重平衡处理（可选） #

配置重平衡监听器，在分区被收回前提交已处理的偏移量，缩小重复消费的范围。

Scala代码实现示例 #

核心配置调整 #

import org.apache.kafka.clients.consumer.ConsumerConfig
import java.util.Properties
import scala.jdk.CollectionConverters._

val props = new Properties()
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-brokers:9092")
props.put(ConsumerConfig.GROUP_ID_CONFIG, "groupid")
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
// 关闭自动提交，改为手动控制
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
// 设置会话超时，加快重平衡触发速度
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000")
// 偏移量重置策略：无有效偏移量时从最新位置开始（可根据业务调整为earliest）
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")

消费者实现（手动提交+幂等处理） #

import org.apache.kafka.clients.consumer.{KafkaConsumer, TopicPartition, OffsetAndMetadata}
import java.time.Duration

// 模拟持久化去重存储（生产环境建议用Redis/数据库）
val processedMsgIds = scala.collection.mutable.Set[String]()

val consumer = new KafkaConsumer[String, String](props)
consumer.subscribe(List("your-target-topic").asJava)

try {
  while (true) {
    val records = consumer.poll(Duration.ofMillis(100))
    if (!records.isEmpty) {
      // 按分区分组处理，便于精准提交偏移量
      val partitionRecords = records.asScala.groupBy(_.partition())
      
      partitionRecords.foreach { case (partition, msgs) =>
        msgs.foreach { record =>
          // 提取消息唯一标识（此处假设Key为业务唯一ID，可根据实际从Value中解析）
          val msgId = record.key()
          
          if (!processedMsgIds.contains(msgId)) {
            // 执行业务处理逻辑
            println(s"Processing message: ${record.value()}")
            
            // 处理成功后标记为已处理
            processedMsgIds.add(msgId)
          }
        }
        
        // 处理完当前分区所有消息后，同步提交偏移量（提交下一个要消费的位置）
        val nextOffset = msgs.last.offset() + 1
        val offsetMap = Map(
          new TopicPartition("your-target-topic", partition) -> new OffsetAndMetadata(nextOffset)
        ).asJava
        consumer.commitSync(offsetMap)
      }
    }
  }
} catch {
  case e: Exception => e.printStackTrace()
} finally {
  consumer.close()
}

重平衡监听器实现（可选） #

import org.apache.kafka.clients.consumer.ConsumerRebalanceListener

val rebalanceListener = new ConsumerRebalanceListener {
  override def onPartitionsRevoked(partitions: java.util.Collection[TopicPartition]): Unit = {
    // 分区被收回前，提交已处理的偏移量
    consumer.commitSync()
    println(s"Revoked partitions: ${partitions.asScala.mkString(",")}, committed offsets")
  }

  override def onPartitionsAssigned(partitions: java.util.Collection[TopicPartition]): Unit = {
    println(s"Assigned partitions: ${partitions.asScala.mkString(",")}")
  }
}

// 订阅时绑定监听器
consumer.subscribe(List("your-target-topic").asJava, rebalanceListener)

关键说明 #

• 手动提交偏移量：commitSync会阻塞直到提交成功，适合一致性要求高的场景；若追求性能可改用commitAsync，但需处理提交失败的回调逻辑。
• 幂等性是核心：内存存储仅为示例，生产环境必须用持久化存储记录已处理的消息ID，避免重启后丢失去重数据。
• Kafka Streams替代方案：如果业务适合流处理模式，直接使用Kafka Streams并配置StreamsConfig.EXACTLY_ONCE_V2=true，框架会自动处理偏移量和精确一次语义，大幅简化开发。

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