当前位置：首页 > news >正文

kakfa发版丢消息事件分析

news 来源：原创 2024/9/29 4:19:01

背景

其他部门同事反馈在项目发版/重启(kill -15)的那段时间，经常会出现导致 C 端业务出现问题，从而产生资损

一听资损，赶紧应答下来，了解了下具体情况，然后立马去排查了

问题分析

结合同事的描述以及对业务的了解，很快就定位到是 kafka 消息丢失导致 C 端业务出现问题

业务当前消费架构图

从上图可以了解到几个点会导致目前这个场景消息丢失

kafka 一秒一次的位移提交
Queue 队列没消费完任务
work 线程池从 Queue 中拉取的任务没消费完(每次拉取一个)

问题所在：因C端业务特性，非准实时的消息是没有意义的(分钟级)，所以kafka的自动提交位移实际上是符合业务需求，三点结合起来看问题应该是出在：在发版时 消费单线程 依旧在拉取消息写入 Queue，并且后续的线程池也没有将 Queue中的任务给处理完

消费架构改造

改造消费流程
启动时增加JVM关闭钩子，在关闭前将 isRunning 修改为fale，从而停止 消费单线程 继续拉取kafka消息
优雅关闭 work线程池

// shutdown() 与 shutdownNow()这里也给到一段shutdown测试代码
ThreadPoolExecutor executorService =new ThreadPoolExecutor(1, 1, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
AtomicInteger integer = new AtomicInteger();
for (int i = 0; i < 100; i++) {executorService.execute(() -> {try {System.out.println(new Date() + "=====>" + integer.incrementAndGet());Thread.sleep(1000L);} catch (Exception e) {e.printStackTrace();}});
}Thread.sleep(5000L);
executorService.shutdown();
// executorService.shutdownNow();
System.out.println("线程池已触发shutdown");

随之而来的另一个问题，若在JVM关闭钩子中对 work线程池 操作shutdown，在任务中是有使用到Spring容器中的bean，若bean销毁了，那么work线程池中的任务都无法再执行成功(具体销毁优先级细则可自行百度，这里不做延伸)。
基于这个问题，回想到之前常用的一个注解 @PostConstruct 的一个孪生兄弟 @PreDestroy，这是在Java规范JSR-250引入的注解，定义了对象的创建和销毁工作，那么Spring必然对它有做支持，测试代码如下

ThreadPoolExecutor executorService =new ThreadPoolExecutor(1, 1, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<>());@PostConstruct
public void postConstruct(){AtomicInteger integer = new AtomicInteger();for (int i = 0; i < 100; i++) {executorService.execute(() -> {try {System.out.println(new Date() + "=====>" + integer.incrementAndGet());Thread.sleep(1000L);} catch (Exception e) {e.printStackTrace();}});}
}@PreDestroy
public void preDestroy(){executorService.shutdown();
}// 增加一个测试关闭的接口
@GetMapping("/shutdown")
public void shutdown() {System.exit(0);
}

测试结果依旧失败，看日志打印是正在处理线程池中已被接收的任务时挂掉的（这不科学，上面shutdown()测试案例结果明明会等待所有任务结束以后再结束），心里一群草姓的马飘过-_-

转念一想：其实这样也对，若一个池任务过多导致一直无法kill掉进程，这种行为也不对…那有没有什么补偿机制可以用，emm，山重水复疑无路，柳暗花明又一村哇，Doug Lea大神名不虚传，早就为我们考虑好了

// 贴出改动方法
@PreDestroy
public void preDestroy(){executorService.shutdown();try {if(executorService.awaitTermination(5, TimeUnit.SECONDS)){System.out.println("任务执行完毕结束");} else {System.out.println("time out 结束");}} catch (InterruptedException e) {System.out.println("Interrupted while waiting for executor");Thread.currentThread().interrupt();executorService.shutdownNow();}
}

嘿嘿，这么一改顺眼多了，线程池在shutdown后再至多等待N秒(若无任务则直接返回true)，业务可以根据特性去决定此值配置

但是这么写多麻烦，那么多重要的线程池各个都要在这里写，那Spring如何实现线程池的优雅停的呢？想到Spring的生命周期中的销毁回调，实现 DisposableBean 即可，那看看ThreadPoolTaskExecutor，其父类ExecutorConfigurationSupport在处理销毁时，会判定其 waitForTasksToCompleteOnShutdown 参数是否为true来决定是否要调用shutdown()，并且根据其 awaitTerminationSeconds 参数来决定是否需要调用 ExecutorService.awaitTermination 去等待线程池处理一定时间