好端端的线程池,怎么就卡死了?

写在前面

最近,我们的业务收到一项报障,线上某个业务模块偶尔会出现无法正常工作的情况。

经过多方排查,最终确认是线程池使用方式不合理导致的。鉴于线程池使用的普遍性和该类问题的隐秘性,本文将其中涉及的“坑”整理出来,与大家分享。

本文将尽可能淡化业务本身,着重介绍其中的技术问题。

场景说明

该业务链路有三个节点,分别为“演示调度入口”、“获取数据”,以及“推送数据”:

好端端的线程池,怎么就卡死了?

其中,三个节点配置了同一个线程池,为呈现更为直观,将实际业务代码做了必要的内联和简化处理,如下:

private ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(5);  public void handle() {     // 线程延迟调度(第一个节点,调度后续两个节点)     executor.schedule(() -> {         doHandle();     }, 10L, TimeUnit.MILLISECONDS); }  private List<String> fetchData()  {     CompletableFuture<List<String>> future = CompletableFuture.supplyAsync(() -> {         // 模拟数据获取业务逻辑(第二个节点)     }, executor);     return future.get(); }  private void doHandle() {     // 推送数据业务逻辑(第三个节点,需要等第二个节点的数据获取完毕)     List<String> data = fetchData(taskId);     executor.schedule(() -> {         push(data);     }, 50L, TimeUnit.MILLISECONDS); } 

有经验的同学可以一眼看出,同一个链路上的多个业务节点,若共用同一个线程池,可能会出问题。

知识回顾

在接着往下讲之前,先来回顾一下基础知识。

Java线程池的运行机制

任务提交到线程池中后:

  1. 如果workerCount < corePoolSize,则创建并启动一个线程来执行新提交的任务。
  2. 如果workerCount >= corePoolSize,且线程池内的阻塞队列未满,则将任务添加到该阻塞队列中。
  3. 如果workerCount >= corePoolSize且线程池内的阻塞队列已满,判断 workerCount < maximumPoolSize 是否成立,若成立,则创建并启动一个线程来执行新提交的任务。
  4. 如果workerCount >= maximumPoolSize,并且线程池内的阻塞队列已满, 则根据拒绝策略来处理该任务。

下面两个图很清楚地呈现了这个过程(源自参考资料[2]):

好端端的线程池,怎么就卡死了?

好端端的线程池,怎么就卡死了?

进一步地,线程池的生命周期定义在java.util.concurrent.ThreadPoolExecutor 类中,这个不是本文重点,有兴趣的同学自行学习即可:

// runState is stored in the high-order bits private static final int RUNNING    = -1 << COUNT_BITS; private static final int SHUTDOWN   =  0 << COUNT_BITS; private static final int STOP       =  1 << COUNT_BITS; private static final int TIDYING    =  2 << COUNT_BITS; private static final int TERMINATED =  3 << COUNT_BITS; 

好端端的线程池,怎么就卡死了?

Java线程的生命周期

定义在 java.lang.Thread 类中:

public enum State {     NEW,     RUNNABLE,     BLOCKED,     WAITING,     TIMED_WAITING,     TERMINATED; } 

状态转换图如下(源自参考资料[1]):

好端端的线程池,怎么就卡死了?

在我们将要描述的场景中,大量线程就是卡在了waiting状态。

问题推演

我们在代码中定义了一个线程池,其核心线程数为5、配置了无界阻塞队列、最大线程数为 Integer.MAX_VALUE(约21亿):

private ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(5);  public ScheduledThreadPoolExecutor(int corePoolSize) {     super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,           new DelayedWorkQueue()); }  public ThreadPoolExecutor(int corePoolSize,                           int maximumPoolSize,                           long keepAliveTime,                           TimeUnit unit,                           BlockingQueue<Runnable> workQueue) {     this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,          Executors.defaultThreadFactory(), defaultHandler); } 

那么任务并发调用时,问题是如何产生的呢?一个简单的办法是进行推演和模拟。
我们写一段代码进行复现(加了一些日志输出):

