kafka重复消费问题--consumer消费能力很低
问题描述 采用kafka读取消息进行处理时,consumer会重复读取afka队列中的数据,使用命令查看kafka消费数据时,发现待消费数据一直没变。问题原因 kafka的consumer消费数据时首先会从broker里读取一批消息数据进行处理,处理完成后再提交offset。而项目中的consumer消费能力比较低,导致取出的一批数据在session.timeout.ms时间内没有处理完成,自动提
采用kafka读取消息进行处理时,consumer会重复读取afka队列中的数据,使用命令查看kafka消费数据时,发现待消费数据一直没变。
问题原因
kafka的consumer消费数据时首先会从broker里读取一批消息数据进行处理,处理完成后再提交offset。而项目中的consumer消费能力比较低,导致取出的一批数据在session.timeout.ms时间内没有处理完成,自动提交offset失败,然后kafka会重新分配partition给消费者,消费者又重新消费之前的一批数据,又出现了消费超时,所以会造成死循环,一直消费相同的数据,无限循环下去。
解决方案
项目中使用的是spring-kafka,所以把kafka消费者的配置enable.auto.commit设为false,禁止kafka自动提交offset,从而使用spring-kafka提供的offset提交策略。spring-kafka中的offset提交策略可以保证一批消息数据没有完成消费的情况下,也能提交offset,从而避免了提交失败而导致永远重复消费的问题。
- <span style="font-size:14px;">先来看看spring-kafka的消费线程逻辑
- if (isRunning() && this.definedPartitions != null) {
- initPartitionsIfNeeded();
- // we start the invoker here as there will be no rebalance calls to
- // trigger it, but only if the container is not set to autocommit
- // otherwise we will process records on a separate thread
- <span style="color:#ff0000;">if (!this.autoCommit) {
- startInvoker();
- }</span>
- }</span>
然后把取到的数据丢给上面提到的阻塞列队,由上面创建的线程去消费,并且如果阻塞队列满了导致取到的数据塞不进去的话,spring-kafka会调用kafka的pause方法,则consumer会停止从kafka里面继续再拿数据。接着spring-kafka还会处理一些异常的情况,比如失败之后是不是需要commit offset这样的逻辑
更多推荐
所有评论(0)