/**
 * 自定義可監(jiān)控的線程池
 */
 public class MonitoredThreadPoolStatisticsExecutor extends ThreadPoolExecutor implements DisposableBean {
 /**
 * 線程池的名稱
 */
 private String poolName;
 /**
 * 預(yù)設(shè)的任務(wù)超時時間閾值，用于統(tǒng)計功能。     * 以任務(wù)提交時間進行計時，單位 ms，大于0則記錄超時次數(shù)。
 */
 private long timeout = 120000l;
 /**
 * 是否記錄任務(wù)超時次數(shù)
 */
 private boolean taskTimeoutFlag = false;
 /**
 * 任務(wù)執(zhí)行超時時間閾值，用于統(tǒng)計功能。     * 以任務(wù)開始執(zhí)行進行計時，單位 ms,大于 0 則記錄任務(wù)執(zhí)行超時次數(shù)。
 */
 private long execTimeout = 120000l;
 /**
 * 是否記錄任務(wù)執(zhí)行超時次數(shù)
 */
 private boolean taskExecTimeoutFlag = false;
 /**
 * 任務(wù)在隊列中等待的時間閾值，用于統(tǒng)計功能。     * 以任務(wù)提交時間開始計時到開始執(zhí)行為止，單位 ms。
 */
 private long waitInQueueTimeout = 60000l;
 /**
 * 是否記錄任務(wù)等待時間超時次數(shù)
 */
 private boolean taskWaitInQueueTimeoutFlag = false;
 /**
 * 任務(wù)隊列使用率告警閾值
 */
 private int queueSizeWarningPercent = 80;
 /**
 * 是否進行隊列容量告警
 */
 private boolean queueSizeWarningFlag = false;
 /**
 * 是否需要隊列容量告警(隊列是否曾經(jīng)達到過預(yù)警值)
 */
 private AtomicBoolean queueSizeHasWarningFlag = new AtomicBoolean(false);
 /**
 * 任務(wù)總時長，用于統(tǒng)計功能。以任務(wù)提交時間進行計時，單位 ms
 */
 private AtomicLong taskTotalTime = new AtomicLong(0);
 /**
 * 任務(wù)總執(zhí)行時長，用于統(tǒng)計功能。以任務(wù)開始執(zhí)行進行計時，單位 ms
 */
 private AtomicLong taskTotalExecTime = new AtomicLong(0);
 /**
 * 最短任務(wù)時長，以提交時間計時，單位 ms
 */
 private long minTaskTime = Long.MAX_VALUE;
 /**
 * 最長任務(wù)時長，以提交時間計時，單位 ms
 */
 private long maxTaskTime = 0;
 /**
 * 任務(wù)超時次數(shù)，以任務(wù)提交進行計時
 */
 private AtomicLong taskTimeoutCount = new AtomicLong(0);
 /**
 * 任務(wù)執(zhí)行超時次數(shù)，以任務(wù)開始執(zhí)行時間進行計時
 */
 private AtomicLong taskExecTimeoutCount = new AtomicLong(0);
 /**
 * 任務(wù)等待時間超過設(shè)定的閾值的次數(shù)
 */
 private AtomicLong taskWaitInQueueTimeoutCount = new AtomicLong(0);
 /**
 * 最短任務(wù)時長，以執(zhí)行時間計時，單位 ms
 */
 private long minTaskExecTime = Long.MAX_VALUE;
 /**
 * 最長任務(wù)時長，以執(zhí)行時間計時，單位 ms
 */
 private long maxTaskExecTime = 0;
 /**
 * 保存任務(wù)信息
 */
 private Map<String, TaskStatistics> taskInfoMap = new ConcurrentHashMap<String, TaskStatistics>();
 public MonitoredThreadPoolStatisticsExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler, String poolName, long timeout, long execTimeout, long waitInQueueTimeout, int queueSizeWarningPercent) {
 super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, new NamedThreadFactory(poolName), handler);
 this.poolName = poolName;
 this.timeout = timeout;
 this.execTimeout = execTimeout;
 this.waitInQueueTimeout = waitInQueueTimeout;
 this.queueSizeWarningPercent = queueSizeWarningPercent;
 if (this.timeout > 0) {
 this.taskTimeoutFlag = true;
 }
 if (this.execTimeout > 0) {
 this.taskExecTimeoutFlag = true;
 }
 if (this.waitInQueueTimeout > 0) {
 this.taskWaitInQueueTimeoutFlag = true;
 }
 if (this.queueSizeWarningPercent > 0) {
 this.queueSizeWarningFlag = true;
 }
 ThreadPoolMonitor.monitor(this);
 }
 public MonitoredThreadPoolStatisticsExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, String poolName, long timeout, long execTimeout, long waitInQueueTimeout, int queueSizeWarningPercent) {
 super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, new NamedThreadFactory(poolName));
 this.poolName = poolName;
 this.timeout = timeout;
 this.execTimeout = execTimeout;
 this.waitInQueueTimeout = waitInQueueTimeout;
 this.queueSizeWarningPercent = queueSizeWarningPercent;
 if (this.timeout > 0) {
 this.taskTimeoutFlag = true;
 }
 if (this.execTimeout > 0) {
 this.taskExecTimeoutFlag = true;
 }
 if (this.waitInQueueTimeout > 0) {
 this.taskWaitInQueueTimeoutFlag = true;
 }
 if (this.queueSizeWarningPercent > 0) {
 this.queueSizeWarningFlag = true;
 }
 ThreadPoolMonitor.monitor(this);
 }
 @Override
 public void execute(Runnable command) {
 this.taskInfoMap.put(String.valueOf(command.hashCode()), new TaskStatistics());
 if (this.queueSizeWarningFlag) {
 float f = (float) getQueue().size() / (getQueue().size() + getQueue().remainingCapacity());
 BigDecimal bd = new BigDecimal(f).setScale(2, BigDecimal.ROUND_HALF_UP);
 int usedPercent = bd.multiply(new BigDecimal(100)).intValue();
 if (usedPercent > this.queueSizeWarningPercent) {
 this.queueSizeHasWarningFlag.set(true);
 System.out.println("queueSize percent Warning!used:" + usedPercent + "%,qSize:" + getQueue().size() + ",remaining:" + getQueue().remainingCapacity());
 }
 }
 super.execute(command);
 }
 @Override
 protected void beforeExecute(Thread t, Runnable r) {
 TaskStatistics taskStatistics = this.taskInfoMap.get(String.valueOf(r.hashCode()));
 if (null != taskStatistics) {
 taskStatistics.setStartExecTime(System.currentTimeMillis());
 }
 super.beforeExecute(t, r);
 }
 @Override
 protected void afterExecute(Runnable r, Throwable t) {
 //重寫此方法做一些統(tǒng)計功能
 long endTime = System.currentTimeMillis();
 TaskStatistics taskStatistics = this.taskInfoMap.remove(String.valueOf(r.hashCode()));
 if (null != taskStatistics) {
 long taskTotalTime = endTime - taskStatistics.getCommitTime();
 long taskExecTime = endTime - taskStatistics.getStartExecTime();
 long taskWaitInQueueTime = taskStatistics.getStartExecTime() - taskStatistics.getCommitTime();
 this.taskTotalTime.addAndGet(taskTotalTime);
 this.taskTotalExecTime.addAndGet(taskExecTime);
 if (this.minTaskTime > taskTotalTime) {
 this.minTaskTime = taskTotalTime;
 }
 if (this.maxTaskTime < taskTotalTime) {
 this.maxTaskTime = taskTotalTime;
 }
 if (this.taskTimeoutFlag && taskTotalTime > this.timeout) {
 this.taskTimeoutCount.incrementAndGet();
 }
 if (this.minTaskExecTime > taskExecTime) {
 this.minTaskExecTime = taskExecTime;
 }
 if (this.maxTaskExecTime < taskExecTime) {
 this.maxTaskExecTime = taskExecTime;
 }
 if (this.taskExecTimeoutFlag && taskExecTime > this.execTimeout) {
 this.taskExecTimeoutCount.incrementAndGet();
 }
 if (this.taskWaitInQueueTimeoutFlag && taskWaitInQueueTime > this.waitInQueueTimeout) {
 this.taskWaitInQueueTimeoutCount.incrementAndGet();
 }
 System.out.println("task cost info[ taskTotalTime:" + taskTotalTime + ",taskExecTime:" + taskExecTime + ",taskWaitInQueueTime:" + taskWaitInQueueTime + " ]");
 // 初始線程數(shù)、核心線程數(shù)、正在執(zhí)行的任務(wù)數(shù)量、
 // 已完成任務(wù)數(shù)量、任務(wù)總數(shù)、隊列里緩存的任務(wù)數(shù)量、池中存在的最大線程數(shù)、
 // 最大允許的線程數(shù)、線程空閑時間、線程池是否關(guān)閉、線程池是否終止
 LOGGER.info("{}-pool-monitor: " +
 " PoolSize: {}, CorePoolSize: {}, Active: {}, " +
 "Completed: {}, Task: {}, Queue: {}, LargestPoolSize: {}, " +
 "MaximumPoolSize: {},  KeepAliveTime: {}, isShutdown: {}, isTerminated: {}",
 this.poolName,
 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());
 }
 super.afterExecute(r, t);
 }
 /**
 * Spring容器管理線程池的生命周期,線程池Bean銷毀之前先關(guān)閉     * @throws Exception
 */
 @Override
 public void destroy() throws Exception {
 shutdown();
 }
 /**
 * 線程池延遲關(guān)閉時（等待線程池里的任務(wù)都執(zhí)行完畢），統(tǒng)計線程池情況
 */
 @Override
 public void shutdown() {
 // 統(tǒng)計已執(zhí)行任務(wù)、正在執(zhí)行任務(wù)、未執(zhí)行任務(wù)數(shù)量
 LOGGER.info("{} Going to shutdown. Executed tasks: {}, Running tasks: {}, Pending tasks: {}",
 this.poolName, this.getCompletedTaskCount(), this.getActiveCount(), this.getQueue().size());
 super.shutdown();
 }
 /**
 * 線程池立即關(guān)閉時，統(tǒng)計線程池情況
 */
 @Override
 public List<Runnable> shutdownNow() {
 // 統(tǒng)計已執(zhí)行任務(wù)、正在執(zhí)行任務(wù)、未執(zhí)行任務(wù)數(shù)量
 LOGGER.info("{} Going to immediately shutdown. Executed tasks: {}, Running tasks: {}, Pending tasks: {}",
 this.poolName, this.getCompletedTaskCount(), this.getActiveCount(), this.getQueue().size());
 return super.shutdownNow();
 }
 /**
 * 任務(wù)平均時長,無已完成任務(wù)時，返回 0
 */
 public long getTaskAvgTime() {
 if (this.getCompletedTaskCount() > 0) {
 return this.getTaskTotalTime().get() / this.getCompletedTaskCount();
 }
 return 0;
 }
 /**
 * 任務(wù)平均執(zhí)行時長,無已完成任務(wù)時，返回 0
 */
 public long getTaskAvgExecTime() {
 if (this.getCompletedTaskCount() > 0) {
 return this.getTaskTotalExecTime().get() / this.getCompletedTaskCount();
 }
 return 0;
 }
 //省略setter/getter方法
} 

