转载

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
:beer:  /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());