场景

最近的一个项目中涉及到文件上传和下载，使用到JUC的线程池ThreadPoolExecutor，在生产环境中出现了某些时刻线程池满负载运作，由于使用了CallerRunsPolicy拒绝策略，导致满负载情况下，应用接口调用无法响应，处于假死状态。考虑到之前用micrometer + prometheus + grafana搭建过监控体系，于是考虑使用micrometer做一次主动的线程池度量数据采集，最终可以相对实时地展示在grafana的面板中。

实践过程

下面通过真正的实战过程做一个仿真的例子用于复盘。

代码改造

首先我们要整理一下ThreadPoolExecutor中提供的度量数据项和micrometer对应的Tag的映射关系：

  • 线程池名称，Tag：thread.pool.name，这个很重要，用于区分各个线程池的数据，如果使用IOC容器管理，可以使用BeanName代替。
  • int getCorePoolSize()：核心线程数，Tag：thread.pool.core.size。
  • int getLargestPoolSize()：历史峰值线程数，Tag：thread.pool.largest.size。
  • int getMaximumPoolSize()：最大线程数(线程池线程容量)，Tag：thread.pool.max.size。
  • int getActiveCount()：当前活跃线程数，Tag：thread.pool.active.size。
  • int getPoolSize()：当前线程池中运行的线程总数(包括核心线程和非核心线程)，Tag：thread.pool.thread.count。
  • 当前任务队列中积压任务的总数，Tag：thread.pool.queue.size，这个需要动态计算得出。

接着编写具体的代码，实现的功能如下：

  • 1、建立一个ThreadPoolExecutor实例，核心线程和最大线程数为10，任务队列长度为10，拒绝策略为AbortPolicy。
  • 2、提供两个方法，分别使用线程池实例模拟短时间耗时的任务和长时间耗时的任务。
  • 3、提供一个方法用于清空线程池实例中的任务队列。
  • 4、提供一个单线程的调度线程池用于定时收集ThreadPoolExecutor实例中上面列出的度量项，保存到micrometer内存态的收集器中。

由于这些统计的值都会跟随时间发生波动性变更，可以考虑选用Gauge类型的Meter进行记录。

  1. COPY// ThreadPoolMonitor
  2. import io.micrometer.core.instrument.Metrics;
  3. import io.micrometer.core.instrument.Tag;
  4. import org.springframework.beans.factory.InitializingBean;
  5. import org.springframework.stereotype.Service;
  7. import java.util.Collections;
  8. import java.util.concurrent.*;
  9. import java.util.concurrent.atomic.AtomicInteger;
  11. /**
  12.  * @author throwable
  13.  * @version v1.0
  14.  * @description
  15.  * @since 2019/4/7 21:02
  16.  */
  17. @Service
  18. public class ThreadPoolMonitor implements InitializingBean {
  20.     private static final String EXECUTOR_NAME = "ThreadPoolMonitorSample";
  21.     private static final Iterable<Tag> TAG = Collections.singletonList(Tag.of("thread.pool.name", EXECUTOR_NAME));
  22.     private final ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
  24.     private final ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10, 0, TimeUnit.SECONDS,
  25.             new ArrayBlockingQueue<>(10), new ThreadFactory() {
  27.         private final AtomicInteger counter = new AtomicInteger();
  29.         @Override
  30.         public Thread newThread(Runnable r) {
  31.             Thread thread = new Thread(r);
  32.             thread.setDaemon(true);
  33.             thread.setName("thread-pool-" + counter.getAndIncrement());
  34.             return thread;
  35.         }
  36.     }, new ThreadPoolExecutor.AbortPolicy());
  39.     private Runnable monitor = () -> {
  40.         //这里需要捕获异常,尽管实际上不会产生异常,但是必须预防异常导致调度线程池线程失效的问题
  41.         try {
  42.             Metrics.gauge("thread.pool.core.size", TAG, executor, ThreadPoolExecutor::getCorePoolSize);
  43.             Metrics.gauge("thread.pool.largest.size", TAG, executor, ThreadPoolExecutor::getLargestPoolSize);
  44.             Metrics.gauge("thread.pool.max.size", TAG, executor, ThreadPoolExecutor::getMaximumPoolSize);
  45.             Metrics.gauge("thread.pool.active.size", TAG, executor, ThreadPoolExecutor::getActiveCount);
  46.             Metrics.gauge("thread.pool.thread.count", TAG, executor, ThreadPoolExecutor::getPoolSize);
  47.             // 注意如果阻塞队列使用无界队列这里不能直接取size
  48.             Metrics.gauge("thread.pool.queue.size", TAG, executor, e -> e.getQueue().size());
  49.         } catch (Exception e) {
  50.             //ignore
  51.         }
  52.     };
  54.     @Override
  55.     public void afterPropertiesSet() throws Exception {
  56.         // 每5秒执行一次
  57.         scheduledExecutor.scheduleWithFixedDelay(monitor, 0, 5, TimeUnit.SECONDS);
  58.     }
  60.     public void shortTimeWork() {
  61.         executor.execute(() -> {
  62.             try {
  63.                 // 5秒
  64.                 Thread.sleep(5000);
  65.             } catch (InterruptedException e) {
  66.                 //ignore
  67.             }
  68.         });
  69.     }
  71.     public void longTimeWork() {
  72.         executor.execute(() -> {
  73.             try {
  74.                 // 500秒
  75.                 Thread.sleep(5000 * 100);
  76.             } catch (InterruptedException e) {
  77.                 //ignore
  78.             }
  79.         });
  80.     }
  82.     public void clearTaskQueue() {
  83.         executor.getQueue().clear();
  84.     }
  85. }
  87. //ThreadPoolMonitorController
  88. import club.throwable.smp.service.ThreadPoolMonitor;
  89. import lombok.RequiredArgsConstructor;
  90. import org.springframework.http.ResponseEntity;
  91. import org.springframework.web.bind.annotation.GetMapping;
  92. import org.springframework.web.bind.annotation.RestController;
  94. /**
  95.  * @author throwable
  96.  * @version v1.0
  97.  * @description
  98.  * @since 2019/4/7 21:20
  99.  */
  100. @RequiredArgsConstructor
  101. @RestController
  102. public class ThreadPoolMonitorController {
  104.     private final ThreadPoolMonitor threadPoolMonitor;
  106.     @GetMapping(value = "/shortTimeWork")
  107.     public ResponseEntity<String> shortTimeWork() {
  108.         threadPoolMonitor.shortTimeWork();
  109.         return ResponseEntity.ok("success");
  110.     }
  112.     @GetMapping(value = "/longTimeWork")
  113.     public ResponseEntity<String> longTimeWork() {
  114.         threadPoolMonitor.longTimeWork();
  115.         return ResponseEntity.ok("success");
  116.     }
  118.     @GetMapping(value = "/clearTaskQueue")
  119.     public ResponseEntity<String> clearTaskQueue() {
  120.         threadPoolMonitor.clearTaskQueue();
  121.         return ResponseEntity.ok("success");
  122.     }
  123. }

