Java线程池监控预警实现

前言

在Java线程池的使用中,如何动态可视化监控线程池各项运行指标是一个比较重要的需求。

线程池监控必要性

如果你有以下需求,那么你可以尝试监控线程池。

• 为动态调优提供参考
• 规避风险
• 帮助定位问题以及运维

在日常开发中,当我们使用到线程池这一技术时,一般会选择动态配置参数的方式,或者是写死参数,并且提供修改参数的api进行调整。所以我们需要监控线程池的各项指标,结合性能分析,来进行调优决策。

除此之外,有一些场景会发生线程池相关问题,进一步引起rpc服务熔断降级,甚至不可用的情况。我们也可以通过配置线程池监控预警来进行人工介入,或者是通过一些规则自动调整,以规避风险发生。或者是已经发生的问题,我们通过监控辅助分析定位问题产生原因。

我在工作中就曾经遇到过各种各样的线程池问题,比较好玩的就比如死锁问题。
比方说线程之间有依赖关系的场景,当核心线程池大小设置小于单次请求会创建的总线程数时,最后需要创建的线程会进入队列,并且在队列满之前永远都没办法获取线程,形成死锁,然后触发rpc超时,引起服务熔断降级等。后来定位问题后发现在一些书中也有描述同类问题,参考<Java多线程编程实战指南>125页死锁。

ThreadPoolExecutor提供的监控api

• int getCorePoolSize():核心线程数。
• int getLargestPoolSize():历史峰值线程数。
• int getMaximumPoolSize():最大线程数(线程池线程容量)。
• int getActiveCount():当前活跃线程数,Tag:thread.pool.active.size。
• int getPoolSize():当前线程池中运行的线程总数
• 当前任务队列中积压任务的总数,getQueue.size();

除此之外我们还需要补充线程池名称,可以通过创建线程池时定义。

实现监控的整体思路

无论是基于开源的监控体系,或者闭源的体系,我个人的思路大体类似。
大体可以分为以下步骤:

• 收拢线程池的创建入口,统一管理线程池对象
• 开启Schedule线程定时访问对象池,基于线程池对象api进行打点。
• 对接监控平台,采集第2步埋的keys,配置规则进行展示。
• 配置阀值预警

基于大家的使用方式不同,每一步可能各有差异,可自由修改。
比如第一步也可以用Spring管理以及查找。第二步也可以不用metrics api,用log解析的方式。

开源解决方案

在开源产品中,我们可以选择基于Grafana、Prometheus、Influxdb、MicroMeter等实现。

介绍比较常见的两种:

• 基于Grafana+Influxdb实现。
• 基于Grafana+Prometheus+MicroMeter实现。

首先介绍一下。

• grafana

golang开源的时序数据展示平台,主要用于大规模指标数据的可视化。

• influxdb

golang开源的时序数据库,阿里云还有二次封装的框架。

• micrometer

metrics框架、类似于alimetrics、ali360、eagle eye metric等。

• prometheus

Prometheus是一套开源的监控&报警&时间序列数据库的组合,特点在于端口采集,适配能力强,可以通过HTTP协议去采集数据,大部分框架都接入了。

线程池创建部分

使用ThreadPoolBuilder创建线程池

influxdb&grafana实现

macos安装influxdb直接使用homebrew。

brew update
brew install influxdb

如果没有安装brew,先安装brew

/usr/bin/ruby -e "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/master/install)"

然后启动Influxdb

brew services start influxdb

或者用launchctl,或者进入bin目录敲一下influxd(brew安装的在/usr/local/Cellar底下)

#启动服务
launchctl load ~/Library/LaunchAgents/homebrew.mxcl.influxdb.plist
#停止服务
launchctl unload ~/Library/LaunchAgents/homebrew.mxcl.influxdb.plist

启动客户端

influx -precision rfc3339

然后我们开始写代码
application.properties

# 数据库地址、用户名、密码
influxDB.serverAddr=http://127.0.0.1:8086
influxDB.username=root
influxDB.password=root
# 库名
influxDB.dataBase=thread_pool_metrics
# 数据保留策略名
influxDB.retentionPolicyName=3_days
# 数据保留策略,3d:保留3天
influxDB.retentionPolicy=3d
server.port=8080

@Configuration
public class InfluxdbConfig {
 @Value("${influxDB.serverAddr}")
 private String serverAddr;
 @Value("${influxDB.username}")
 private String username;
 @Value("${influxDB.password}")
 private String password;
 @Value("${influxDB.dataBase}")
 private String dataBase;
 @Value("${influxDB.retentionPolicyName}")
 private String retentionPolicyName;
 @Value("${influxDB.retentionPolicy}")
 private String retentionPolicy;
 @Bean
 public InfluxDB influxDB() {
 // 连接influxDB数据库
 InfluxDB influxDB = InfluxDBFactory.connect(serverAddr, username, password);
 // 创建数据库
 influxDB.query(new Query("CREATE DATABASE " + dataBase));
 influxDB.setDatabase(dataBase);
 // 创建数据保留策略
 influxDB.query(new Query("CREATE RETENTION POLICY \"" + retentionPolicyName + "\" ON \"" + dataBase
 + "\" DURATION " + retentionPolicy + " REPLICATION 1 DEFAULT"));
 influxDB.setRetentionPolicy(retentionPolicyName);
 return influxDB;
 }
}

