在我們實(shí)際項(xiàng)目開發(fā)中，常常會(huì)為不同的優(yōu)先級的任務(wù)設(shè)置相對應(yīng)的線程池。
一般我們只關(guān)注相關(guān)池的相關(guān)參數(shù)如核心線程數(shù)據(jù)，最大線程數(shù)據(jù)等等參數(shù)，容易忽略了對線程池中實(shí)際運(yùn)行情況的監(jiān)控。
綜上所述：線程池如果相當(dāng)于黑盒一樣在運(yùn)行的話，對系統(tǒng)的不利的。本文提供了一種簡單獲取線程池運(yùn)行狀態(tài)的方式，可以將詳情打印到日志或者對接到Prometheus上進(jìn)行展示。
詳細(xì)有不少博主給出了動(dòng)態(tài)修改線程的方式，但是由于生產(chǎn)環(huán)境是禁止，因此本文只提供了監(jiān)控的功能。
本代碼應(yīng)用項(xiàng)目架構(gòu)為springboot。

代碼

代碼類結(jié)構(gòu)

ThreadPoolMonitor:線程池?cái)U(kuò)展類
ThreadPoolUtil:線程池工具類
ThreadPoolDetailInfo：bean類
ExecutorThreadPoolManager:線程池實(shí)現(xiàn)類
ThreadPoolController:線程池測試方法

線程池?cái)U(kuò)展類

從類主要重寫了ThreadPoolExecutor類中的shutdown/shutdownNow/beforeExecute/afterExecute,用于對每個(gè)任務(wù)進(jìn)行執(zhí)行前后的攔截，計(jì)算出每個(gè)任務(wù)的運(yùn)行時(shí)間。

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;
/**
 * @ClassName ThreadPoolMonitor
 * @authors kantlin
 * @Date 2021/12/16 17:45
 **/
public class ThreadPoolMonitor extends ThreadPoolExecutor {
    private static final Logger LOGGER = LoggerFactory.getLogger(ThreadPoolMonitor.class);
    private final ConcurrentHashMap<String, Date> startTimes;
    private final String poolName;
    private long totalDiff;

    public ThreadPoolMonitor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, String poolName) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        this.startTimes = new ConcurrentHashMap();
        this.poolName = poolName;
    }

    @Override
    public void shutdown() {
        LOGGER.info("{} Going to shutdown. Executed tasks: {}, Running tasks: {}, Pending tasks: {}", new Object[]{this.poolName, this.getCompletedTaskCount(), this.getActiveCount(), this.getQueue().size()});
        super.shutdown();
    }
    @Override
    public List<Runnable> shutdownNow() {
        LOGGER.info("{} Going to immediately shutdown. Executed tasks: {}, Running tasks: {}, Pending tasks: {}", new Object[]{this.poolName, this.getCompletedTaskCount(), this.getActiveCount(), this.getQueue().size()});
        return super.shutdownNow();
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        this.startTimes.put(String.valueOf(r.hashCode()), new Date());
    }
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        Date startDate = this.startTimes.remove(String.valueOf(r.hashCode()));
        Date finishDate = new Date();
        long diff = finishDate.getTime() - startDate.getTime();
        this.totalDiff += diff;
    }

    public long getTotalDiff() {
        return this.totalDiff;
    }
}

線程工具類

import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName ThreadPoolUtil
 * @authors kantlin
 * @Date 2021/12/16 17:45
 **/

@Component
public class ThreadPoolUtil {
    private final HashMap<String, ThreadPoolMonitor> threadPoolExecutorHashMap = new HashMap();

    public ThreadPoolUtil() {
    }

    public ThreadPoolMonitor creatThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,String poolName) {
        ThreadPoolMonitor threadPoolExecutor = new ThreadPoolMonitor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,threadFactory, poolName);
        this.threadPoolExecutorHashMap.put(poolName, threadPoolExecutor);
        return threadPoolExecutor;
    }

    public HashMap<String, ThreadPoolMonitor> getThreadPoolExecutorHashMap() {
        return this.threadPoolExecutorHashMap;
    }

