1.问题
项目中通过kafka来对接上游，在项目中写一个listener监听topTopic队列，for循环消费records，在for循环中处理成存储到es的对象，一次拉50条，使用自定义线程池esThreadPool异步推送到es中，但是每次停机就会丢数据，例：kafka消费了1000条，但是往es中存储比较慢，优雅停机的时候，esThreadPool还没完成存储的这部分数据就会丢失。
@KafkaListener(topics={"kafka.topics.publicFlow.topicName",groupId="TOP_TOPIC_READER",containerFactory="cluster1KafkaListenerFactory"})
public void consumer(List<ConsumerRecord<String,String>> records,Acknowledgment acknowledgment){
List<PublicFlowVo> esList = new ArrayList<>(records.size);
try{
for(ConsumerRecord<String,String> record : records){
//解析成PublicFlowVo
esList.add(build(record));
}
}catch (Exception e){
//打印错误日志
}finally{
//提交offset
acknowledgment.acknowledge();
//推送到es
esThreadPool.submit(()->{
//写入es
sendEs(esList);
});
}
}
2.分析原因
之所以这样，是因为项目集成了graceful优雅停机,springboot从2.3.0版本开始支持，想当然的认为优雅停机就能实现线程任务执行完之后再停机，这样就不会丢数。实际上，验证过后，才发现，graceful只能监听到主线程，假如是用了自定义线程池，要手动实现优雅停机。如果想不丢数，就必须先关闭消费kafka的线程，其次是等待esThreadPool任务执行完成之后，再关闭esThreadPool线程池，最后再停机。
3.解决方案
esThreadPool实现SmartLifecycle，可以自定义停机顺序，以及停机逻辑，下面是两种实现方案，一种可以设置最大等待时间，一种一直等到任务全部完成。
方案一：支持设置最大等待时间
@Component
public class EsThreadPoolManager implements SmartLifecycle {
private boolean isRunning = false;
@Resource(name="esThreadPool")
private ThreadPoolExecutor threadPoolExcutor;
@Override
public void start(){ isRunning = true;}
@Override
public void stop(){ stop(()->{});}
@Override
public void stop(@NotNull Runnable callback){
log.info("esThreadPool线程池开始停止!")
long start = System.currentTimeMillis();
// 关闭ThreadPoolExecutor，等待已提交的任务完成
threadPoolExcutor.shutdown();
try {
// 等待线程池终止，设置最长等待时间，这里示例为30秒，这个值和graceful优雅停机时间无关，这个值先生效，这个时间到了才会去按graceful的时间判断是否需要强制关闭
if (!threadPoolExcutor.awaitTermination(30, TimeUnit.SECONDS)) {
// 超时后强制关闭
threadPoolExcutor.shutdownNow();
}
} catch (InterruptedException ex) {
// 如果等待被中断，则直接关闭
Thread.currentThread().interrupt();
threadPoolExcutor.shutdownNow();
} finally {
long end = System.currentTimeMillis();
// 完成停机处理
log.info("esThreadPool线程池停止!耗时:{}s",(end-start)/1000);
callback.run();
isRunning = false;
}
}
@Override
public boolean isRunning() {
// 返回当前运行状态
return isRunning;
}
@Override
public int getPhase() {
// 控制停机顺序，数值越大越先停止
// 可以根据需要调整,如果先停止kafka，建议设置成小于2147483547的值，这个值可以在KafkaListenerEndpointRegistry这个类的getPhase()方法中找到。
return Integer.MAX_VALUE;
}
}
方案二：直到所有任务全部执行完成
@Component
public class EsThreadPoolManager implements SmartLifecycle {
private boolean isRunning = false;
@Resource(name="esThreadPool")
private ThreadPoolExecutor threadPoolExcutor;
@Override
public void start(){ isRunning = true;}
@Override
public void stop(){ stop(()->{});}
@Override
public void stop(@NotNull Runnable callback){
log.info("esThreadPool线程池开始停止!")
long start = System.currentTimeMillis();
// 关闭ThreadPoolExecutor，等待已提交的任务完成
threadPoolExcutor.shutdown();
try {
// 每隔一段时间检查线程池状态
while (true) {
// 线程池中活动线程为0，且队列中没有待执行任务时，可以安全关闭
if (gracefulThreadPool1.getActiveCount() == 0 && gracefulThreadPool1.getQueue().isEmpty()) {
break;
}
// 等待一段时间后再次检查
Thread.sleep(2000); // 2秒钟检查一次
}
} catch (InterruptedException ex) {
// 如果等待被中断，则直接关闭
Thread.currentThread().interrupt();
threadPoolExcutor.shutdownNow();
} finally {
long end = System.currentTimeMillis();
// 完成停机处理
log.info("esThreadPool线程池停止!耗时:{}s",(end-start)/1000);
callback.run();
isRunning = false;
}
}
@Override
public boolean isRunning() {
// 返回当前运行状态
return isRunning;
}
@Override
public int getPhase() {
// 控制停机顺序，数值越大越先停止
// 可以根据需要调整,如果先停止kafka，建议设置成小于2147483547的值，这个值可以在KafkaListenerEndpointRegistry这个类的getPhase()方法中找到。
return Integer.MAX_VALUE;
}
}