app开发定制通过字面意思我们不难app开发定制理解这是的自动提交功能。
app开发定制配置消费者(配置ENABLE_AUTO_COMMIT_CONFIG为 true 配置自动提交)
enable.auto.commit 的默认值是 true;就是默认采用自动提交的机制。
auto.commit.interval.ms 的默认值是 5000,单位是毫秒。
此时我们配置消息消费后自动提交offset 位置
- @Bean
- public KafkaConsumer<String, String> kafkaConsumer() {
- Map<String, Object> config = new HashMap<>();
- config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
- config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
- config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
- config.put(ConsumerConfig.GROUP_ID_CONFIG, "groupId");
- config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
- KafkaConsumer<String, String> consumer = new KafkaConsumer<>(config);
- return consumer;
- }
配置消息监听
- @Slf4j
- @Component
- public class PackageMessageConsumer {
-
-
- @Autowired
- KafkaConsumer<String, String> kafkaConsumer;
- @Autowired
- EventProcessMaster eventProcessMaster;
-
- @PostConstruct
- public void listenRealTimeGroup() {
- new Thread(() -> consumer()).start();
- }
-
-
- private void consumer() {
- List<String> topics = new ArrayList<>();
- topics.add("test-kafka-auto.commit");
- kafkaConsumer.subscribe(topics);
- while(true) {
- try {
- // 拉取消息时间间隔 ms
- ConsumerRecords<String, String> records = kafkaConsumer.poll(10);
- for (ConsumerRecord<String, String> record : records) {
- String key = record.key();
- Object content = record.value();
- eventProcessMaster.processRequest(new Event(record.topic(), key, content));
- }
- } catch (Exception e){
- log.error(e.getMessage());
- }
- }
- }
- }
配置自动提交offset 位置之后,我们不必关心消息消费到了什么位置,当程序重启后,消息也不会重复消费;
配置消费者(配置ENABLE_AUTO_COMMIT_CONFIG为 false 配置手动提交)
手动提交顾名思义就是每次我们消费后,kafka不会手动更新offset 位置,同时 auto.commit.interval.ms 也就不被再考虑了。
- @Bean
- public KafkaConsumer<String, String> kafkaConsumer() {
- Map<String, Object> config = new HashMap<>();
- config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
- config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
- config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
- config.put(ConsumerConfig.GROUP_ID_CONFIG, "groupId");
- // 与自动提交的区别
- config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
- KafkaConsumer<String, String> consumer = new KafkaConsumer<>(config);
- return consumer;
- }
当我们设置成手动提交后,不修改其他代码会发现每次重启程序,kafka都会将我们没清理的所有数据都重新消费一遍,与我们需求的幂等性不符,将代码进行完善
- @Bean
- public KafkaConsumer<String, String> kafkaConsumer() {
- Map<String, Object> config = new HashMap<>();
- config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
- config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
- config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
- config.put(ConsumerConfig.GROUP_ID_CONFIG, "groupId");
- // 与自动提交的区别
- config.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");// 自动提交时间间隔
- config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
- KafkaConsumer<String, String> consumer = new KafkaConsumer<>(config);
- return consumer;
- }
- @Slf4j
- @Component
- public class DependPackageMessageConsumer {
-
-
- @Autowired
- KafkaConsumer<String, String> kafkaConsumer;
- @Autowired
- EventProcessMaster eventProcessMaster;
-
-
- @PostConstruct
- public void listenRealTimeGroup() {
- new Thread(() -> consumer()).start();
- }
-
-
- private void consumer() {
- List<String> topics = new ArrayList<>();
- topics.add("test-kafka-auto.commit");
- kafkaConsumer.subscribe(topics);
- while(true) {
- try {
- ConsumerRecords<String, String> records = kafkaConsumer.poll(10);
- for (ConsumerRecord<String, String> record : records) {
- String key = record.key();
- Object content = record.value();
- eventProcessMaster.processRequest(new Event(record.topic(), key, content));
- }
- // 手动提交 offset 位置
- kafkaConsumer.commitSync();
- } catch (Exception e){
- log.error(e.getMessage());
- }
- }
- }
- }
加上手动确认后服务重启,每次都会从上次offset 确认的位置开始消费