kafka核心消費邏輯源碼分析

本篇內容主要講解“kafka核心消費邏輯源碼分析”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“kafka核心消費邏輯源碼分析”吧!

消費邏輯

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

代碼設計

按照之前的代碼劃分，核心邏輯寫在matrix-core子模塊中，整體結構用maven的父子模塊依賴繼承的特性管理依賴。

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的包，這樣其他內部模塊直接引用common模塊即可使用。

<dependency>
    <groupId>com.google.code.gson</groupId>
    <artifactId>gson</artifactId>
</dependency>

消費池

首先要創建出一個線程池出來，由于我們的業務要實時監聽數據，所以線程池提交的線程必須是個常駐線程。所以需要重寫線程池的任務失敗策略和異常處理器。

// 自定義異常處理器，捕獲錯誤日志
@Slf4j
public class ConsumerExceptionHandler implements Thread.UncaughtExceptionHandler {
   @Override
   public void uncaughtException(Thread t, Throwable e) {
        log.error(e.getMessage(), e);
   }
}

// 任務失敗策略
@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);
      //若線程執行某任務失敗了，重新提交該任務
      if (t != null) {
         log.error("restart kafka consumer task for {}", (Object) t.getStackTrace());
      }
      execute(r);
   }
}

剩下的創建出線程池即可，消費邏輯中只需要注入到具體類中即可。

@Data
@Component
@Slf4j
public class KafkaConsumerConfig {
    // 線程池維護線程的最少數量
    @Value(value = "${kafka.core-pool-size:20}")
    private int corePoolSize;
    // 線程池維護線程的最大數量
    @Value(value = "${kafka.max-pool-size:20}")
    private int maxPoolSize;
    // 線程池維護線程所允許的空閑時間
    @Value(value = "${kafka.keep-alive-time:0}")
    private int keepAliveTime;
    // 線程池所使用的緩沖隊列大小
    @Value(value = "${kafka.work-queue-size:0}")
    private int workQueueSize;
   // 統一存放kafka客戶端的map
   @Bean
   public Map<String, KafkaConsumerRunnable> globalKafkaConsumerThreadMap() {
      return Maps.newConcurrentMap();
   }
    /**
     * kafka監聽任務 線程池
     */
    @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()
      );
    }
}

這么搞的主要原因是防止消費線程中出現消費異常，比如反序列化異常、客戶端監聽網絡異常等，為啥不在任務中try catch住異常是因為這樣做更優雅點，讓kafka client和線程的生命綁定一塊，比較好管理。

到此，相信大家對“kafka核心消費邏輯源碼分析”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！