public class TaskStatistics {
 /**
 * 任務(wù)提交時間
 */
 private long commitTime;
 /**
 * 任務(wù)開始執(zhí)行時間
 */
 private long startExecTime;
 public TaskStatistics() {
 this.commitTime = System.currentTimeMillis();
 }
}

public class ThreadPoolMonitor {
 private static final Map<String, FutureWrapper> POOL_TASK_FUTURE_MAP = new ConcurrentHashMap<>();
 private static final ScheduledThreadPoolExecutor SCHEDULE_THREAD_POOL = new ScheduledThreadPoolExecutor(8, new NamedThreadFactory("ThreadPoolMonitor"));
 private static final Long DEFAULT_MONITOR_PERIOD_TIME_MILLS = 1000L;
 public ThreadPoolMonitor() {
 }
 public static void monitor(String name, ThreadPoolExecutor threadPoolExecutor) {
 if (threadPoolExecutor instanceof MonitoredThreadPoolStatisticsExecutor) {
 throw new IllegalArgumentException("MonitoredThreadPoolStatisticsExecutor is already monitored.");
 } else {
 monitor0(name, threadPoolExecutor, DEFAULT_MONITOR_PERIOD_TIME_MILLS);
 }
 }
 public static void remove(String name) {
 ThreadPoolMonitor.FutureWrapper futureWrapper = POOL_TASK_FUTURE_MAP.remove(name);
 if (futureWrapper != null) {
 futureWrapper.future.cancel(false);
 }
 }
 public static void remove(String name, ThreadPoolExecutor threadPoolExecutor) {
 ThreadPoolMonitor.FutureWrapper futureWrapper = POOL_TASK_FUTURE_MAP.get(name);
 if (futureWrapper != null && futureWrapper.threadPoolExecutor == threadPoolExecutor) {
 POOL_TASK_FUTURE_MAP.remove(name, futureWrapper);
 futureWrapper.future.cancel(false);
 }
 }
 static void monitor(MonitoredThreadPoolStatisticsExecutor threadPoolExecutor) {
 monitor0(threadPoolExecutor.poolName(), threadPoolExecutor, DEFAULT_MONITOR_PERIOD_TIME_MILLS);
 }
 private static void monitor0(String name, ThreadPoolExecutor threadPoolExecutor, long monitorPeriodTimeMills) {
 PoolMonitorTask poolMonitorTask = new PoolMonitorTask(threadPoolExecutor, name);
 POOL_TASK_FUTURE_MAP.compute(name, (k, v) -> {
 if (v == null) {
 return new ThreadPoolMonitor.FutureWrapper(SCHEDULE_THREAD_POOL.scheduleWithFixedDelay(poolMonitorTask, 0L, monitorPeriodTimeMills, TimeUnit.MILLISECONDS), threadPoolExecutor);
 } else {
 throw new IllegalStateException("duplicate pool name: " + name);
 }
 });
 }
 static {
 Runtime.getRuntime().addShutdownHook(new Thread(ThreadPoolMonitor.SCHEDULE_THREAD_POOL::shutdown));
 }
 static class FutureWrapper {
 private final Future<?> future;
 private final ThreadPoolExecutor threadPoolExecutor;
 public FutureWrapper(Future<?> future, ThreadPoolExecutor threadPoolExecutor) {
 this.future = future;
 this.threadPoolExecutor = threadPoolExecutor;
 }
 }
}

