您好,登錄后才能下訂單哦!
小編給大家分享一下springboot中怎么實(shí)現(xiàn)kafa指定offset消費(fèi),相信大部分人都還不怎么了解,因此分享這篇文章給大家參考一下,希望大家閱讀完這篇文章后大有收獲,下面讓我們一起去了解一下吧!
kafka消費(fèi)過程難免會遇到需要重新消費(fèi)的場景,例如我們消費(fèi)到kafka數(shù)據(jù)之后需要進(jìn)行存庫操作,若某一時(shí)刻數(shù)據(jù)庫down了,導(dǎo)致kafka消費(fèi)的數(shù)據(jù)無法入庫,為了彌補(bǔ)數(shù)據(jù)庫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啟動時(shí)自動運(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一對一的消費(fèi)模型,每個(gè)consumer各自管理自己的partition。
@Service@Slf4jpublic class TspLogbookAnalysisService implements CommandLineRunner { //聲明kafka分區(qū)數(shù)相等的消費(fèi)線程數(shù),一個(gè)分區(qū)對應(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對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對應(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中怎么實(shí)現(xiàn)kafa指定offset消費(fèi)”這篇文章的所有內(nèi)容,感謝各位的閱讀!相信大家都有了一定的了解,希望分享的內(nèi)容對大家有所幫助,如果還想學(xué)習(xí)更多知識,歡迎關(guān)注億速云行業(yè)資訊頻道!
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。