springboot中怎么實現(xiàn)kafa指定offset消費(fèi)-創(chuàng)新互聯(lián)

小編給大家分享一下springboot中怎么實現(xiàn)kafa指定offset消費(fèi)，相信大部分人都還不怎么了解，因此分享這篇文章給大家參考一下，希望大家閱讀完這篇文章后大有收獲，下面讓我們一起去了解一下吧！

創(chuàng)新互聯(lián)建站是由多位在大型網(wǎng)絡(luò)公司、廣告設(shè)計公司的優(yōu)秀設(shè)計人員和策劃人員組成的一個具有豐富經(jīng)驗的團(tuán)隊，其中包括網(wǎng)站策劃、網(wǎng)頁美工、網(wǎng)站程序員、網(wǎng)頁設(shè)計師、平面廣告設(shè)計師、網(wǎng)絡(luò)營銷人員及形象策劃。承接：網(wǎng)站設(shè)計制作、網(wǎng)站設(shè)計、網(wǎng)站改版、網(wǎng)頁設(shè)計制作、網(wǎng)站建設(shè)與維護(hù)、網(wǎng)絡(luò)推廣、數(shù)據(jù)庫開發(fā),以高性價比制作企業(yè)網(wǎng)站、行業(yè)門戶平臺等全方位的服務(wù)。

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

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

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

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

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

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

@Service@Slf4jpublic class TspLogbookAnalysisService implements CommandLineRunner { //聲明kafka分區(qū)數(shù)相等的消費(fèi)線程數(shù)，一個分區(qū)對應(yīng)一個消費(fèi)線程 private static final int consumeThreadNum = 9; //特殊指定每個分區(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對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, ""); ...更多配置項 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對應(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));  } }}

以上是“springboot中怎么實現(xiàn)kafa指定offset消費(fèi)”這篇文章的所有內(nèi)容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內(nèi)容對大家有所幫助，如果還想學(xué)習(xí)更多知識，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道！

當(dāng)前文章：springboot中怎么實現(xiàn)kafa指定offset消費(fèi)-創(chuàng)新互聯(lián)
地址分享：http://muchs.cn/article30/dpodpo.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供App設(shè)計、App開發(fā)、Google、靜態(tài)網(wǎng)站、云服務(wù)器、移動網(wǎng)站建設(shè)

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)