public class PoolMonitorTask implements Runnable {
 private final ThreadPoolExecutor monitoredThreadPool;
 private final String poolName;
 private volatile long lastTaskCount = 0L;
 public PoolMonitorTask(ThreadPoolExecutor monitoredThreadPool, String poolName) {
 this.monitoredThreadPool = monitoredThreadPool;
 this.poolName = poolName;
 }
 @Override
 public void run() {
 int activeCount = this.monitoredThreadPool.getActiveCount();
 int corePoolSize = this.monitoredThreadPool.getCorePoolSize();
 int maximumPoolSize = this.monitoredThreadPool.getMaximumPoolSize();
 int queueTaskSize = this.monitoredThreadPool.getQueue().size();
 long taskCount = this.monitoredThreadPool.getTaskCount();
 int executedTask = (int) (taskCount - this.lastTaskCount);
 log.info("線程池名稱 = {}, 活躍線程數(shù)峰值 = {}, 隊列任務(wù)數(shù)峰值 = {}, 核心線程數(shù) = {}, 最大線程數(shù) = {}, 執(zhí)行的任務(wù)總數(shù) = {}",
 this.poolName, activeCount, queueTaskSize, corePoolSize, maximumPoolSize, executedTask);
 this.lastTaskCount = taskCount;
 if (this.monitoredThreadPool.isTerminated()) {
 ThreadPoolMonitor.remove(this.poolName, this.monitoredThreadPool);
 }
 }
}

