最近我们组杨青同学遇到一个使用线程池不当的问题：异步处理的线程池线程将主线程hang住了，分析代码发现是线程池的拒绝策略设置得不合理，设置为CallerRunsPolicy。当异步线程的执行效率降低时，阻塞队列满了，触发了拒绝策略，进而导致主线程hang死。

    从这个问题中，我们学到了两点：

    • 线程池的使用，需要充分分析业务场景后作出选择，必要的情况下需要自定义线程池；
    • 线程池的运行状况，也需要监控

    关于线程池的监控，我参考了《Java编程的艺术》中提供的思路实现的，分享下我的代码片段，如下：

    1. public class AsyncThreadExecutor implements AutoCloseable {
    3.     private static final int DEFAULT_QUEUE_SIZE = 1000;
    5.     private static final int DEFAULT_POOL_SIZE = 10;
    7.     @Setter
    8.     private int queueSize = DEFAULT_QUEUE_SIZE;
    10.     @Setter
    11.     private int poolSize = DEFAULT_POOL_SIZE;
    13.     /**
    14.      * 用于周期性监控线程池的运行状态
    15.      */
    16.     private final ScheduledExecutorService scheduledExecutorService =
    17.         Executors.newSingleThreadScheduledExecutor(new BasicThreadFactory.Builder().namingPattern("async thread executor monitor").build());
    19.     /**
    20.      * 自定义异步线程池
    21.      * （1）任务队列使用有界队列
    22.      * （2）自定义拒绝策略
    23.      */
    24.     private final ThreadPoolExecutor threadPoolExecutor =
    25.         new ThreadPoolExecutor(poolSize, poolSize, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue(queueSize),
    26.                                new BasicThreadFactory.Builder().namingPattern("async-thread-%d").build(),
    27.                                (r, executor) -> log.error("the async executor pool is full!!"));
    28.     private final ExecutorService executorService = threadPoolExecutor;
    30.     @PostConstruct
    31.     public void init() {
    32.         scheduledExecutorService.scheduleAtFixedRate(() -> {
    33.             /**
    34.              * 线程池需要执行的任务数
    35.              */
    36.             long taskCount = threadPoolExecutor.getTaskCount();
    37.             /**
    38.              * 线程池在运行过程中已完成的任务数
    39.              */
    40.             long completedTaskCount = threadPoolExecutor.getCompletedTaskCount();
    41.             /**
    42.              * 曾经创建过的最大线程数
    43.              */
    44.             long largestPoolSize = threadPoolExecutor.getLargestPoolSize();
    45.             /**
    46.              * 线程池里的线程数量
    47.              */
    48.             long poolSize = threadPoolExecutor.getPoolSize();
    49.             /**
    50.              * 线程池里活跃的线程数量
    51.              */
    52.             long activeCount = threadPoolExecutor.getActiveCount();
    54.             log.info("async-executor monitor. taskCount:{}, completedTaskCount:{}, largestPoolSize:{}, poolSize:{}, activeCount:{}",
    55.                      taskCount, completedTaskCount, largestPoolSize, poolSize, activeCount);
    56.         }, 0, 10, TimeUnit.MINUTES);
    57.     }
    59.     public void execute(Runnable task) {
    60.         executorService.execute(task);
    61.     }
    63.     @Override
    64.     public void close() throws Exception {
    65.         executorService.shutdown();
    66.     }
    67. }

    这里的主要思路是：（1）使用有界队列的固定数量线程池；（2）拒绝策略是将任务丢弃，但是需要记录错误日志；（3）使用一个调度线程池对业务线程池进行监控。

    在查看监控日志的时候，看到下图所示的监控日志：
    Java线程池监控小结 - 图1

    这里我对largestPooSize的含义比较困惑，按字面理解是“最大的线程池数量”，但是按照线程池的定义，maximumPoolSize和coreSize相同的时候（在这里，都是10），一个线程池里的最大线程数是10，那么为什么largestPooSize可以是39呢？我去翻这块的源码：

    1.     /**
    2.      * Returns the largest number of threads that have ever
    3.      * simultaneously been in the pool.
    4.      *
    5.      * @return the number of threads
    6.      */
    7.     public int getLargestPoolSize() {
    8.         final ReentrantLock mainLock = this.mainLock;
    9.         mainLock.lock();
    10.         try {
    11.             return largestPoolSize;
    12.         } finally {
    13.             mainLock.unlock();
    14.         }
    15.     }

    注释的翻译是：返回在这个线程池里曾经同时存在过的线程数。再看这个变量largestPoolSize在ThreadExecutor中的赋值的地方，代码如下：

    1. private boolean addWorker(Runnable firstTask, boolean core) {
    2.         retry:
    3.         for (;;) {
    4.             int c = ctl.get();
    5.             int rs = runStateOf(c);
    7.             // Check if queue empty only if necessary.
    8.             if (rs >= SHUTDOWN &&
    9.                 ! (rs == SHUTDOWN &&
    10.                    firstTask == null &&
    11.                    ! workQueue.isEmpty()))
    12.                 return false;
    14.             for (;;) {
    15.                 int wc = workerCountOf(c);
    16.                 if (wc >= CAPACITY ||
    17.                     wc >= (core ? corePoolSize : maximumPoolSize))
    18.                     return false;
    19.                 if (compareAndIncrementWorkerCount(c))
    20.                     break retry;
    21.                 c = ctl.get();  // Re-read ctl
    22.                 if (runStateOf(c) != rs)
    23.                     continue retry;
    24.                 // else CAS failed due to workerCount change; retry inner loop
    25.             }
    26.         }
    28.         boolean workerStarted = false;
    29.         boolean workerAdded = false;
    30.         Worker w = null;
    31.         try {
    32.             w = new Worker(firstTask);
    33.             final Thread t = w.thread;
    34.             if (t != null) {
    35.                 final ReentrantLock mainLock = this.mainLock;
    36.                 mainLock.lock();
    37.                 try {
    38.                     // Recheck while holding lock.
    39.                     // Back out on ThreadFactory failure or if
    40.                     // shut down before lock acquired.
    41.                     int rs = runStateOf(ctl.get());
    43.                     if (rs < SHUTDOWN ||
    44.                         (rs == SHUTDOWN && firstTask == null)) {
    45.                         if (t.isAlive()) // precheck that t is startable
    46.                             throw new IllegalThreadStateException();
    47.                         workers.add(w);
    48.                         int s = workers.size();
    49.                         if (s > largestPoolSize)
    50.                             largestPoolSize = s;//这里这里！
    51.                         workerAdded = true;
    52.                     }
    53.                 } finally {
    54.                     mainLock.unlock();
    55.                 }
    56.                 if (workerAdded) {
    57.                     t.start();
    58.                     workerStarted = true;
    59.                 }
    60.             }
    61.         } finally {
    62.             if (! workerStarted)
    63.                 addWorkerFailed(w);
    64.         }
    65.         return workerStarted;
    66.     }

    发现两点：

    • largestPoolSize是worker集合的历史最大值，只增不减。largestPoolSize的大小是线程池曾创建的线程个数，跟线程池的容量无关；
    • largestPoolSize<=maximumPoolSize。

    PS：杨青同学是这篇文章的灵感来源，他做了很多压测。给了我很多思路，并跟我一起分析了一些代码。

    本号专注于后端技术、JVM问题排查和优化、Java面试题、个人成长和自我管理等主题，为读者提供一线开发者的工作和成长经验，期待你能在这里有所收获。
    Java线程池监控小结 - 图2