流式图表拒绝增删改查之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客户端的map算是做一个统计，统计内存中已提交的kafka监听线程数，具体的Runable任务放在下一篇提供，毕竟上班写文章容易翻车。

来源：https://juejin.cn/post/7220235452291416123