線程bean類

import lombok.Data;

@Data
public class ThreadPoolDetailInfo {
    //線程池名字
    private String threadPoolName;
    //當(dāng)前線程池大小
    private Integer poolSize;
    //線程池核心線程數(shù)量
    private Integer corePoolSize;
    //線程池生命周期中最大線程數(shù)量
    private Integer largestPoolSize;
    //線程池中允許的最大線程數(shù)
    private Integer maximumPoolSize;
    //線程池完成的任務(wù)數(shù)目
    private long completedTaskCount;
    //線程池中當(dāng)前活躍個(gè)數(shù)
    private Integer active;
    //線程池完成的任務(wù)個(gè)數(shù)
    private long task;
    //線程最大空閑時(shí)間
    private long keepAliveTime;
    //當(dāng)前活躍線程的占比
    private int activePercent;
    //任務(wù)隊(duì)列容量（阻塞隊(duì)列）
    private Integer queueCapacity;
    //當(dāng)前隊(duì)列中任務(wù)的數(shù)量
    private Integer queueSize;
    //線程池中任務(wù)平均執(zhí)行時(shí)長
    private long avgExecuteTime;

    public ThreadPoolDetailInfo(String threadPoolName, Integer poolSize, Integer corePoolSize, Integer largestPoolSize, Integer maximumPoolSize, long completedTaskCount, Integer active, long task, long keepAliveTime, int activePercent, Integer queueCapacity, Integer queueSize, long avgExecuteTime) {
        this.threadPoolName = threadPoolName;
        this.poolSize = poolSize;
        this.corePoolSize = corePoolSize;
        this.largestPoolSize = largestPoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.completedTaskCount = completedTaskCount;
        this.active = active;
        this.task = task;
        this.keepAliveTime = keepAliveTime;
        this.activePercent = activePercent;
        this.queueCapacity = queueCapacity;
        this.queueSize = queueSize;
        this.avgExecuteTime = avgExecuteTime;
    }
}

線程池實(shí)現(xiàn)類

在我的項(xiàng)目中，將線程池依次劃分為high、normal、low、single四種線程池類型。不同優(yōu)先級的任務(wù)將會(huì)被submit到不同的線程池中執(zhí)行。
在業(yè)務(wù)中有判斷線程池中queue的長度來決定是否投遞任務(wù)，由于沒有相應(yīng)的拒絕策略，所以隊(duì)列不設(shè)置長度。

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.*.newThread.ThreadPoolUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Component
public class ExecutorThreadPoolManager {

    @Autowired
    private ThreadPoolUtil threadPoolUtil;

    @Value("${thread_pool_normal_level_thread_max_num}")
    private Integer normalLevelThreadPoolThreadMaxNum;
    @Value("${thread_pool_normal_level_core_thread_num}")
    private Integer normalLevelThreadPoolCoreThreadNum;
    @Value("${thread_pool_low_level_thread_max_num}")
    private Integer lowLevelThreadPoolThreadMaxNum;
    @Value("${thread_pool_low_level_core_thread_num}")
    private Integer lowLevelThreadPoolCoreThreadNum;

    private ThreadPoolExecutor normalThreadPoolExecutor;

    private ThreadPoolExecutor highPriorityExecutor;

    private ThreadPoolExecutor lowPriorityExecutor;

    private ThreadPoolExecutor singleThreadPoolExecutor;


    @PostConstruct
    public void initExecutor() {
        ThreadFactory normalThreadFactory = new ThreadFactoryBuilder().setNameFormat("normal_task_thread_%d").build();
        normalThreadPoolExecutor = threadPoolUtil.creatThreadPool(normalLevelThreadPoolCoreThreadNum, normalLevelThreadPoolThreadMaxNum, 0L,
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), normalThreadFactory,"normal_level_thread_pool");

