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核心消费逻辑源码分析”有了更深的了解，不妨来实际操作一番吧！这里是亿速云网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！