SpringBoot 中使用JDK提供的线程池ThreadPoolExecutor

#### JDK提供的线程池ThreadPoolExecutor的构造方法如下
```java
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)
```
```
corePoolSize：核心线程线，池中保留的线程数。如果设置了allowCoreThreadTimeOut=true，线程空闲时会销毁
maximumPoolSize：池中允许的最大线程数；
keepAliveTime：当线程数大于核心数时，多余线程的存活时间；
unit：keepAliveTime参数的时间单位；
workQueue：用于存放已提交至线程池但未被执行的任务
threadFactory：用于创建线程的工厂，开发者可以通过自定义ThreadFactory来创建线程，比如，根据业务名为线程命名、设置线程优先级、设置线程是否为守护线程等、设置线程所属的线程组等；
handler：拒绝策略，当阻塞队列已满并且已到达最大线程数，这个时候线程池就会拒绝新增的任务，该参数主要用于设置拒绝策略
```
#### 四种拒绝策略
`ThreadPoolExecutor.AbortPolicy`默认拒绝策略，即丢弃任务并抛出RejectedExecutionException异常
`ThreadPoolExecutor.DiscardPolicy`丢弃任务，但是不抛出异常
`ThreadPoolExecutor.DiscardOldestPolicy`丢弃队列最前面的任务，然后提交被当前任务
`ThreadPoolExecutor.CallerRunsPolicy`由调用者线程处理当前任务

#### 任务执行机制
1、首先检测线程池运行状态，如果不是Running，则直接拒绝；线程池要保证在Running的状态下执行任务。
2、如果workerCount < corePoolSize，则创建并启动一个线程来执行新提交的任务。
3、如果workerCount >= corePoolSize，且线程池内的阻塞队列未满，则将任务添加到该阻塞队列中。
4、如果workerCount >= corePoolSize && workerCount < maximumPoolSize，且线程池内的阻塞队列已满，则创建并启动一个线程来执行新提交的任务。执行完新提交的任务后，再去执行任务队列里面的任务，把任务队列里的任务全都执行完之后，空闲下来的时间到达设置的空闲存活时间就被销毁
5、如果workerCount >= maximumPoolSize，并且线程池内的阻塞队列已满, 则根据拒绝策略来处理该任务, 默认的处理方式是直接抛异常。
![](/images/20220221095851834.png)

#### 在SpringBoot如何使用
1、新建`threadpool.properties`线程池配置文件，各参数如下
```
thread-pool.config.corePoolSize = 5
thread-pool.config.maxPoolSize = 20
thread-pool.config.queueCapacity = 20
thread-pool.config.threadNamePrefix = MyThread-
thread-pool.config.rejectedExecutionHandler = CallerRunsPolicy
```
2、新建`ThreadPoolConfig.java `线程池配置类
```java
@EnableAsync
@Configuration
@PropertySource("classpath:threadpool.properties") // 指定配置文件
public class ThreadPoolConfig {

@Value("${thread-pool.config.corePoolSize:10}")
    private Integer corePoolSize; // 核心线程数

@Value("${thread-pool.config.maxPoolSize:20}")
    private Integer maxPoolSize; // 最大线程数

@Value("${thread-pool.config.keepAliveTime:60}")
    private Integer keepAliveTime; // 空闲线程的等待时间

@Value("${thread-pool.config.queueCapacity:50}")
    private Integer queueCapacity; // 等待队列的长度

@Value("${thread-pool.config.threadNamePrefix:MyThread-}")
    private String threadNamePrefix;  // 线程名前缀

@Value("${thread-pool.config.rejectedExecutionHandler:CallerRunsPolicy}")
    private String rejectedExecutionHandler; // 拒绝策略

@Bean(name = "myThreadPool")
    public Executor myThreadPool() {
        // 自定义线程工厂
        ThreadFactory threadFactory = new ThreadFactory() {
            private int i = 1;
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName(threadNamePrefix + i);
                i++;
                return thread;
            }
        };
        // 初始化线程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maxPoolSize,
                keepAliveTime, TimeUnit.SECONDS, new LinkedBlockingQueue<>(queueCapacity),
                threadFactory, getRejectedExecutionHandler(rejectedExecutionHandler));
        return threadPoolExecutor;
    }

/**
     * 根据传入的参数获取拒绝策略
     * @param rejectedName 拒绝策略名，比如 CallerRunsPolicy
     * @return RejectedExecutionHandler 实例对象，没有匹配的策略时，默认取 CallerRunsPolicy 实例
     */
    public RejectedExecutionHandler getRejectedExecutionHandler(String rejectedName){
        Map<String, RejectedExecutionHandler> rejectedExecutionHandlerMap = new HashMap<>(16);
        rejectedExecutionHandlerMap.put("CallerRunsPolicy", new ThreadPoolExecutor.CallerRunsPolicy());
        rejectedExecutionHandlerMap.put("AbortPolicy", new ThreadPoolExecutor.AbortPolicy());
        rejectedExecutionHandlerMap.put("DiscardPolicy", new ThreadPoolExecutor.DiscardPolicy());
        rejectedExecutionHandlerMap.put("DiscardOldestPolicy", new ThreadPoolExecutor.DiscardOldestPolicy());
        return rejectedExecutionHandlerMap.getOrDefault(rejectedName, new ThreadPoolExecutor.CallerRunsPolicy());
    }

}
```
3、新建测试Controller以及Service类
ThreadPoolController.java
```java
@RestController
@RequestMapping("/threadPool")
public class ThreadPoolController {

@Autowired
    ThreadPoolService threadPoolService;

@RequestMapping("/test")
    public void test(){
        // 循环调用10次
        for (int i = 0; i < 10; i++) {
            threadPoolService.threadPoolTest();
        }
    }
}
```
ThreadPoolService.java
```java
@Service
public class ThreadPoolService {

private static final Logger log = LoggerFactory.getLogger(ThreadPoolService.class);

@Async("myThreadPool") // 指定前面ThreadPoolConfig配置类中配置的线程池Bean名
    public void threadPoolTest() {
        log.info("=========="+Thread.currentThread().getName()+"-处理任务==========");
        try {
            Thread.sleep(10*1000); // 睡眠10秒，模拟处理业务逻辑耗费时间
            log.info("=========="+Thread.currentThread().getName()+"-处理任务结束==========");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}
```
4、查看测试结果打印输出，因为只配了5个核心线程数，同时调用了10次异步方法，所以只有5个线程在同时工作，其他5个任务会放到阻塞队列中等待，直到5个工作线程中有结束的线程，该空闲线程才从等待队列中拿出待执行的任务开始执行。
![](/images/20220222102127084.png)