这是一个创建于 1542 天前的主题，其中的信息可能已经有所发展或是发生改变。
最近要做个东西，需要从 kafka 消费大量消息，服务器上 kafka 是 0.9 版本的，应用是 springboot-1.5.6 spring-kafka 1.0.0 版本
下面是两个消费者：
@KafkaListener(topicPartitiOns={@TopicPartition(topic = "spyfool",partiiOns= {"0"})},group = "home")
public void doConsume(ConsumerRecord a){
log.info("000000 收到 kafka 消息:{},偏移量={}",a.value(),a.offset());
}
@KafkaListener(topicPartitiOns={@TopicPartition(topic = "spyfool",partitiOns= {"0"})},group = "home")
public void doConsume1(ConsumerRecord a){
log.info("111111 收到 kafka 消息:{},偏移量={}",a.value(),a.offset());
}
配置文件：
spring.kafka.consumer.bootstrap-servers=127.0.0.1:9092
spring.kafka.consumer.group-id=home
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.max-poll-records=10
spring.kafka.consumer.auto-commit-interval=100
spring.kafka.consumer.fetch-max-wait=10000
spring.kafka.listener.ack-mode=manual_immediate
但结果明明同组的两个消费者发生了重复消费：
111111 收到 kafka 消息:say my name!3d0e7171-cae1-44c9-8a9a-f21289160108,偏移量=42141
000000 收到 kafka 消息:say my name!3d0e7171-cae1-44c9-8a9a-f21289160108,偏移量=42141
111111 收到 kafka 消息:say my name!90851ac4-1e5f-4ac8-93c9-72acd64ce529,偏移量=42142
000000 收到 kafka 消息:say my name!90851ac4-1e5f-4ac8-93c9-72acd64ce529,偏移量=42142
111111 收到 kafka 消息:say my name!4d0fa93d-c7e7-4f44-beba-bfa6c25aee12,偏移量=42143
111111 收到 kafka 消息:say my name!d1707959-0be2-4ec1-a81d-86ef211cb73d,偏移量=42144
000000 收到 kafka 消息:say my name!4d0fa93d-c7e7-4f44-beba-bfa6c25aee12,偏移量=42143
111111 收到 kafka 消息:say my name!79cfea0d-548d-437d-bd05-8b65a27f9e9a,偏移量=42145
000000 收到 kafka 消息:say my name!d1707959-0be2-4ec1-a81d-86ef211cb73d,偏移量=42144
000000 收到 kafka 消息:say my name!79cfea0d-548d-437d-bd05-8b65a27f9e9a,偏移量=42145
百思不得其解，请问大家能指点一下吗
谢谢了！
下面是两个消费者：
@KafkaListener(topicPartitiOns={@TopicPartition(topic = "spyfool",partiiOns= {"0"})},group = "home")
public void doConsume(ConsumerRecord a){
log.info("000000 收到 kafka 消息:{},偏移量={}",a.value(),a.offset());
}
@KafkaListener(topicPartitiOns={@TopicPartition(topic = "spyfool",partitiOns= {"0"})},group = "home")
public void doConsume1(ConsumerRecord a){
log.info("111111 收到 kafka 消息:{},偏移量={}",a.value(),a.offset());
}
配置文件：
spring.kafka.consumer.bootstrap-servers=127.0.0.1:9092
spring.kafka.consumer.group-id=home
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.max-poll-records=10
spring.kafka.consumer.auto-commit-interval=100
spring.kafka.consumer.fetch-max-wait=10000
spring.kafka.listener.ack-mode=manual_immediate
但结果明明同组的两个消费者发生了重复消费：
111111 收到 kafka 消息:say my name!3d0e7171-cae1-44c9-8a9a-f21289160108,偏移量=42141
000000 收到 kafka 消息:say my name!3d0e7171-cae1-44c9-8a9a-f21289160108,偏移量=42141
111111 收到 kafka 消息:say my name!90851ac4-1e5f-4ac8-93c9-72acd64ce529,偏移量=42142
000000 收到 kafka 消息:say my name!90851ac4-1e5f-4ac8-93c9-72acd64ce529,偏移量=42142
111111 收到 kafka 消息:say my name!4d0fa93d-c7e7-4f44-beba-bfa6c25aee12,偏移量=42143
111111 收到 kafka 消息:say my name!d1707959-0be2-4ec1-a81d-86ef211cb73d,偏移量=42144
000000 收到 kafka 消息:say my name!4d0fa93d-c7e7-4f44-beba-bfa6c25aee12,偏移量=42143
111111 收到 kafka 消息:say my name!79cfea0d-548d-437d-bd05-8b65a27f9e9a,偏移量=42145
000000 收到 kafka 消息:say my name!d1707959-0be2-4ec1-a81d-86ef211cb73d,偏移量=42144
000000 收到 kafka 消息:say my name!79cfea0d-548d-437d-bd05-8b65a27f9e9a,偏移量=42145
百思不得其解，请问大家能指点一下吗
谢谢了！
 | 1 wellsc 2021-08-01 20:10:20 +08:00 重复消费解决方案网上一搜一大把。。 |
 | 2 jorneyr 2021-08-02 08:20:23 +08:00 Kafka 没注意过，最近看 RocketMQ，提示一个 JVM 中同一个 Group 里只应该有一个 Consumer 实例，且 Consumer 的个数最多等于 partition 的个数。 你在同一个 JVM 中同一个 Partition，同一个 Group 里有 2 个 Consumer，可以考虑下是否设计问题。修改为每个 JVM 只有一个 Consumer，然后 2 个这个程序再看看是否有重复消费问题。 |
 | 3 securityCoding 2021-08-02 09:50:34 +08:00 via Android 消息体给一个唯一 keyq 业务中实现一下幂等 |
 | 4 ericson 2021-08-02 11:28:59 +08:00 可能是因为没有及时 commit 如果想实现 exactly once 的消费策略，需要参考 @securityCoding 提到的在业务层面实现幂等消费(或者去重) 关于 commit 可以看看 [这篇文章]( https://quarkus.io/blog/kafka-commit-strategies/) |
 | 5 huang119412 2021-08-02 11:31:06 +08:00 注意两个参数，auto.offset.reset，enable.auto.commit 。即使一般不会重复消费。重复消费问题，也需要应用层解决。 |
 | 6 BQsummer 2021-08-02 13:56:16 +08:00 1. 没 commit ？ 2. 两个消费者都绑同一个分区 0 上了？ |
Select Language