@Component
public class ThreadPoolMonitorProcessor {
 private static final Logger logger = LoggerFactory.getLogger(ThreadPoolMonitorProcessor.class);
 private String host;
 @Value("${server.port}")
 private int port;
 @Autowired
 private InfluxDB influxDB;
 @PostConstruct
 public void init() {
 try {
 host = InetAddress.getLocalHost().getHostAddress();
 } catch (Exception e) {
 logger.error("get host&port error", e);
 }
 }
 @Scheduled(fixedRate = 1000)
 public void threadPoolMetricsRecord() {
 // 线程池指标数据
 ThreadPoolRegistrar.threadPools().forEach(threadPoolMonitorConsumer());
 }
 private BiConsumer<String, ThreadPoolExecutor> threadPoolMonitorConsumer() {
 return (name, threadPool) -> {
 // 当前活跃线程数
 int currentActive = threadPool.getActiveCount();
 // 当前线程池大小
 int currentPoolSize = threadPool.getPoolSize();
 // 最大线程数
 long maximumPoolSize = threadPool.getMaximumPoolSize();
 // 核心线程数
 long corePoolSize = threadPool.getCorePoolSize();
 // 队列容量
 int queueSize = threadPool.getQueue().size();
 /**
 * measurement相当于表名
 * tag是用于统计或分类的参数,这里用ip+端口进行标记,用于后续图表展示
 * field相当于列,存储各类指标值
 */
 influxDB.write(Point.measurement("threadPool")
 .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
 .tag("host", host)
 .tag("port", port + "")
 .tag("addr", host + ":" + port)
 .tag("group", name)
 .addField("currentPoolSize", currentPoolSize)
 .addField("currentActive", currentActive)
 .addField("corePoolSize", corePoolSize)
 .addField("maximumPoolSize", maximumPoolSize)
 .addField("queneSize", queueSize)
 .build());
 };
 }
}

至此完成了打点,然后我们看一下ThreadPoolBuilder创建线程池时调用的register。

@Component
public class ThreadPoolRegistrar implements ApplicationRunner {
 private static final Logger LOGGER = LoggerFactory.getLogger(ThreadPoolRegistrar.class);
 private static Map<String, ThreadPoolExecutor> CONTAINER = Maps.newConcurrentMap();
 public static void register(String threadPoolName, ThreadPoolExecutor executorService) {
 Preconditions.checkNotNull(executorService);
 CONTAINER.put(StringUtils.defaultString(threadPoolName, UUID.randomUUID().toString()), executorService);
 }
 public static ThreadPoolExecutor getInstance(String threadPoolName) {
 return CONTAINER.get(threadPoolName);
 }
 public static Map<String, ThreadPoolExecutor> threadPools(){
 return CONTAINER;
 }
 @Override
 public void run(ApplicationArguments applicationArguments) {
 Runtime.getRuntime().addShutdownHook(new Thread(() -> {
 for (ExecutorService executorService : CONTAINER.values()) {
 try {
 ThreadPoolUtils.gracefulShutdown(executorService, 10, TimeUnit.SECONDS);
 } catch (Exception ex) {
 LOGGER.error(ex.getMessage());
 }
 }
 }));
 }
}

brew安装influxdb以及grafana。

brew update
brew install grafana
==> Downloading https://homebrew.bintray.com/bottles/grafana-4.6.3.high_sierra.bottl
######################################################################## 100.0%
==> Pouring grafana-4.6.3.high_sierra.bottle.tar.gz
==> Caveats
To have launchd start grafana now and restart at login:
 brew services start grafana
Or, if you don't want/need a background service you can just run:
 grafana-server --config=/usr/local/etc/grafana/grafana.ini --homepath /usr/local/share/grafana cfg:default.paths.logs=/usr/local/var/log/grafana cfg:default.paths.data=/usr/local/var/lib/grafana cfg:default.paths.plugins=/usr/local/var/lib/grafana/plugins
==> Summary
???? /usr/local/Cellar/grafana/4.6.3: 2,874 files, 127.2MB

brew tap homebrew/services
brew services start grafana
==> Successfully started `grafana` (label: homebrew.mxcl.grafana)

然后chrome输入http://localhost:3000/
用户名和密码都是admin
进入grafana之后选择influxdb数据源,之后进行指标集配置,即可看到监控界面。

基于micrometer+prometheus+grafana实现

• Metrics.gauge埋点

Metrics.gauge("thread.pool.core.size", TAG, executor, ThreadPoolExecutor::getCorePoolSize);
 Metrics.gauge("thread.pool.largest.size", TAG, executor, ThreadPoolExecutor::getLargestPoolSize);
 Metrics.gauge("thread.pool.max.size", TAG, executor, ThreadPoolExecutor::getMaximumPoolSize);
 Metrics.gauge("thread.pool.active.size", TAG, executor, ThreadPoolExecutor::getActiveCount);
 Metrics.gauge("thread.pool.thread.count", TAG, executor, ThreadPoolExecutor::getPoolSize);
 // 注意如果阻塞队列使用无界队列这里不能直接取size
 Metrics.gauge("thread.pool.queue.size", TAG, executor, e -> e.getQueue().size());

• 配置prometheus采点
• grafana配置源和指标集。

我在网上找到了一篇手把手具体实现的博客,以供参考通过micrometer实时监控线程池的各项指标

本文来自:简书

感谢作者:但时间也偷换概念

查看原文:Java线程池监控预警实现

• 编辑
• 预览