Java并发（六）线程池监控

目录

　　一、线程池监控参数

　　二、线程池监控类

　　三、注意事项

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

一、线程池监控参数

上一篇博文提到，线程池提供了以下几个方法可以监控线程池的使用情况：

方法	含义
getActiveCount()	线程池中正在执行任务的线程数量
getCompletedTaskCount()	线程池已完成的任务数量，该值小于等于taskCount
getCorePoolSize()	线程池的核心线程数量
getLargestPoolSize()	线程池曾经创建过的最大线程数量。通过这个数据可以知道线程池是否满过，也就是达到了maximumPoolSize
getMaximumPoolSize()	线程池的最大线程数量
getPoolSize()	线程池当前的线程数量
getTaskCount()	线程池已经执行的和未执行的任务总数

通过这些方法，可以对线程池进行监控，在 ThreadPoolExecutor 类中提供了几个空方法，如 beforeExecute 方法， afterExecute 方法和 terminated 方法，可以扩展这些方法在执行前或执行后增加一些新的操作，例如统计线程池的执行任务的时间等，可以继承自 ThreadPoolExecutor 来进行扩展。

二、线程池监控类

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Date;

import java.util.List;

import java.util.Objects;

import java.util.concurrent.*;

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


继承ThreadPoolExecutor类，覆盖了shutdown(), shutdownNow(), beforeExecute() 和 afterExecute()


方法来统计线程池的执行情况


<p>


Created by on 2019/4/19.

*/

public class ThreadPoolMonitor extends ThreadPoolExecutor {
private static final Logger LOGGER = LoggerFactory.getLogger(ThreadPoolMonitor.class);
/**

保存任务开始执行的时间，当任务结束时，用任务结束时间减去开始时间计算任务执行时间

*/

private ConcurrentHashMap<String, Date> startTimes;

/**

线程池名称，一般以业务名称命名，方便区分

*/

private String poolName;

/**

调用父类的构造方法，并初始化HashMap和线程池名称

@param corePoolSize    线程池核心线程数
@param maximumPoolSize 线程池最大线程数
@param keepAliveTime   线程的最大空闲时间
@param unit            空闲时间的单位
@param workQueue       保存被提交任务的队列
@param poolName        线程池名称

*/

public ThreadPoolMonitor(int corePoolSize, int maximumPoolSize, long keepAliveTime,

TimeUnit unit, BlockingQueue<Runnable> workQueue, String poolName) {

this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,

Executors.defaultThreadFactory(), poolName);

}

/**

调用父类的构造方法，并初始化HashMap和线程池名称

@param corePoolSize    线程池核心线程数
@param maximumPoolSize 线程池最大线程数
@param keepAliveTime   线程的最大空闲时间
@param unit            空闲时间的单位
@param workQueue       保存被提交任务的队列
@param threadFactory   线程工厂
@param poolName        线程池名称

*/

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: {}",

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: {}",

this.poolName, this.getCompletedTaskCount(), this.getActiveCount(), this.getQueue().size());

return super.shutdownNow();

}

/**

任务执行之前，记录任务开始时间

*/

@Override

protected void beforeExecute(Thread t, Runnable r) {

startTimes.put(String.valueOf(r.hashCode()), new Date());

}

/**

任务执行之后，计算任务结束时间

*/

@Override

protected void afterExecute(Runnable r, Throwable t) {

Date startDate = startTimes.remove(String.valueOf(r.hashCode()));

Date finishDate = new Date();

long diff = finishDate.getTime() - startDate.getTime();

// 统计任务耗时、初始线程数、核心线程数、正在执行的任务数量、

// 已完成任务数量、任务总数、队列里缓存的任务数量、池中存在的最大线程数、

// 最大允许的线程数、线程空闲时间、线程池是否关闭、线程池是否终止

LOGGER.info("{}-pool-monitor: " +

"Duration: {} ms, PoolSize: {}, CorePoolSize: {}, Active: {}, " +

"Completed: {}, Task: {}, Queue: {}, LargestPoolSize: {}, " +

"MaximumPoolSize: {},  KeepAliveTime: {}, isShutdown: {}, isTerminated: {}",

this.poolName,

diff, this.getPoolSize(), this.getCorePoolSize(), this.getActiveCount(),

this.getCompletedTaskCount(), this.getTaskCount(), this.getQueue().size(), this.getLargestPoolSize(),

this.getMaximumPoolSize(), this.getKeepAliveTime(TimeUnit.MILLISECONDS), this.isShutdown(), this.isTerminated());

}

/**

创建固定线程池，代码源于Executors.newFixedThreadPool方法，这里增加了poolName

@param nThreads 线程数量
@param poolName 线程池名称
@return ExecutorService对象

*/

public static ExecutorService newFixedThreadPool(int nThreads, String poolName) {

return new ThreadPoolMonitor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), poolName);

}

/**

创建缓存型线程池，代码源于Executors.newCachedThreadPool方法，这里增加了poolName

@param poolName 线程池名称
@return ExecutorService对象

*/

public static ExecutorService newCachedThreadPool(String poolName) {

return new ThreadPoolMonitor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<>(), poolName);

}

/**


生成线程池所用的线程，只是改写了线程池默认的线程工厂，传入线程池名称，便于问题追踪

*/

static class EventThreadFactory implements ThreadFactory {

private static final AtomicInteger poolNumber = new AtomicInteger(1);

private final ThreadGroup group;

private final AtomicInteger threadNumber = new AtomicInteger(1);

private final String namePrefix;
/**

初始化线程工厂

@param poolName 线程池名称

*/

EventThreadFactory(String poolName) {

SecurityManager s = System.getSecurityManager();

group = Objects.nonNull(s) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();

namePrefix = poolName + "-pool-" + poolNumber.getAndIncrement() + "-thread-";

}

@Override

public Thread newThread(Runnable r) {

Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);

if (t.isDaemon())

t.setDaemon(false);

if (t.getPriority() != Thread.NORM_PRIORITY)

t.setPriority(Thread.NORM_PRIORITY);

return t;

}

}

}