devbox
你好赵伟
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

springboot中实现kafka指定offset消费_spring kafka offset

作者：你好赵伟 | 2024-05-02 15:49:56

赞

踩

spring kafka offset

1115df6a0266f75ada0e2c211719310a.png

kafka消费过程难免会遇到需要重新消费的场景，例如我们消费到kafka数据之后需要进行存库操作，若某一时刻数据库down了，导致kafka消费的数据无法入库，为了弥补数据库down期间的数据损失，有一种做法我们可以指定kafka消费者的offset到之前某一时间的数值，然后重新进行消费。
首先创建kafka消费服务

  1. @Service
  2. @Slf4j
  3. //实现CommandLineRunner接口，在springboot启动时自动运行其run方法。
  4. public class TspLogbookAnalysisService implements CommandLineRunner {
  5.     @Override
  6.     public void run(String... args) {
  7.         //do something
  8.     }
  9. }

kafka消费模型建立
kafka server中每个主题存在多个分区（partition），每个分区自己维护一个偏移量（offset），我们的目标是实现kafka consumer指定offset消费。

在这里使用consumer-->partition一对一的消费模型，每个consumer各自管理自己的partition。
format,png

  1. @Service
  2. @Slf4j
  3. public class TspLogbookAnalysisService implements CommandLineRunner {
  4.     //声明kafka分区数相等的消费线程数，一个分区对应一个消费线程
  5.     private  static final int consumeThreadNum = 9;
  6.     //特殊指定每个分区开始消费的offset
  7.     private List<Long> partitionOffsets = Lists.newArrayList(1111,1112,1113,1114,1115,1116,1117,1118,1119);
  8.    
  9.     private ExecutorService executorService = Executors.newFixedThreadPool(consumeThreadNum);
  10.  
  11.     @Override
  12.     public void run(String... args) {
  13.         //循环遍历创建消费线程
  14.         IntStream.range(0, consumeThreadNum)
  15.                 .forEach(partitionIndex -> executorService.submit(() -> startConsume(partitionIndex)));
  16.     }
  17. }

 kafka consumer对offset的处理

声明kafka consumer的配置类

  1. private Properties buildKafkaConfig() {
  2.     Properties kafkaConfiguration = new Properties();
  3.     kafkaConfiguration.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "");
  4.     kafkaConfiguration.put(ConsumerConfig.GROUP_ID_CONFIG, "");
  5.     kafkaConfiguration.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "");
  6.     kafkaConfiguration.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "");
  7.     kafkaConfiguration.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "");
  8.     kafkaConfiguration.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "");
  9.     kafkaConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"");
  10.     kafkaConfiguration.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "");
  11.     ...更多配置项
  12.  
  13.     return kafkaConfiguration;
  14. }

创建kafka consumer，处理offset，开始消费数据任务

  1. private void startConsume(int partitionIndex) {
  2.     //创建kafka consumer
  3.     KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(buildKafkaConfig());
  4.  
  5.     try {
  6.         //指定该consumer对应的消费分区
  7.         TopicPartition partition = new TopicPartition(kafkaProperties.getKafkaTopic(), partitionIndex);
  8.         consumer.assign(Lists.newArrayList(partition));
  9.  
  10.         //consumer的offset处理
  11.         if (collectionUtils.isNotEmpty(partitionOffsets)  &&  partitionOffsets.size() == consumeThreadNum) {
  12.             Long seekOffset = partitionOffsets.get(partitionIndex);
  13.             log.info("partition:{} , offset seek from {}", partition, seekOffset);
  14.             consumer.seek(partition, seekOffset);
  15.         }
  16.         
  17.         //开始消费数据任务
  18.         kafkaRecordConsume(consumer, partition);
  19.     } catch (Exception e) {
  20.         log.error("kafka consume error:{}", ExceptionUtils.getFullStackTrace(e));
  21.     } finally {
  22.         try {
  23.             consumer.commitSync();
  24.         } finally {
  25.             consumer.close();
  26.         }
  27.     }
  28. }

消费数据逻辑，offset操作

  1. private void kafkaRecordConsume(KafkaConsumer<String, byte[]> consumer, TopicPartition partition) {
  2.     while (true) {
  3.         try {
  4.             ConsumerRecords<String, byte[]> records = consumer.poll(TspLogbookConstants.POLL_TIMEOUT);
  5.             //具体的处理流程
  6.             records.forEach((k) -> handleKafkaInput(k.key(), k.value()));
  7.  
  8.             //
    声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/你好赵伟/article/detail/524835
    推荐阅读
    相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。