提问说明
1、我在springboot 整合kafka的项目中设置了自动提交偏移量为false后,发现消息消费后,还是会把偏移量给提交了
首先贴出我的配置:
server:
port: 8080
spring:
kafka:
bootstrap-servers: ip1:9092,ip2:9092,ip3: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
consumer:
group-id: test-offset-group
auto-offset-reset: earliest
enable-auto-commit: false
auto-commit-interval: 100
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
这是主启动类:
@SpringBootApplication
public class KafkaApplication {
public static void main(String[] args) {
SpringApplication.run(KafkaApplication.class, args);
}
}
下面贴出我的程序代码:
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
/**
* @author 无风三尺浪
*/
@Component
@Slf4j
public class KafkaReceiver {
@KafkaListener(topics = {"test-auto-create-topic"})
public void listen(ConsumerRecord<?, ?> record) {
Optional<?> kafkaMessage = Optional.ofNullable(record.value());
if (kafkaMessage.isPresent()) {
Object o = kafkaMessage.get();
log.info("---------------------record--1 = " + record);
log.info("---------------------message--1 = " + o);
}
}
我程序里并没有手动提交过偏移量。为啥,偏移量会自动提交?
https://blog.csdn.net/yht874690625/article/details/88311161
首先要明确一点,当SpringBoot整合kafka后,
kafka.consumer.enable.auto.commit=false 意思是让spring自动去提交offset
kafka.consumer.enable.auto.commit=true 意识是让kafka去自动提交offset
解决办法:
在kafkaconsumer的配置类里加入以下即可:yaml文件配置不够吧,程序里面需要手动设置一下
mark, 关注后续结果.
自动提交,拉取后,到了一定时间,就提交了。不管你消费成功与否
可是我这里配置的是:enable-auto-commit: false,那我想不提交偏移量怎么办啊?
额,看错了。确认下配置是否生效了。
生效了。我看后台启动日志打印的也是enable.auto.commit = false
我也遇到了同样的问题,求大神解惑。是否springboot配置未生效,还是什么原因。
程序运行的时候,会把所有配置信息打出来,看看auto.commit是什么
同样的问题,不知楼主解决了没?
你的答案