前言
在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实现。
首先介绍一下。
golang开源的时序数据展示平台，主要用于大规模指标数据的可视化。
golang开源的时序数据库，阿里云还有二次封装的框架。
metrics框架、类似于alimetrics、ali360、eagle eye metric等。
Prometheus是一套开源的监控&报警&时间序列数据库的组合，特点在于端口采集，适配能力强，可以通过HTTP协议去采集数据，大部分框架都接入了。
线程池创建部分
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实时监控线程池的各项指标
有疑问加站长微信联系（非本文作者）