import java.util.ArrayList; import java.util.List; import java.util.Random; import java.util.concurrent.*;  public class TaskScheduler {      // 增加线程池的线程数量,以便模拟多线程抢占     private ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(5);      public void handle(int taskId) {         // 线程延迟调度(第一个节点,调度后续两个节点)         executor.schedule(() -> {             System.out.println("任务 " + taskId + " - 第一个节点开始执行,线程编号: " + Thread.currentThread().getId());             try {                 doHandle(taskId);             } catch (ExecutionException e) {                 throw new RuntimeException(e);             } catch (InterruptedException e) {                 throw new RuntimeException(e);             }             System.out.println("任务 " + taskId + " - 第一个节点执行完成,线程编号: " + Thread.currentThread().getId());         }, 10L, TimeUnit.MILLISECONDS);     }      private List<String> fetchData(int taskId) throws ExecutionException, InterruptedException {         CompletableFuture<List<String>> future = CompletableFuture.supplyAsync(() -> {             System.out.println("任务 " + taskId + " - 第二个节点开始执行,线程编号: " + Thread.currentThread().getId());             // 模拟数据获取业务逻辑(第二个节点)              Random random = new Random();             int delay = 500 + random.nextInt(2000); // 随机延迟时间             try {                 Thread.sleep(delay); // 模拟耗时操作             } catch (InterruptedException e) {                 throw new RuntimeException(e);             }             System.out.println("任务 " + taskId + " - 第二个节点执行完成,线程编号: " + Thread.currentThread().getId());             return new ArrayList<String>() {{                 add("data1");                 add("data2");                 add("data3");             }};         }, executor);          return future.get();     }      private void doHandle(int taskId) throws ExecutionException, InterruptedException {         // 推送数据业务逻辑(第三个节点,需要等第二个节点的数据获取完毕)         List<String> data = fetchData(taskId);         System.out.println("任务 " + taskId + " - 第二个节点的数据已获取,线程编号: " + Thread.currentThread().getId());          executor.schedule(() -> {             System.out.println("任务 " + taskId + " - 第三个节点开始执行,线程编号: " + Thread.currentThread().getId());             push(data);             System.out.println("33[0;31m任务 " + taskId + " - 第三个节点执行完成,线程编号: " + Thread.currentThread().getId() + "33[0m");         }, 50L, TimeUnit.MILLISECONDS);     }      private void push(List<String> data) {         // 模拟数据推送         System.out.println("推送数据: " + data);     }          /**      * 监控线程池状态      */     private static void monitorThreadPoolStatus(ScheduledThreadPoolExecutor executor) {         ScheduledExecutorService monitorExecutor = Executors.newSingleThreadScheduledExecutor();              monitorExecutor.scheduleAtFixedRate(() -> {             int poolSize = executor.getPoolSize();             int corePoolSize = executor.getCorePoolSize();             int maximumPoolSize = executor.getMaximumPoolSize();             long completedTaskCount = executor.getCompletedTaskCount();             int activeCount = executor.getActiveCount();             int queueSize = executor.getQueue().size();                  System.out.println("n【线程池状态】");             System.out.println("核心线程数: " + corePoolSize);             System.out.println("最大线程数: " + maximumPoolSize);             System.out.println("当前线程数: " + poolSize);             System.out.println("活跃线程数: " + activeCount);             System.out.println("已完成任务数: " + completedTaskCount);             System.out.println("队列中等待的任务数: " + queueSize);             System.out.println("是否关闭: " + executor.isShutdown());             System.out.println("是否终止: " + executor.isTerminated());         }, 0, 1, TimeUnit.SECONDS); // 每秒打印一次              // 在适当的时候关闭监控器(例如主程序结束时)         Runtime.getRuntime().addShutdownHook(new Thread(() -> {             monitorExecutor.shutdownNow();         }));     }      // 模拟任务数量(可自由调整)     private static final int TASK_COUNT = 5;      public static void main(String[] args) throws InterruptedException {         TaskScheduler scheduler = new TaskScheduler();         ExecutorService taskExecutor = Executors.newFixedThreadPool(TASK_COUNT); // 创建一个固定大小的线程池          CountDownLatch startSignal = new CountDownLatch(1); // 起跑线         CountDownLatch readySignal = new CountDownLatch(TASK_COUNT); // 用于通知主线程所有线程已就位          // 模拟多个并发任务         for (int i = 1; i <= TASK_COUNT; i++) {             final int taskId = i;             taskExecutor.submit(() -> {                 readySignal.countDown(); // 通知主线程当前线程已准备就绪                 try {                     startSignal.await(); // 等待起跑信号                     scheduler.handle(taskId);                 } catch (InterruptedException e) {                     Thread.currentThread().interrupt();                 }             });         }          System.out.println("等待所有线程就绪...");         readySignal.await(); // 等待所有线程准备完毕         System.out.println("所有线程已准备就绪,开始并发执行!");          startSignal.countDown(); // 放行所有等待的线程                  // 启动线程池状态监控         monitorThreadPoolStatus(scheduler.executor);          taskExecutor.shutdown(); // 后续会自动关闭,当所有任务完成时     } } 

执行之,观察输出日志。

当任务数量小于5时(比如2):

等待所有线程就绪... 所有线程已准备就绪,开始并发执行! 任务 2 - 第一个节点开始执行,线程编号: 25 任务 1 - 第一个节点开始执行,线程编号: 23 任务 1 - 第二个节点开始执行,线程编号: 27 任务 2 - 第二个节点开始执行,线程编号: 28 任务 1 - 第二个节点执行完成,线程编号: 27 任务 1 - 第二个节点的数据已获取,线程编号: 23 任务 1 - 第一个节点执行完成,线程编号: 23 任务 1 - 第三个节点开始执行,线程编号: 27 推送数据: [data1, data2, data3] 任务 1 - 第三个节点执行完成,线程编号: 27 任务 2 - 第二个节点执行完成,线程编号: 28 任务 2 - 第二个节点的数据已获取,线程编号: 25 任务 2 - 第一个节点执行完成,线程编号: 25 任务 2 - 第三个节点开始执行,线程编号: 23 推送数据: [data1, data2, data3] 任务 2 - 第三个节点执行完成,线程编号: 23 

最终线程池状态为:

核心线程数: 5 最大线程数: 2147483647 当前线程数: 5 活跃线程数: 0 已完成任务数: 6 队列中等待的任务数: 0 是否关闭: false 是否终止: false 

当任务数量大于等于5时(比如5):

等待所有线程就绪... 所有线程已准备就绪,开始并发执行! 任务 5 - 第一个节点开始执行,线程编号: 27 任务 1 - 第一个节点开始执行,线程编号: 29 任务 4 - 第一个节点开始执行,线程编号: 28 任务 3 - 第一个节点开始执行,线程编号: 30 任务 2 - 第一个节点开始执行,线程编号: 31 

最终线程池状态为:

核心线程数: 5 最大线程数: 2147483647 当前线程数: 5 活跃线程数: 5 已完成任务数: 0 队列中等待的任务数: 5 是否关闭: false 是否终止: false 

可以看出,当任务数量大于线程池的数量时,任务就会卡死,因为后续的节点已经没有线程可用了,对应的任务始终无法完成,因此已经被占用的线程无法释放(因为属于同一个任务)。此后继续提交的新任务会被放到无界阻塞队列中,表现出来就是系统处于假死状态。

以5个任务为例,卡死之后的线程池是这样:

好端端的线程池,怎么就卡死了?

以6个任务为例,卡死之后的线程池是这样:

好端端的线程池,怎么就卡死了?

以此类推...

再看看线程状态(以5个任务为例):

好端端的线程池,怎么就卡死了?

好端端的线程池,怎么就卡死了?

可以看到,状态均为WAITING。

刨根问底

1. 为何多年前的雷,现在才炸?

其实不是最近才出问题,而是可能早已出问题了,但无人知晓。相关的几个原因:

  • 业务本身流量不高,问题可能要在某些流量高峰期间才被触发。从上面的分析过程来看,只要同一时刻提交的任务数量小于5,就是可以持续运行下去的(因为后续两个节点,一定会被逐步执行完毕,对应的线程就会被释放掉了)。
  • 需求迭代频繁,机器隔几天就会部署重启。即使之前部分机器引起了线程卡死,也会被“不经意间”解决掉(从这个角度看,大促期间扩容还是有必要的/doge)。
  • 业务监控不完善。线程卡死后,后续的业务逻辑即使未执行,也没有被感知到,这反映出:缺乏对应的监控预警信息。
  • 相关功能在此前不够受重视。本次的故障是一个辅助功能,此前可能已经出问题了,但没有收到业务反馈。最近业务方开始关注数据指标,进而发现了异常。

2. 为什么CPU没有异常?

因为本次故障并不消耗CPU。线程处于等待状态,没有实际“工作”,当然也就不会引发CPU占用率升高。

另外提一句,当时我们还看了火焰图,尝试发现相关的“死锁”,显而易见,并不会有任何收获。

3. 为什么内存没有异常?

根据上述分析,发生线程池卡死的现象时,后续提交进来的任务均会被放到阻塞队列中,按理会使得内存不断增长,从而引发内存溢出,但我们在排查过程中并未观察到内存异常。那么是什么原因呢?

我们看看提交到阻塞队列中的对象占了多大内存,改造一下handle方法,引入jol工具将其打印出来:

public void handle(int taskId) {     Runnable task = () -> {         System.out.println("任务 " + taskId + " - 第一个节点开始执行,线程编号: " + Thread.currentThread().getId());         try {             doHandle(taskId);         } catch (ExecutionException e) {             throw new RuntimeException(e);         } catch (InterruptedException e) {             throw new RuntimeException(e);         }         System.out.println("任务 " + taskId + " - 第一个节点执行完成,线程编号: " + Thread.currentThread().getId());     };      // 使用 JOL 打印该任务对象的内存大小     System.out.println("n【任务对象内存占用】");     System.out.println(GraphLayout.parseInstance(task).toFootprint());      executor.schedule(task, 10L, TimeUnit.MILLISECONDS); } 

输出:

     COUNT       AVG       SUM   DESCRIPTION          1        16        16   TaskScheduler          1        24        24   TaskScheduler$$Lambda$5/74534624 

可以看到对象大小为24字节。当然,这是简化后的程序,实际的业务代码还要大一些,保守起见,预估为50字节。
假设每天10万次请求,那么会产生内存约为5M,按照平均每个月发版一次(实际大家的业务应该更为频繁)的节奏,仅会积累150M的内存,这个量级并不算高,也未引起JVM的“重视”。其实,如果把持续运行很久的内存dump下来,是可以发现端倪的。

内存计算方案有多种,除了JOL,也可以使用这个开源工具类:https://github.com/sunshanpeng/dark_magic

解决方案

讲完了问题,谈谈如何解决。

对于我们遇到的问题来说,是因为一个任务链路中的多个节点共用了同一个线程池,从而导致多个任务的前置节点把线程消耗完毕,后续资源没有线程去执行。从这个角度来看,可行的解决方案有几种:

方案1. 假设确实需要共用线程池,可以把线程池的核心线程数调大,比业务高峰期间的流量更高即可。当然,这个方案,并不算很优雅。

方案2. 任务链路中的多个节点,拆分独立线程池。这种可以从根源上避免线程争用(因为节点3总是会执行完毕的,对应任务占用的线程池2和线程池1会被逐级释放)。如下图:

好端端的线程池,怎么就卡死了?

方案3. 重新审视链路中的多层节点,是否必须异步执行,如某些地方其实可以改为同步执行。

最终,我们采用了方案2,改造后的代码类似于:

// 独立线程池:用于第一个调度节点(延迟执行 doHandle) private ScheduledThreadPoolExecutor stage1Pool = new ScheduledThreadPoolExecutor(5);  // 独立线程池:用于数据获取(第二个节点) private ExecutorService stage2Pool = Executors.newFixedThreadPool(5);  // 独立线程池:用于推送数据(第三个节点) private ExecutorService stage3Pool = Executors.newFixedThreadPool(10);  public void handle() {     // 线程延迟调度(第一个节点,调度后续两个节点)     stage1Pool.schedule(() -> {         doHandle();     }, 10L, TimeUnit.MILLISECONDS); }   private List<String> fetchData()  {     CompletableFuture<List<String>> future = CompletableFuture.supplyAsync(() -> {         // 模拟数据获取业务逻辑(第二个节点)     }, stage2Pool);     return future.get(); }  private void doHandle() {     // 推送数据业务逻辑(第三个节点,需要等第二个节点的数据获取完毕)     List<String> data = fetchData(taskId);     stage3Pool.schedule(() -> {         push(data);     }, 50L, TimeUnit.MILLISECONDS); } 

另外还有个新问题:如何合理设置线程池参数?其实这里面也有一套方法论。由于不是本文重点,不再展开,感兴趣的读者请参考此前写的一篇文章:https://www.cnblogs.com/xiaoxi666/p/16755570.html

经验教训

要有全局观

本文为了表述方便,对代码做了简化,实际的业务逻辑较长,且为不同时期的历史逻辑,写代码时容易忽略全局,导致同一个线程池配置在同一个链路的多个节点而不自知。这是很有风险的。

要深刻掌握技术,才能直击本质

即便看出来了同一个线程池被链路中的多个节点复用,也不一定能意识到可能的风险。我们在排查的过程中就曾忽略这个方向,多花了很多时间(又是查CPU,又是看内存和GC,又是看火焰图,直到发现各项指标都正常时,才回过头重新审视代码,进而找到根因)。

要有意识地逐步重构代码

在开发过程中,遇到历史上不合理的逻辑,鼓励大胆提出来,共同探讨出更合适的方案并执行小步重构,防患于未然。

闭环思维

遇到“诡异”问题,势必要挖掘根因,不能让可能的问题处于悬而未决的状态,可能出问题的地方在将来一定会出问题。

参考文档

[1] Life Cycle of a Thread in Java

[2] Java线程池实现原理及其在美团业务中的实践

发表评论

评论已关闭。

相关文章