        ThreadFactory highPriorityThreadFactory = new ThreadFactoryBuilder().setNameFormat("high_priority_level_task_thread_%d").build();
        highPriorityExecutor = threadPoolUtil.creatThreadPool(normalLevelThreadPoolCoreThreadNum, normalLevelThreadPoolThreadMaxNum, 0L,
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), highPriorityThreadFactory,"high_level_thread_pool");

        ThreadFactory lowPriorityThreadFactory = new ThreadFactoryBuilder().setNameFormat("low_priority_level_task_thread_%d").build();
        lowPriorityExecutor = threadPoolUtil.creatThreadPool(lowLevelThreadPoolCoreThreadNum, lowLevelThreadPoolThreadMaxNum, 0L,
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), lowPriorityThreadFactory,"low_level_thread_pool");

        ThreadFactory singleFactory = new ThreadFactoryBuilder().setNameFormat("single_task_thread_%d").build();
        singleThreadPoolExecutor =threadPoolUtil.creatThreadPool(1, 1,
                0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), singleFactory,"single_level_thread_pool");
    }

    /**
     * @author kantlin
     * @date 2021/9/9
     * @describe 定義三種線程池, 一般采集類的用低優(yōu), 正常業(yè)務(wù)的用中優(yōu), 用戶手動(dòng)請求API的用高優(yōu)線程池
     **/
    public ThreadPoolExecutor getNormalThreadPoolExecutor() {
        return normalThreadPoolExecutor;
    }

    public ThreadPoolExecutor getHighPriorityExecutor() {
        return highPriorityExecutor;
    }

    public ThreadPoolExecutor getLowPriorityExecutor() {
        return lowPriorityExecutor;
    }

    public ThreadPoolExecutor getSingleThreadPoolExecutor() {
        return singleThreadPoolExecutor;
    }

}

線程池監(jiān)控接口類

import com.alibaba.fastjson.JSONObject;
import com.*.newThread.ThreadPoolDetailInfo;
import com.*.newThread.ThreadPoolMonitor;
import com.*.newThread.ThreadPoolUtil;
import com.*.thread.ExecutorThreadPoolManager;
import io.swagger.annotations.Api;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.web.bind.annotation.*;
import java.math.BigDecimal;
import java.text.NumberFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName ThreadPoolController
 * @authors kantlin
 * @Date 2021/12/17 14:53
 **/
@Api(description = "線程池監(jiān)控接口")
@RestController
@RequestMapping(value = "api/threadpool")
public class ThreadPoolController {
    private static final Logger LOGGER = LoggerFactory.getLogger(ThreadPoolController.class);

    @Autowired
    private ExecutorThreadPoolManager threadPool;

    @Autowired
    private ThreadPoolUtil threadPoolUtil;

    @GetMapping(value = "/getThreadPools")
    private List<String> getThreadPools() {
        List<String> threadPools = new ArrayList();
        if (!this.threadPoolUtil.getThreadPoolExecutorHashMap().isEmpty()) {
            Iterator var2 = this.threadPoolUtil.getThreadPoolExecutorHashMap().entrySet().iterator();

            while (var2.hasNext()) {
                Map.Entry<String, ThreadPoolMonitor> entry = (Map.Entry) var2.next();
                threadPools.add(entry.getKey());
            }
        }

        return threadPools;
    }

