通过micrometer实时监控线程池的各项指标

通过micrometer实时监控线程池的各项指标

前提

最近的一个项目中涉及到文件上传和下载，使用到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进行记录。

// ThreadPoolMonitor
import io.micrometer.core.instrument.Metrics;
import io.micrometer.core.instrument.Tag;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import java.util.Collections;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author throwable
 * @version v1.0
 * @description
 * @since 2019/4/7 21:02
 */
@Service
public class ThreadPoolMonitor implements InitializingBean {

    private static final String EXECUTOR_NAME = "ThreadPoolMonitorSample";
    private static final Iterable<Tag> TAG = Collections.singletonList(Tag.of("thread.pool.name", EXECUTOR_NAME));
    private final ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor();

    private final ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10, 0, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(10), new ThreadFactory() {

        private final AtomicInteger counter = new AtomicInteger();

        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setDaemon(true);
            thread.setName("thread-pool-" + counter.getAndIncrement());
            return thread;
        }
    }, new ThreadPoolExecutor.AbortPolicy());

    private Runnable monitor = () -> {
        //这里需要捕获异常,尽管实际上不会产生异常,但是必须预防异常导致调度线程池线程失效的问题
        try {
            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());
        } catch (Exception e) {
            //ignore
        }
    };

    @Override
    public void afterPropertiesSet() throws Exception {
        // 每5秒执行一次
        scheduledExecutor.scheduleWithFixedDelay(monitor, 0, 5, TimeUnit.SECONDS);
    }

    public void shortTimeWork() {
        executor.execute(() -> {
            try {
                // 5秒
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                //ignore
            }
        });
    }

    public void longTimeWork() {
        executor.execute(() -> {
            try {
                // 500秒
                Thread.sleep(5000 * 100);
            } catch (InterruptedException e) {
                //ignore
            }
        });
    }

    public void clearTaskQueue() {
        executor.getQueue().clear();
    }
}

//ThreadPoolMonitorController
import club.throwable.smp.service.ThreadPoolMonitor;
import lombok.RequiredArgsConstructor;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author throwable
 * @version v1.0
 * @description
 * @since 2019/4/7 21:20
 */
@RequiredArgsConstructor
@RestController
public class ThreadPoolMonitorController {

    private final ThreadPoolMonitor threadPoolMonitor;

    @GetMapping(value = "/shortTimeWork")
    public ResponseEntity<String> shortTimeWork() {
        threadPoolMonitor.shortTimeWork();
        return ResponseEntity.ok("success");
    }

    @GetMapping(value = "/longTimeWork")
    public ResponseEntity<String> longTimeWork() {
        threadPoolMonitor.longTimeWork();
        return ResponseEntity.ok("success");
    }

    @GetMapping(value = "/clearTaskQueue")
    public ResponseEntity<String> clearTaskQueue() {
        threadPoolMonitor.clearTaskQueue();
        return ResponseEntity.ok("success");
    }
}

配置如下：

server:
  port: 9091
management:
  server:
    port: 9091
  endpoints:
    web:
      exposure:
        include: '*'
      base-path: /management

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

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

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

grafana面板配置

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

  • 基本配置：

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

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

查询配置具体如下：

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

最终效果

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

小结

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

个人博客原文链接：http://www.throwable.club/2019/04/14/jvm-micrometer-thread-pool-monitor

(本文完 c-2-d 20190414)

原文地址：https://www.cnblogs.com/throwable/p/10708351.html

时间： 2024-11-08 19:28:07

通过micrometer实时监控线程池的各项指标的相关文章

SpringBoot-技术专区-实战方案-应用监控线程池

背景 废话不多说,做这个监控的背景很简单,我们的项目都是以spring boot框架为基础开发的,代码里所有的异步线程都是通过@Async标签标注的,并且标注的时候都是指定对应线程池的,如果不知@Async标注的,可以参考@Async异步线程池用法总结, 如果你用的不是spring,就参考上文提到的公众号文章就好.再回到背景,我们当时经常遇到的问题就是这些线程池的队列满了之后,新的异步任务无法添加进去的错误,因此我们想对所有这种类型的线程池进行监控. 监控方式 再来介绍一下我们最终采用的方式 —

perl 多线程，实时监控线程数，支持max thread

#!/usr/bin/perl -w #Description:rerun eod job group by system #Auther:Suzm #Date  :2015-06-23 use DBI; use Thread; use strict; push( @INC, $ENV{TPMS_EOD_PERL_LIB} ); require public_pg; my %dbc_info; my $maxnum = 3; my %threads; my $tx_date; my $path_

教你如何监控 Java 线程池运行状态

之前写过一篇 Java 线程池的使用介绍文章<线程池全面解析>,全面介绍了什么是线程池.线程池核心类.线程池工作流程.线程池分类.拒绝策略.及如何提交与关闭线程池等. 但在实际开发过程中,在线程池使用过程中可能会遇到各方面的故障,如线程池阻塞,无法提交新任务等. 如果你想监控某一个线程池的执行状态,线程池执行类 ThreadPoolExecutor 也给出了相关的 API, 能实时获取线程池的当前活动线程数.正在排队中的线程数.已经执行完成的线程数.总线程数等. 总线程数 = 排队线程数 +

java线程池监控

原因 最近在完善公司的基础发布平台的时候,使用到了一线程去做一些异步的事情,在开发环境和测试环境验证没有任何问题,但是在程序在生产运行一段时间后,发现没有得到自己想要的结果,为此开始了漫长的排查bug的之路,因为用到了一些线程,但是实际又没有对这些线程足够的监控,所以在排查问题的时候也是历经艰难险阻: 原始代码 protected ScheduledExecutorService executorService = Executors.newScheduledThreadPool(2); /**

Java并发（四）线程池监控

目录 一.线程池监控参数 二.线程池监控类 三.注意事项 在上一篇博文中,我们介绍了线程池的基本原理和使用方法.了解了基本概念之后,我们可以使用 Executors 类创建线程池来执行大量的任务,使用线程池的并发特性提高系统的吞吐量.但是,线程池使用不当也会使服务器资源枯竭,导致异常情况的发生,比如固定线程池的阻塞队列任务数量过多.缓存线程池创建的线程过多导致内存溢出.系统假死等问题.因此,我们需要一种简单的监控方案来监控线程池的使用情况,比如完成任务数量.未完成任务数量.线程大小等信息. 一.

线程池ThreadPoolExecutor、Executors参数详解与源代码分析

欢迎探讨,如有错误敬请指正 如需转载,请注明出处 http://www.cnblogs.com/nullzx/ 1. ThreadPoolExecutor数据成员 Private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING,0)); ctl主要用于存储线程池的工作状态以及池中正在运行的线程数.显然要在一个整型变量存储两个数据,只能将其一分为二.其中高3bit用于存储线程池的状态,低位的29bit用于存储正在运行的线程数. 线

JAVA线程池的分析和使用

http://www.infoq.com/cn/articles/java-threadPool/ 1. 引言 合理利用线程池能够带来三个好处.第一:降低资源消耗.通过重复利用已创建的线程降低线程创建和销毁造成的消耗.第二:提高响应速度.当任务到达时,任务可以不需要等到线程创建就能立即执行.第三:提高线程的可管理性.线程是稀缺资源,如果无限制的创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一的分配,调优和监控.但是要做到合理的利用线程池,必须对其原理了如指掌. 2. 线程池

Java线程池ThreadPoolExecutor

线程池的好处 1. 降低资源的消耗 通过重复利用已创建的线程降低线程创建和销毁所造成的消耗 2. 提高响应速度 当任务到达时,任务可以不需要等到线程创建就能立即执行 3. 提高线程的可管理型 线程是稀缺资源,如果无限制地创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一分配.调优和监控. 实现原理 当提交一个新任务到线程池时,线程池的处理流程为: 1). 线程池判断核心线程池里的线程是否都在执行任务. 如果不是,则创建一个新的工作线程来执行任务.如果核心线程池里的线程都在执行

再聊线程池

引言 最近恰好在组内分享线程池,又看了看四年前自己写的线程池文章,一是感叹时光荏苒,二是感叹当时的理解太浅薄了,三是感叹自己这么多年依然停留在浅薄的理解当中,没有探究其实现,羞愧难当.遂把分享的内容整理出来,希望能够让读者对线程池有一个全新的认识. 池化 这里池化并不是深度学习中的池化,而是将资源交给池来管理的这一过程.我们在开发中经常回接触到池化资源的技术,最常见的当然是数据库连接池,以及我们今天要讲的线程池,那这种池化资源的特点和好处是什么呢? 特点 通常管理昂贵的资源,如连接.线程等 资源