通過micrometer實(shí)時(shí)監(jiān)控線程池的各項(xiàng)指標(biāo)

前提

最近的一個(gè)項(xiàng)目中涉及到文件上傳和下載，使用到JUC的線程池ThreadPoolExecutor，在生產(chǎn)環(huán)境中出現(xiàn)了某些時(shí)刻線程池滿負(fù)載運(yùn)作，由于使用了CallerRunsPolicy拒絕策略，導(dǎo)致滿負(fù)載情況下，應(yīng)用接口調(diào)用無法響應(yīng)，處于假死狀態(tài)?？紤]到之前用micrometer + prometheus + grafana搭建過監(jiān)控體系，于是考慮使用micrometer做一次主動(dòng)的線程池度量數(shù)據(jù)采集，最終可以相對實(shí)時(shí)地展示在grafana的面板中。

實(shí)踐過程

下面通過真正的實(shí)戰(zhàn)過程做一個(gè)仿真的例子用于復(fù)盤。

代碼改造

首先我們要整理一下ThreadPoolExecutor中提供的度量數(shù)據(jù)項(xiàng)和micrometer對應(yīng)的Tag的映射關(guān)系：

• 線程池名稱，Tag：thread.pool.name，這個(gè)很重要，用于區(qū)分各個(gè)線程池的數(shù)據(jù)，如果使用IOC容器管理，可以使用BeanName代替。
• int getCorePoolSize()：核心線程數(shù)，Tag：thread.pool.core.size。
• int getLargestPoolSize()：歷史峰值線程數(shù)，Tag：thread.pool.largest.size。
• int getMaximumPoolSize()：最大線程數(shù)(線程池線程容量)，Tag：thread.pool.max.size。
• int getActiveCount()：當(dāng)前活躍線程數(shù)，Tag：thread.pool.active.size。
• int getPoolSize()：當(dāng)前線程池中運(yùn)行的線程總數(shù)(包括核心線程和非核心線程)，Tag：thread.pool.thread.count。
• 當(dāng)前任務(wù)隊(duì)列中積壓任務(wù)的總數(shù)，Tag：thread.pool.queue.size，這個(gè)需要?jiǎng)討B(tài)計(jì)算得出。

接著編寫具體的代碼，實(shí)現(xiàn)的功能如下：

• 1、建立一個(gè)ThreadPoolExecutor實(shí)例，核心線程和最大線程數(shù)為10，任務(wù)隊(duì)列長度為10，拒絕策略為AbortPolicy。
• 2、提供兩個(gè)方法，分別使用線程池實(shí)例模擬短時(shí)間耗時(shí)的任務(wù)和長時(shí)間耗時(shí)的任務(wù)。
• 3、提供一個(gè)方法用于清空線程池實(shí)例中的任務(wù)隊(duì)列。
• 4、提供一個(gè)單線程的調(diào)度線程池用于定時(shí)收集ThreadPoolExecutor實(shí)例中上面列出的度量項(xiàng)，保存到micrometer內(nèi)存態(tài)的收集器中。

由于這些統(tǒng)計(jì)的值都會(huì)跟隨時(shí)間發(fā)生波動(dòng)性變更，可以考慮選用Gauge類型的Meter進(jìn)行記錄。

// 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 = () -> {
        //這里需要捕獲異常,盡管實(shí)際上不會(huì)產(chǎn)生異常,但是必須預(yù)防異常導(dǎo)致調(diào)度線程池線程失效的問題
        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);
            // 注意如果阻塞隊(duì)列使用無界隊(duì)列這里不能直接取size
            Metrics.gauge("thread.pool.queue.size", TAG, executor, e -> e.getQueue().size());
        } catch (Exception e) {
            //ignore
        }
    };

    @Override
    public void afterPropertiesSet() throws Exception {
        // 每5秒執(zhí)行一次
        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的調(diào)度Job也可以適當(dāng)調(diào)高頻率，這里默認(rèn)是15秒拉取一次/prometheus端點(diǎn)，也就是會(huì)每次提交3個(gè)收集周期的數(shù)據(jù)。項(xiàng)目啟動(dòng)之后，可以嘗試調(diào)用/management/prometheus查看端點(diǎn)提交的數(shù)據(jù)：

[圖片上傳失敗...(image-6ca6f9-1555258570421)]

因?yàn)?code>ThreadPoolMonitorSample是我們自定義命名的Tag，看到相關(guān)字樣說明數(shù)據(jù)收集是正常的。如果prometheus的Job沒有配置錯(cuò)誤，在本地的spring-boot項(xiàng)目起來后，可以查下prometheus的后臺(tái)：

