快捷導(dǎo)航

springboot中如何實(shí)現(xiàn)kafa指定offset消費(fèi)

更新時(shí)間：2019年12月14日 16:38:21 作者：東溪陳姓少年

這篇文章主要介紹了springboot中如何實(shí)現(xiàn)kafa指定offset消費(fèi),文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

kafka消費(fèi)過程難免會(huì)遇到需要重新消費(fèi)的場(chǎng)景，例如我們消費(fèi)到kafka數(shù)據(jù)之后需要進(jìn)行存庫(kù)操作，若某一時(shí)刻數(shù)據(jù)庫(kù)down了，導(dǎo)致kafka消費(fèi)的數(shù)據(jù)無(wú)法入庫(kù)，為了彌補(bǔ)數(shù)據(jù)庫(kù)down期間的數(shù)據(jù)損失，有一種做法我們可以指定kafka消費(fèi)者的offset到之前某一時(shí)間的數(shù)值，然后重新進(jìn)行消費(fèi)。

首先創(chuàng)建kafka消費(fèi)服務(wù)

@Service
@Slf4j
//實(shí)現(xiàn)CommandLineRunner接口，在springboot啟動(dòng)時(shí)自動(dòng)運(yùn)行其run方法。
public class TspLogbookAnalysisService implements CommandLineRunner {
 @Override
 public void run(String... args) {
  //do something
 }
}

kafka消費(fèi)模型建立

kafka server中每個(gè)主題存在多個(gè)分區(qū)（partition），每個(gè)分區(qū)自己維護(hù)一個(gè)偏移量（offset），我們的目標(biāo)是實(shí)現(xiàn)kafka consumer指定offset消費(fèi)。

在這里使用consumer-->partition一對(duì)一的消費(fèi)模型，每個(gè)consumer各自管理自己的partition。

@Service
@Slf4j
public class TspLogbookAnalysisService implements CommandLineRunner {
 //聲明kafka分區(qū)數(shù)相等的消費(fèi)線程數(shù)，一個(gè)分區(qū)對(duì)應(yīng)一個(gè)消費(fèi)線程
 private static final int consumeThreadNum = 9;
 //特殊指定每個(gè)分區(qū)開始消費(fèi)的offset
 private List<Long> partitionOffsets = Lists.newArrayList(1111,1112,1113,1114,1115,1116,1117,1118,1119);
 
 private ExecutorService executorService = Executors.newFixedThreadPool(consumeThreadNum);

 @Override
 public void run(String... args) {
  //循環(huán)遍歷創(chuàng)建消費(fèi)線程
  IntStream.range(0, consumeThreadNum)
    .forEach(partitionIndex -> executorService.submit(() -> startConsume(partitionIndex)));
 }
}

kafka consumer對(duì)offset的處理

聲明kafka consumer的配置類

private Properties buildKafkaConfig() {
 Properties kafkaConfiguration = new Properties();
 kafkaConfiguration.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "");
 kafkaConfiguration.put(ConsumerConfig.GROUP_ID_CONFIG, "");
 kafkaConfiguration.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "");
 kafkaConfiguration.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "");
 kafkaConfiguration.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "");
 kafkaConfiguration.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "");
 kafkaConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"");
 kafkaConfiguration.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "");
 ...更多配置項(xiàng)

 return kafkaConfiguration;
}

創(chuàng)建kafka consumer，處理offset，開始消費(fèi)數(shù)據(jù)任務(wù)#

private void startConsume(int partitionIndex) {
 //創(chuàng)建kafka consumer
 KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(buildKafkaConfig());

 try {
  //指定該consumer對(duì)應(yīng)的消費(fèi)分區(qū)
  TopicPartition partition = new TopicPartition(kafkaProperties.getKafkaTopic(), partitionIndex);
  consumer.assign(Lists.newArrayList(partition));

  //consumer的offset處理
  if (collectionUtils.isNotEmpty(partitionOffsets) && partitionOffsets.size() == consumeThreadNum) {
   Long seekOffset = partitionOffsets.get(partitionIndex);
   log.info("partition:{} , offset seek from {}", partition, seekOffset);
   consumer.seek(partition, seekOffset);
  }
  
  //開始消費(fèi)數(shù)據(jù)任務(wù)
  kafkaRecordConsume(consumer, partition);
 } catch (Exception e) {
  log.error("kafka consume error:{}", ExceptionUtils.getFullStackTrace(e));
 } finally {
  try {
   consumer.commitSync();
  } finally {
   consumer.close();
  }
 }
}

消費(fèi)數(shù)據(jù)邏輯，offset操作

private void kafkaRecordConsume(KafkaConsumer<String, byte[]> consumer, TopicPartition partition) {
 while (true) {
  try {
   ConsumerRecords<String, byte[]> records = consumer.poll(TspLogbookConstants.POLL_TIMEOUT);
   //具體的處理流程
   records.forEach((k) -> handleKafkaInput(k.key(), k.value()));

   //🌿很重要：日志記錄當(dāng)前consumer的offset，partition相關(guān)信息(之后如需重新指定offset消費(fèi)就從這里的日志中獲取offset，partition信息)
   if (records.count() > 0) {
    String currentOffset = String.valueOf(consumer.position(partition));
    log.info("current records size is：{}, partition is: {}, offset is:{}", records.count(), consumer.assignment(), currentOffset);
   }
 
   //offset提交  
   consumer.commitAsync();
  } catch (Exception e) {
   log.error("handlerKafkaInput error{}", ExceptionUtils.getFullStackTrace(e));
  }
 }
}

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: