欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

流式圖表拒絕增刪改查之kafka核心消費邏輯上篇

 更新時間:2023年04月12日 15:07:17   作者:在下uptown  
這篇文章主要為大家介紹了流式圖表拒絕增刪改查之kafka核心消費邏輯詳解的上篇,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪

消費邏輯

上文 流式圖表框架搭建

框架搭建好之后著手開發(fā)下kafka的核心消費邏輯,流式圖表的核心消費邏輯就是實現(xiàn)一個消費鏈接池維護(hù)消費者客戶端鏈接,將kafka client封裝成Runable任務(wù)提交到線程池里做一個常駐線程,實時消費數(shù)據(jù),消費到數(shù)據(jù)后存到redis中,并通過websocket推送到瀏覽器,瀏覽器刷新圖表實現(xiàn)流式圖表功能。

代碼設(shè)計

按照之前的代碼劃分,核心邏輯寫在matrix-core子模塊中,整體結(jié)構(gòu)用maven的父子模塊依賴?yán)^承的特性管理依賴。

maxtrix-core模塊只做kafka client的管理和消費邏輯,盡量輕一點,只需要引入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>

消費池

首先要創(chuàng)建出一個線程池出來,由于我們的業(yè)務(wù)要實時監(jiān)聽數(shù)據(jù),所以線程池提交的線程必須是個常駐線程。所以需要重寫線程池的任務(wù)失敗策略和異常處理器。

// 自定義異常處理器,捕獲錯誤日志
@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)建出線程池即可,消費邏輯中只需要注入到具體類中即可。

@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ù)線程所允許的空閑時間
    @Value(value = "${kafka.keep-alive-time:0}")
    private int keepAliveTime;
    // 線程池所使用的緩沖隊列大小
    @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()
      );
    }
}

這么搞的主要原因是防止消費線程中出現(xiàn)消費異常,比如反序列化異常、客戶端監(jiān)聽網(wǎng)絡(luò)異常等,為啥不在任務(wù)中try catch住異常是因為這樣做更優(yōu)雅點,讓kafka client和線程的生命綁定一塊,比較好管理。

統(tǒng)一存放kafka客戶端的map算是做一個統(tǒng)計,統(tǒng)計內(nèi)存中已提交的kafka監(jiān)聽線程數(shù),具體的Runable任務(wù)放在下一篇提供,畢竟上班寫文章容易翻車。

以上就是流式圖表拒絕增刪改查之kafka核心消費邏輯上篇的詳細(xì)內(nèi)容,更多關(guān)于kafka消費流式圖表的資料請關(guān)注腳本之家其它相關(guān)文章!

相關(guān)文章

  • 詳解JAVA類加載機(jī)制(推薦)

    詳解JAVA類加載機(jī)制(推薦)

    這篇文章主要介紹了JAVA類加載機(jī)制的相關(guān)資料,非常不錯,具有參考借鑒價值,需要的朋友可以參考下
    2016-09-09
  • Java中雙冒號(::)運算操作符用法詳解

    Java中雙冒號(::)運算操作符用法詳解

    這篇文章主要給大家介紹了關(guān)于Java中雙冒號(::)運算操作符用法的相關(guān)資料,雙冒號運算操作符是類方法的句柄,lambda表達(dá)式的一種簡寫,這種簡寫的學(xué)名叫eta-conversion或者叫η-conversion,需要的朋友可以參考下
    2023-11-11
  • java實現(xiàn)圖片任意角度旋轉(zhuǎn)

    java實現(xiàn)圖片任意角度旋轉(zhuǎn)

    這篇文章主要為大家詳細(xì)介紹了java實現(xiàn)圖片任意角度旋轉(zhuǎn),具有一定的參考價值,感興趣的小伙伴們可以參考一下
    2019-04-04
  • SpringBoot?集成短信和郵件的配置示例詳解

    SpringBoot?集成短信和郵件的配置示例詳解

    這篇文章主要介紹了SpringBoot?集成短信和郵件的相關(guān)知識,項目中使用lombok插件和swagger依賴,無相關(guān)依賴的請自行修改,本文通過實例代碼給大家介紹的非常詳細(xì),需要的朋友可以參考下
    2022-04-04
  • java并發(fā)編程專題(八)----(JUC)實例講解CountDownLatch

    java并發(fā)編程專題(八)----(JUC)實例講解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)度處理)

    這篇文章主要介紹了關(guān)于任務(wù)調(diào)度框架quartz使用(異常處理,解決恢復(fù)后多次調(diào)度處理),具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教
    2023-12-12
  • JAVA中的字符串常量池使用操作代碼

    JAVA中的字符串常量池使用操作代碼

    Java中的字符串常量池是Java堆中的一塊特殊存儲區(qū)域,用于存儲字符串。它的實現(xiàn)是為了提高字符串操作的性能并節(jié)省內(nèi)存,這篇文章主要介紹了JAVA中的字符串常量池,需要的朋友可以參考下
    2022-12-12
  • Java實現(xiàn)基本排序算法的示例代碼

    Java實現(xiàn)基本排序算法的示例代碼

    排序就是將一串記錄按照其中某個或某些關(guān)鍵字的大小,遞增或遞減的排列起來的操作。本文將用Java實現(xiàn)一些基本的排序算法,感興趣的可以了解一下
    2022-07-07
  • Java 并行數(shù)據(jù)處理和性能分析

    Java 并行數(shù)據(jù)處理和性能分析

    這篇文章主要介紹了Java 并行數(shù)據(jù)處理和性能分析,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧
    2020-10-10
  • Mybatis?saveAndUpdate空值不更新問題及解決

    Mybatis?saveAndUpdate空值不更新問題及解決

    這篇文章主要介紹了Mybatis?saveAndUpdate空值不更新問題及解決方案,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教
    2023-02-02

最新評論