[圖片上傳失敗...(image-415de3-1555258570421)]

[圖片上傳失敗...(image-71bc5a-1555258570421)]

OK，完美，可以進(jìn)行下一步。

grafana面板配置

確保JVM應(yīng)用和prometheus的調(diào)度Job是正常的情況下，接下來重要的一步就是配置grafana面板。如果暫時(shí)不想認(rèn)真學(xué)習(xí)一下prometheus的PSQL的話，可以從prometheus后臺(tái)的/graph面板直接搜索對應(yīng)的樣本表達(dá)式拷貝進(jìn)去grafana配置中就行，當(dāng)然最好還是去看下prometheus的文檔系統(tǒng)學(xué)習(xí)一下怎么編寫PSQL。

• 基本配置：

[圖片上傳失敗...(image-3f92be-1555258570421)]

• 可視化配置，把右邊的標(biāo)簽勾選，寬度盡量調(diào)大點(diǎn)：

[圖片上傳失敗...(image-3711f2-1555258570421)]

• 查詢配置，這個(gè)是最重要的，最終圖表就是靠查詢配置展示的：

[圖片上傳失敗...(image-6d3070-1555258570421)]

查詢配置具體如下：

• A：thread_pool_active_size，Legend：{{instance}}-{{thread_pool_name}}線程池活躍線程數(shù)。
• B：thread_pool_largest_size，Legend：{{instance}}-{{thread_pool_name}}線程池歷史峰值線程數(shù)。
• C：thread_pool_max_size，Legend：{{instance}}-{{thread_pool_name}}線程池容量。
• D：thread_pool_core_size，Legend：{{instance}}-{{thread_pool_name}}線程池核心線程數(shù)。
• E：thread_pool_thread_count，Legend：{{instance}}-{{thread_pool_name}}線程池運(yùn)行中的線程數(shù)。
• F：thread_pool_queue_size，Legend：{{instance}}-{{thread_pool_name}}線程池積壓任務(wù)數(shù)。

最終效果

多調(diào)用幾次例子中提供的幾個(gè)接口，就能得到一個(gè)監(jiān)控線程池呈現(xiàn)的圖表：

[圖片上傳失敗...(image-c2224-1555258570421)]

小結(jié)

針對線程池ThreadPoolExecutor的各項(xiàng)數(shù)據(jù)進(jìn)行監(jiān)控，有利于及時(shí)發(fā)現(xiàn)使用線程池的接口的異常，如果想要快速恢復(fù)，最有效的途徑是：清空線程池中任務(wù)隊(duì)列中積壓的任務(wù)。具體的做法是：可以把ThreadPoolExecutor委托到IOC容器管理，并且把ThreadPoolExecutor的任務(wù)隊(duì)列清空的方法暴露成一個(gè)REST端點(diǎn)即可。像HTTP客戶端的連接池如Apache-Http-Client或者OkHttp等的監(jiān)控，可以用類似的方式實(shí)現(xiàn)，數(shù)據(jù)收集的時(shí)候可能由于加鎖等原因會(huì)有少量的性能損耗，不過這些都是可以忽略的，如果真的怕有性能影響，可以嘗試用反射API直接獲取ThreadPoolExecutor實(shí)例內(nèi)部的屬性值，這樣就可以避免加鎖的性能損耗。

(本文完 c-2-d 20190414)