/** The capacity bound, or Integer.MAX_VALUE if none */
private final int capacity;
//將上述原生代碼改為：
private volatile int capacity;
public void setCapacity(int capacity) { 
 this.capacity = capacity;
}

public class ThreadPoolConfigUpdateListener {
 @Value("${apollo.bootstrap.namespaces:application}")
 private String namespace;
 @Autowired
 private DynamicThreadPoolFacade dynamicThreadPoolManager;
 @Autowired
 private DynamicThreadPoolProperties poolProperties;
 @PostConstruct
 public void init() {
 initConfigUpdateListener();
 }
 public void initConfigUpdateListener() {
 String apolloNamespace = namespace;
 if (StringUtils.hasText(poolProperties.getApolloNamespace())) {
 apolloNamespace = poolProperties.getApolloNamespace();
 }
 String finalApolloNamespace = apolloNamespace;
 Config config = ConfigService.getConfig(finalApolloNamespace);
 config.addChangeListener(changeEvent -> {
 try {
 Thread.sleep(poolProperties.getWaitRefreshConfigSeconds() * 1000);
 } catch (InterruptedException e) {
 log.error("配置刷新異常",e);
 }
 dynamicThreadPoolManager.refreshThreadPoolExecutor();
 log.info("線程池配置有變化，刷新完成");
 });
 }
}

public void refreshThreadPoolExecutor(DynamicThreadPoolProperties dynamicThreadPoolProperties) {
 try {
 dynamicThreadPoolProperties.getExecutors().forEach(poolProperties -> {
 ThreadPoolMonitor threadPoolExecutor =ThreadPoolMonitor.POOL_TASK_FUTURE_MAP.get(poolProperties.getThreadPoolName()).getThreadPoolExecutor();
 executor.setCorePoolSize(poolProperties.getCorePoolSize());
 executor.setMaxPoolSize(poolProperties.getMaximumPoolSize());
 executor.setKeepAliveSeconds((int) poolProperties.getKeepAliveTime());
 });
 }catch(Exception e){
 log.error("Executor 參數(shù)設(shè)置異常",e);
 }
 }