配置如下：

  1. COPYserver:
  2.   port: 9091
  3. management:
  4.   server:
  5.     port: 9091
  6.   endpoints:
  7.     web:
  8.       exposure:
  9.         include: '*'
  10.       base-path: /management

prometheus的调度Job也可以适当调高频率，这里默认是15秒拉取一次/prometheus端点，也就是会每次提交3个收集周期的数据。项目启动之后，可以尝试调用/management/prometheus查看端点提交的数据：

9.Prometheus 线程池监控 - 图1j-m-t-p-1.png

因为ThreadPoolMonitorSample是我们自定义命名的Tag，看到相关字样说明数据收集是正常的。如果prometheus的Job没有配置错误，在本地的spring-boot项目起来后，可以查下prometheus的后台：

9.Prometheus 线程池监控 - 图2j-m-t-p-2.png

9.Prometheus 线程池监控 - 图3j-m-t-p-3.png

OK，完美，可以进行下一步。

grafana面板配置

确保JVM应用和prometheus的调度Job是正常的情况下，接下来重要的一步就是配置grafana面板。如果暂时不想认真学习一下prometheus的PSQL的话，可以从prometheus后台的/graph面板直接搜索对应的样本表达式拷贝进去grafana配置中就行，当然最好还是去看下prometheus的文档系统学习一下怎么编写PSQL。

  • 基本配置：

9.Prometheus 线程池监控 - 图4j-m-t-p-4.png

  • 可视化配置，把右边的标签勾选，宽度尽量调大点：

9.Prometheus 线程池监控 - 图5j-m-t-p-5.png

  • 查询配置，这个是最重要的，最终图表就是靠查询配置展示的：

9.Prometheus 线程池监控 - 图6j-m-t-p-6.png

查询配置具体如下：

  • A：thread_pool_active_size，Legend：-线程池活跃线程数。
  • B：thread_pool_largest_size，Legend：-线程池历史峰值线程数。
  • C：thread_pool_max_size，Legend：-线程池容量。
  • D：thread_pool_core_size，Legend：-线程池核心线程数。
  • E：thread_pool_thread_count，Legend：-线程池运行中的线程数。
  • F：thread_pool_queue_size，Legend：-线程池积压任务数。

最终效果

多调用几次例子中提供的几个接口，就能得到一个监控线程池呈现的图表：

9.Prometheus 线程池监控 - 图7j-m-t-p-7.png

小结

针对线程池ThreadPoolExecutor的各项数据进行监控，有利于及时发现使用线程池的接口的异常，如果想要快速恢复，最有效的途径是：清空线程池中任务队列中积压的任务。具体的做法是：可以把ThreadPoolExecutor委托到IOC容器管理，并且把ThreadPoolExecutor的任务队列清空的方法暴露成一个REST端点即可。像HTTP客户端的连接池如Apache-Http-Client或者OkHttp等的监控，可以用类似的方式实现，数据收集的时候可能由于加锁等原因会有少量的性能损耗，不过这些都是可以忽略的，如果真的怕有性能影响，可以尝试用反射API直接获取ThreadPoolExecutor实例内部的属性值，这样就可以避免加锁的性能损耗。

(本文完 c-2-d 20190414)