/**
 * 自定义线程池<p>
 * 1.监控线程池状态及异常关闭等情况<p>
 * 2.监控线程池运行时的各项指标, 比如:任务执行时间、任务等待数、已完成任务数、任务异常信息、核心线程数、最大线程数等<p>
 * author: maoyingxu
 */
public class ThreadPoolExt extends ThreadPoolExecutor{
 
    private TimeUnit timeUnit;
 
    public ThreadPoolExt(int corePoolSize,
                         int maximumPoolSize,
                         long keepAliveTime,
                         TimeUnit unit,
                         BlockingQueue<Runnable> workQueue,
                         ThreadFactory threadFactory,
                         RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        this.timeUnit = unit;
    } 
 
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        monitor("ThreadPool monitor data:"); // 监控线程池运行时的各项指标
    }
 
    @Override
    protected void afterExecute(Runnable r, Throwable ex) {
        // 记录线程池执行任务的时间
        ELKLogUtils.addAppendedValue(StoredLogTag.RUNNING_DETAIL, MessageFormat.format("ThreadPool task executeTime:{0}", executeTime));
        if (ex != null) { // 监控线程池中的线程执行是否异常
            LogUtils.warn("unknown exception caught in ThreadPool afterExecute:", ex);
        }
    }
 
    @Override
    public void shutdown() {
        monitor("ThreadPool will be shutdown:"); // 线程池将要关闭事件,此方法会等待线程池中正在执行的任务和队列中等待的任务执行完毕再关闭
        super.shutdown();
    }
 
    @Override
    public List<Runnable> shutdownNow() {
        monitor("ThreadPool going to immediately be shutdown:"); // 线程池立即关闭事件,此方法会立即关闭线程池,但是会返回队列中等待的任务
 
        // 记录被丢弃的任务, 目前只记录日志, 后续可根据业务场景做进一步处理
        List<Runnable> dropTasks = null;
        try {
            dropTasks = super.shutdownNow();
            ELKLogUtils.addAppendedValue(StoredLogTag.RUNNING_DETAIL, MessageFormat.format("{0}ThreadPool discard task count:{1}{2}",
                    System.lineSeparator(), dropTasks!=null ? dropTasks.size() : 0, System.lineSeparator()));
        } catch (Exception e) {
            LogUtils.addClogException("ThreadPool shutdownNow error", e);
        }
        return dropTasks;
    }
 
    /**
     * 监控线程池运行时的各项指标, 比如:任务等待数、任务异常信息、已完成任务数、核心线程数、最大线程数等
     * @param title
     */
    private void monitor(String title){
        try {
            // 线程池监控信息记录, 这里需要注意写ES的时机,尤其是多个子线程的日志合并到主流程的记录方式
            String threadPoolMonitor = MessageFormat.format(
                    "{0}{1}core pool size:{2}, current pool size:{3}, queue wait size:{4}, active count:{5}, completed task count:{6}, " +
                            "task count:{7}, largest pool size:{8}, max pool size:{9}, keep alive time:{10}, is shutdown:{11}, is terminated:{12}, " +
                            "thread name:{13}{14}",
                    System.lineSeparator(), title, this.getCorePoolSize(), this.getPoolSize(),
                    this.getQueue().size(), this.getActiveCount(), this.getCompletedTaskCount(), this.getTaskCount(), this.getLargestPoolSize(),
                    this.getMaximumPoolSize(), this.getKeepAliveTime(timeUnit != null ? timeUnit : TimeUnit.SECONDS), this.isShutdown(),
                    this.isTerminated(), Thread.currentThread().getName(), System.lineSeparator());
            ELKLogUtils.addAppendedValue(StoredLogTag.RUNNING_DETAIL, threadPoolMonitor);
            LogUtils.info(title, threadPoolMonitor);
 
            ELKLogUtils.addFieldValue(APPIndexedLogTag.THREAD_POOL_USE_RATE, useRate); // ES埋点线程池使用率, useRate = (getActiveCount()/getMaximumPoolSize())*100
            Cat.logEvent(key, String.valueOf(useRate)); // 报警设置
        } catch (Exception e) {
            LogUtils.addClogException("ThreadPool monitor error", e);
        }
    }
 
}

2. 重写线程池拒绝策略, 拒绝策略主要参考了 Dubbo的线程池拒绝策略

public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {
 
    // 省略部分代码
 
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        String msg = String.format("Thread pool is EXHAUSTED!" +
                " Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: "
                + "%d)," +
                " Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s), in %s://%s:%d!",
            threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(),
            e.getLargestPoolSize(),
            e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating(),
            url.getProtocol(), url.getIp(), url.getPort());
        logger.warn(msg); // 记录最大负载情况下线程池的核心线程数,活跃数,最大线程数等参数
        dumpJStack(); // 记录线程堆栈信息包括锁争用信息
        throw new RejectedExecutionException(msg);
    }
 
    private void dumpJStack() {
        long now = System.currentTimeMillis();
 
        //dump every 10 minutes 每隔10分钟记录一次
        if (now - lastPrintTime < TEN_MINUTES_MILLS) {
            return;
        }
 
        if (!guard.tryAcquire()) { // 加锁访问
            return;
        }
 
        ExecutorService pool = Executors.newSingleThreadExecutor(); // 这里单独开启一个新的线程去执行(阿里的Java开发规范不允许直接调用Executors.newSingleThreadExecutor, 估计dubbo那时候还没出开发规范...)
        pool.execute(() -> {
            String dumpPath = url.getParameter(DUMP_DIRECTORY, System.getProperty("user.home"));
 
            SimpleDateFormat sdf;
 
            String os = System.getProperty(OS_NAME_KEY).toLowerCase();
 
            // window system don't support ":" in file name
            if (os.contains(OS_WIN_PREFIX)) {
                sdf = new SimpleDateFormat(WIN_DATETIME_FORMAT);
            } else {
                sdf = new SimpleDateFormat(DEFAULT_DATETIME_FORMAT);
            }
 
            String dateStr = sdf.format(new Date());
            //try-with-resources
            try (FileOutputStream jStackStream = new FileOutputStream(
                new File(dumpPath, "Dubbo_JStack.log" + "." + dateStr))) {
                JVMUtil.jstack(jStackStream);
            } catch (Throwable t) {
                logger.error("dump jStack error", t);
            } finally {
                guard.release();
            }
            lastPrintTime = System.currentTimeMillis();
        });
        //must shutdown thread pool ,if not will lead to OOM
        pool.shutdown();
 
    }
 
}