    @GetMapping(value = "/getThreadPoolListInfo")
    @Scheduled(cron = "${thread.poll.status.cron}")
    private List<ThreadPoolDetailInfo> getThreadPoolListInfo() {
        List<ThreadPoolDetailInfo> detailInfoList = new ArrayList();
        if (!this.threadPoolUtil.getThreadPoolExecutorHashMap().isEmpty()) {
            Iterator var2 = this.threadPoolUtil.getThreadPoolExecutorHashMap().entrySet().iterator();
            while (var2.hasNext()) {
                Map.Entry<String, ThreadPoolMonitor> entry = (Map.Entry) var2.next();
                ThreadPoolDetailInfo threadPoolDetailInfo = this.threadPoolInfo(entry.getValue(), (String) entry.getKey());
                detailInfoList.add(threadPoolDetailInfo);
            }
        }
        LOGGER.info("Execute details of cuurent thread poll:{}", JSONObject.toJSONString(detailInfoList));
        return detailInfoList;
    }

    private ThreadPoolDetailInfo threadPoolInfo(ThreadPoolMonitor threadPool, String threadPoolName) {
        BigDecimal activeCount = new BigDecimal(threadPool.getActiveCount());
        BigDecimal maximumPoolSize = new BigDecimal(threadPool.getMaximumPoolSize());
        BigDecimal result = activeCount.divide(maximumPoolSize, 2, 4);
        NumberFormat numberFormat = NumberFormat.getPercentInstance();
        numberFormat.setMaximumFractionDigits(2);
        int queueCapacity = 0;
        return new ThreadPoolDetailInfo(threadPoolName, threadPool.getPoolSize(), threadPool.getCorePoolSize(), threadPool.getLargestPoolSize(), threadPool.getMaximumPoolSize(), threadPool.getCompletedTaskCount(), threadPool.getActiveCount(), threadPool.getTaskCount(), threadPool.getKeepAliveTime(TimeUnit.MILLISECONDS), new Double(result.doubleValue() * 100).intValue(), queueCapacity, threadPool.getQueue().size(), threadPool.getTaskCount() == 0L ? 0L : threadPool.getTotalDiff() / threadPool.getTaskCount());
    }
}

運(yùn)行結(jié)果

上面controller中的方法除了可以通過接口進(jìn)行暴露外，還設(shè)置了定時(shí)任務(wù)定期的打印到日志中。方便對系統(tǒng)狀態(tài)進(jìn)行排查。

[
  {
    "active": 0,
    "activePercent": 0,
    "avgExecuteTime": 0,
    "completedTaskCount": 0,
    "corePoolSize": 20,
    "keepAliveTime": 0,
    "largestPoolSize": 0,
    "maximumPoolSize": 20,
    "poolSize": 0,
    "queueCapacity": 0,
    "queueSize": 0,
    "task": 0,
    "threadPoolName": "high_level_thread_pool"
  },
  {
    "active": 0,
    "activePercent": 0,
    "avgExecuteTime": 0,
    "completedTaskCount": 0,
    "corePoolSize": 33,
    "keepAliveTime": 0,
    "largestPoolSize": 0,
    "maximumPoolSize": 33,
    "poolSize": 0,
    "queueCapacity": 0,
    "queueSize": 0,
    "task": 0,
    "threadPoolName": "low_level_thread_pool"
  },
  {
    "active": 0,
    "activePercent": 0,
    "avgExecuteTime": 371,
    "completedTaskCount": 14,
    "corePoolSize": 20,
    "keepAliveTime": 0,
    "largestPoolSize": 14,
    "maximumPoolSize": 20,
    "poolSize": 14,
    "queueCapacity": 0,
    "queueSize": 0,
    "task": 14,
    "threadPoolName": "normal_level_thread_pool"
  },
  {
    "active": 0,
    "activePercent": 0,
    "avgExecuteTime": 0,
    "completedTaskCount": 0,
    "corePoolSize": 1,
    "keepAliveTime": 0,
    "largestPoolSize": 0,
    "maximumPoolSize": 1,
    "poolSize": 0,
    "queueCapacity": 0,
    "queueSize": 0,
    "task": 0,
    "threadPoolName": "single_level_thread_pool"
  }
]

到此這篇關(guān)于springboot線程池監(jiān)控的簡單實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)springboot線程池監(jiān)控內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: