如何创建最优的线程池

它是各种度量数据的容器，类似于 windows 的系统注册表，各项度量数据都可以在其中进行注册。

根据以上的度量类型，以及线程池的特点，我们可以想到这些度量指标：

这些度量指标如此重要，在于只有了解了它们，我们才能知道线程池的工作状况，才能进行有的放矢的调优，下面让我们用一个实例来具体说明一下度量驱动调优的方法。

这是一个线程池度量调优的实例，主要有三个步骤：

• 创建线程池并注册各项度量指标
• 运行线程池并收集度量指标
• 观察度量指标并相应地调整参数

无度量，不调优， 没有调查取证，就无法做好调优。我们可以应用 dropwizard 的 metrics 库中的 https://metrics.dropwizard.io/ 类库 InstrumentedExecutorService 来帮助我们进行上述指标的统计, 它通过装饰器模式对原来的 Executor Service 进行包装，记录了 submited, running, completed, idle , duration 这些指标，我们可以另外再记录一些指标，来反映线程池的工作情况，以及线程池中运行的任务的执行情况。

部分代码：

1. 先定义一个线程池参数对象

   package com.github.walterfan.helloconcurrency;
   
   import com.codahale.metrics.Gauge;
   import com.codahale.metrics.MetricRegistry;
   import lombok.Builder;
   import lombok.Getter;
   import lombok.Setter;
   
   import java.time.Duration;
   
   
   /**
    * @Author: Walter Fan
    * 这是一个数据传输对象，封装了线程池的关键参数
    *
    * 它也应用了 lombok 的 @Getter, @Setter, @Builder 注解
    * 来自动生成 setter, getter 和 build 方法
    *
    **/
   
   @Getter
   @Setter
   @Builder
   public class ThreadPoolParam {
       private int minPoolSize;
       private int maxPoolSize;
       private Duration keepAliveTime;
       private int queueSize;
       private String threadPrefix;
       private boolean daemon;
       private MetricRegistry metricRegistry;
   
   }
   
   

2. 再写一个创建线程池的工具类

   ThreadPoolUtil.java
   
   package com.github.walterfan.helloconcurrency;
   
   import com.codahale.metrics.Counter;
   import com.codahale.metrics.Gauge;
   import com.codahale.metrics.InstrumentedExecutorService;
   import com.codahale.metrics.Meter;
   
   import com.codahale.metrics.MetricRegistry;
   import com.google.common.util.concurrent.ThreadFactoryBuilder;
   import lombok.extern.slf4j.Slf4j;
   
   import java.util.concurrent.ExecutorService;
   import java.util.concurrent.LinkedBlockingQueue;
   import java.util.concurrent.RejectedExecutionHandler;
   import java.util.concurrent.ThreadFactory;
   import java.util.concurrent.ThreadPoolExecutor;
   import java.util.concurrent.TimeUnit;
   
   import java.util.function.Supplier;
   
   import static com.codahale.metrics.MetricRegistry.name;
   
   /**
    * @Author: Walter Fan
    **/
   @Slf4j
   public class ThreadPoolUtil {
       /*
       和系统内置的 ThreadPoolExecutor.CallerRunsPolicy 差不多，
       如果被拒绝，就用提交任务的线程来执行任务.
       */
       public static class DiscardAndLogPolicy implements RejectedExecutionHandler {
           final MetricRegistry metricRegistry;
           final Meter rejectedMeter;
           final Counter rejectedCounter;
   
           public DiscardAndLogPolicy(String threadPrefix, MetricRegistry metricRegistry) {
               this.metricRegistry = metricRegistry;
               this.rejectedMeter =  metricRegistry.meter(threadPrefix + ".rejected-meter");
               this.rejectedCounter = metricRegistry.counter(threadPrefix + ".rejected-counter");
           }
   
   
           public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
               rejectedMeter.mark();
               rejectedCounter.inc();
               if (!e.isShutdown()) {
                   log.warn("reject task and run {} directly", r);
                   r.run();
   
               }
           }
       }
   
       // 创建线程执行器，注册了几个度量指标
       public static ThreadPoolExecutor createThreadExecutor(ThreadPoolParam threadPoolParam) {
           MetricRegistry metricRegistry = threadPoolParam.getMetricRegistry();
   
           metricRegistry.register(threadPoolParam.getThreadPrefix() + ".min", createIntGauge(() -> threadPoolParam.getMinPoolSize()));
           metricRegistry.register(threadPoolParam.getThreadPrefix() + ".max", createIntGauge(() -> threadPoolParam.getMaxPoolSize()));
           metricRegistry.register(threadPoolParam.getThreadPrefix() + ".queue_limitation", createIntGauge(() -> threadPoolParam.getQueueSize()));
   
   
           ThreadPoolExecutor executor = new ThreadPoolExecutor(threadPoolParam.getMinPoolSize(),
                   threadPoolParam.getMaxPoolSize(),
                   threadPoolParam.getKeepAliveTime().toMillis(),
                   TimeUnit.MILLISECONDS,
                   new LinkedBlockingQueue<Runnable>(threadPoolParam.getQueueSize()),
                   createThreadFactory(threadPoolParam),
                   createRejectedExecutionHandler(threadPoolParam));
   
           metricRegistry.register(threadPoolParam.getThreadPrefix() + ".pool_size", createIntGauge(() -> executor.getPoolSize()));
           metricRegistry.register(threadPoolParam.getThreadPrefix() + ".queue_size", createIntGauge(() -> executor.getQueue().size()));
           return executor;
       }
   
       // 创建线程执行服务，用 InstrumentedExecutorService 来包装和度量线程任务
       public static ExecutorService createExecutorService(ThreadPoolParam threadPoolParam) {
           ThreadPoolExecutor executor = createThreadExecutor(threadPoolParam);
   
           return new InstrumentedExecutorService(executor,
                   threadPoolParam.getMetricRegistry(),
                   threadPoolParam.getThreadPrefix());
       }
   
       private static Gauge<Integer> createIntGauge(Supplier<Integer> suppier) {
           return () -> suppier.get();
       }
   
       public static ThreadFactory createThreadFactory(ThreadPoolParam threadPoolParam) {
           return new ThreadFactoryBuilder()
                   .setDaemon(threadPoolParam.isDaemon())
                   .setNameFormat(threadPoolParam.getThreadPrefix() + "-%d")
                   .build();
       }
   
       public static RejectedExecutionHandler createRejectedExecutionHandler(ThreadPoolParam threadPoolParam) {
           return new DiscardAndLogPolicy(threadPoolParam.getThreadPrefix(), threadPoolParam.getMetricRegistry());
       }
   }
   

