欢迎您访问365答案网,请分享给你的朋友!
生活常识 学习资料

线上问题:SpringBoot消费者自动提交关闭无效

时间:2023-04-20
1 场景复现

关闭消费者自动提交offset,即enable-auto-commit: false,但是,消费者消费数据后,查询group-id的offset还是更新了。

配置profile:

spring: application: name: tutorial # 服务名称:注册中心server devtools: restart: enabled: true kafka: bootstrap-servers: 192.168.211.129:9092 producer: retries: 0 batch-size: 16384 buffer-memory: 33554432 key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer properties: linger.ms: 1 consumer: enable-auto-commit: false auto-commit-interval: 100ms key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer properties: session.timeout.ms: 15000 group-id: tutorial

消费者:

package com.monkey.tutorial.modules.kafka;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.kafka.annotation.KafkaListener;import org.springframework.kafka.support.Acknowledgment;import org.springframework.stereotype.Service;@Servicepublic class Consumer { private static final Logger logger = LoggerFactory.getLogger(Consumer.class); @KafkaListener(topics = {"tutorial"}, groupId = "g1") public void consumer1(ConsumerRecord consumerRecord) { String topic = consumerRecord.topic(); String data = consumerRecord.value(); logger.info(">>>>>>>>>>Kafka consumer1, topic:{}, data:{}", topic, data); }}

2 原因

Kafka包中的消费者的自动提交不能单独使用,enable-auto-commit单独使用是无效的,即单独配置,最终都会自动提交offset。
当只配置消费者enable-auto-commit: false,不会生效,因为此时手动提交也为false,按照表达式:!this.isAnyManualAck && !this.autoCommit最终都会为true,从而自动提交。
方法:org.springframework.kafka.listener.KafkaMessageListenerContainer.ListenerConsumer#invokeBatchOnMessage

3 方案

消费者自动提交offset时,无需配置参数:enable-auto-commit;

spring: application: name: tutorial # 服务名称:注册中心server devtools: restart: enabled: true kafka: bootstrap-servers: 192.168.211.129:9092 producer: retries: 0 batch-size: 16384 buffer-memory: 33554432 key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer properties: linger.ms: 1 consumer: auto-commit-interval: 100ms key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer properties: session.timeout.ms: 15000 group-id: tutorial

消费者手动提交offset时,配置enable-auto-commit: false和ack-mode: manual。

spring: application: name: tutorial # 服务名称:注册中心server devtools: restart: enabled: true kafka: bootstrap-servers: 192.168.211.129:9092 producer: retries: 0 batch-size: 16384 buffer-memory: 33554432 key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer properties: linger.ms: 1 consumer: enable-auto-commit: false key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer properties: session.timeout.ms: 15000 group-id: tutorial listener: missing-topics-fatal: false type: batch ack-mode: manual

4 小结

Kafka包中的消费者的自动提交不能单独使用,enable-auto-commit单独使用是无效的,即单独配置,最终都会自动提交offset。
(1)enable-auto-commit是为手动提交设置的;
(2)Ack手动提交,配置消费者enable-auto-commit: false和手动提交ack-mode: manual,此时,在消费者中手动ack。

Copyright © 2016-2020 www.365daan.com All Rights Reserved. 365答案网 版权所有 备案号:

部分内容来自互联网,版权归原作者所有,如有冒犯请联系我们,我们将在三个工作时内妥善处理。