流式圖表拒絕增刪改查之kafka核心消費(fèi)邏輯上篇
消費(fèi)邏輯
上文 流式圖表框架搭建
框架搭建好之后著手開發(fā)下kafka的核心消費(fèi)邏輯,流式圖表的核心消費(fèi)邏輯就是實(shí)現(xiàn)一個(gè)消費(fèi)鏈接池維護(hù)消費(fèi)者客戶端鏈接,將kafka client封裝成Runable任務(wù)提交到線程池里做一個(gè)常駐線程,實(shí)時(shí)消費(fèi)數(shù)據(jù),消費(fèi)到數(shù)據(jù)后存到redis中,并通過websocket推送到瀏覽器,瀏覽器刷新圖表實(shí)現(xiàn)流式圖表功能。
代碼設(shè)計(jì)
按照之前的代碼劃分,核心邏輯寫在matrix-core子模塊中,整體結(jié)構(gòu)用maven的父子模塊依賴?yán)^承的特性管理依賴。
maxtrix-core模塊只做kafka client的管理和消費(fèi)邏輯,盡量輕一點(diǎn),只需要引入redis和kafka依賴即可。
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency> <dependency> <groupId>com.uptown</groupId> <artifactId>matrix-common</artifactId> <version>1.0-SNAPSHOT</version> </dependency>
反序列化工具、線程池工具、lombok都放到matrix-common中,具體用google的包,這樣其他內(nèi)部模塊直接引用common模塊即可使用。
<dependency> <groupId>com.google.code.gson</groupId> <artifactId>gson</artifactId> </dependency>
消費(fèi)池
首先要?jiǎng)?chuàng)建出一個(gè)線程池出來,由于我們的業(yè)務(wù)要實(shí)時(shí)監(jiān)聽數(shù)據(jù),所以線程池提交的線程必須是個(gè)常駐線程。所以需要重寫線程池的任務(wù)失敗策略和異常處理器。
// 自定義異常處理器,捕獲錯(cuò)誤日志 @Slf4j public class ConsumerExceptionHandler implements Thread.UncaughtExceptionHandler { @Override public void uncaughtException(Thread t, Throwable e) { log.error(e.getMessage(), e); } }
// 任務(wù)失敗策略 @Slf4j class ConsumerThreadPoolExecutor extends ThreadPoolExecutor { ConsumerThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, rejectedExecutionHandler); } @Override protected void afterExecute(Runnable r, Throwable t) { super.afterExecute(r, t); //若線程執(zhí)行某任務(wù)失敗了,重新提交該任務(wù) if (t != null) { log.error("restart kafka consumer task for {}", (Object) t.getStackTrace()); } execute(r); } }
剩下的創(chuàng)建出線程池即可,消費(fèi)邏輯中只需要注入到具體類中即可。
@Data @Component @Slf4j public class KafkaConsumerConfig { // 線程池維護(hù)線程的最少數(shù)量 @Value(value = "${kafka.core-pool-size:20}") private int corePoolSize; // 線程池維護(hù)線程的最大數(shù)量 @Value(value = "${kafka.max-pool-size:20}") private int maxPoolSize; // 線程池維護(hù)線程所允許的空閑時(shí)間 @Value(value = "${kafka.keep-alive-time:0}") private int keepAliveTime; // 線程池所使用的緩沖隊(duì)列大小 @Value(value = "${kafka.work-queue-size:0}") private int workQueueSize; // 統(tǒng)一存放kafka客戶端的map @Bean public Map<String, KafkaConsumerRunnable> globalKafkaConsumerThreadMap() { return Maps.newConcurrentMap(); } /** * kafka監(jiān)聽任務(wù) 線程池 */ @Bean(name = "defaultThreadPool") public ThreadPoolExecutor defaultThreadPool() { // 使用google線程工廠 線程掛掉重啟策略 ConsumerExceptionHandler exceptionHandler = new ConsumerExceptionHandler(); ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("kafka-consumer-%d") .setUncaughtExceptionHandler(exceptionHandler).build(); return new ConsumerThreadPoolExecutor( corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS, new LinkedBlockingDeque<>(maxPoolSize), threadFactory, new ThreadPoolExecutor.CallerRunsPolicy() ); } }
這么搞的主要原因是防止消費(fèi)線程中出現(xiàn)消費(fèi)異常,比如反序列化異常、客戶端監(jiān)聽網(wǎng)絡(luò)異常等,為啥不在任務(wù)中try catch住異常是因?yàn)檫@樣做更優(yōu)雅點(diǎn),讓kafka client和線程的生命綁定一塊,比較好管理。
統(tǒng)一存放kafka客戶端的map算是做一個(gè)統(tǒng)計(jì),統(tǒng)計(jì)內(nèi)存中已提交的kafka監(jiān)聽線程數(shù),具體的Runable任務(wù)放在下一篇提供,畢竟上班寫文章容易翻車。
以上就是流式圖表拒絕增刪改查之kafka核心消費(fèi)邏輯上篇的詳細(xì)內(nèi)容,更多關(guān)于kafka消費(fèi)流式圖表的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
Java中雙冒號(hào)(::)運(yùn)算操作符用法詳解
這篇文章主要給大家介紹了關(guān)于Java中雙冒號(hào)(::)運(yùn)算操作符用法的相關(guān)資料,雙冒號(hào)運(yùn)算操作符是類方法的句柄,lambda表達(dá)式的一種簡寫,這種簡寫的學(xué)名叫eta-conversion或者叫η-conversion,需要的朋友可以參考下2023-11-11java實(shí)現(xiàn)圖片任意角度旋轉(zhuǎn)
這篇文章主要為大家詳細(xì)介紹了java實(shí)現(xiàn)圖片任意角度旋轉(zhuǎn),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2019-04-04java并發(fā)編程專題(八)----(JUC)實(shí)例講解CountDownLatch
這篇文章主要介紹了java CountDownLatch的相關(guān)資料,文中示例代碼非常詳細(xì),幫助大家理解和學(xué)習(xí),感興趣的朋友可以了解下2020-07-07關(guān)于任務(wù)調(diào)度框架quartz使用(異常處理,解決恢復(fù)后多次調(diào)度處理)
這篇文章主要介紹了關(guān)于任務(wù)調(diào)度框架quartz使用(異常處理,解決恢復(fù)后多次調(diào)度處理),具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-12-12Mybatis?saveAndUpdate空值不更新問題及解決
這篇文章主要介紹了Mybatis?saveAndUpdate空值不更新問題及解決方案,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-02-02