注意: 我们在这个线程池中埋设了 12 个度量指标，看你能不能在代码中找出来设置的地方 。

• cards-thread-pool.completed
• cards-thread-pool.max
• cards-thread-pool.queue_limitation
• cards-thread-pool.rejected-meter
• cards-thread-pool.duration
• cards-thread-pool.min
• cards-thread-pool.queue_size
• cards-thread-pool.running
• cards-thread-pool.idle
• cards-thread-pool.pool_size
• cards-thread-pool.rejected-counter
• cards-thread-pool.submitted
• 用线程池执行多副扑克牌的排序任务

以我们最常用的打扑克牌为例，分别用冒泡排序，插入排序和 JDK 自带的 TimSort 来对若干副牌排序，总共创建 20 个任务，都放入线程池中执行，当我们采用不同的线程池参数时，效果大不相同。扑克牌对象类及其排序代码这里从略, 可以参见 GitHub 上的源码.

这段代码让线程池执行了 30 个排序任务，最多排序 1000 副牌（52000 张）,10 个任务用冒泡排序，10 个任务用插入排序，10 个任务用 Tim 排序, 总共花了 18 秒多。

我用 CsvReporter 把若干度量指标打印到 Csv 文件中，一共有 12 个 CSV 文件。

• cards-thread-pool.completed.csv
• cards-thread-pool.max.csv
• cards-thread-pool.queue_limitation.csv
• cards-thread-pool.rejected-meter.csv
• cards-thread-pool.duration.csv
• cards-thread-pool.min.csv
• cards-thread-pool.queue_size.csv
• cards-thread-pool.running.csv
• cards-thread-pool.idle.csv
• cards-thread-pool.pool_size.csv
• cards-thread-pool.rejected-counter.csv
• cards-thread-pool.submitted.csv
• 分析度量指标，根据分析结果进行优化

基于这些度量指标，我们可以看到任务特点和线程池的参数是否合理。最直观的是批量任务的执行时间，任务在线程池队列中堆积的个数和时长，线程池中工作线程的变化，线程池由于过载而丢弃的任务，下面让我们来逐个分析这四个关键指标。