SpringBoot 中使用Spring提供的线程池ThreadPoolTaskExecutor

接着前面的文章中的`threadpool.properties`线程池配置文件中加入参数如下
```
# ThreadPoolExecutor(JDK提供的) 线程池的配置参数
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

# ThreadPoolTaskExecutor(Spring提供的) 线程池的配置参数
thread-pool-task.config.corePoolSize = 6
thread-pool-task.config.maxPoolSize = 20
thread-pool-task.config.queueCapacity = 10
thread-pool-task.config.threadNamePrefix = MyTaskThread-
thread-pool-task.config.rejectedExecutionHandler = CallerRunsPolicy
```

新建`ThreadPoolTaskConfig.java`配置类
```java
@Configuration
@PropertySource("classpath:threadpool.properties") // 指定配置文件
public class ThreadPoolTaskConfig  implements AsyncConfigurer {

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

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

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

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

@Value("${thread-pool-task.config.queueCapacity:200}")
    private Integer queueCapacity; // 阻塞队列

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

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

@Bean(name = "threadPoolTaskExecutor")
    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        // 核心线程数
        threadPoolTaskExecutor.setCorePoolSize(corePoolSize);
        // 最大线程数
        threadPoolTaskExecutor.setMaxPoolSize(maxPoolSize);
        // 当线程数大于核心数时，多余空闲线程在终止前等待新任务的最长时间；
        threadPoolTaskExecutor.setKeepAliveSeconds(keepAliveTime);
        // 阻塞队列容量
        threadPoolTaskExecutor.setQueueCapacity(queueCapacity);
        // 待任务在关闭时完成--表明等待所有线程执行完
        threadPoolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        // 线程名称前缀
        threadPoolTaskExecutor.setThreadNamePrefix(threadNamePrefix);
        // 设置拒绝策略
        threadPoolTaskExecutor.setRejectedExecutionHandler(getRejectedExecutionHandler(rejectedExecutionHandler));

threadPoolTaskExecutor.initialize();
        return threadPoolTaskExecutor;
    }

@Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (throwable, method, obj) -> {
            log.error("[ThreadPool Exception]：Message [{}], Method [{}]", throwable.getMessage(), method.getName());
            for (Object param : obj) {
                log.error("Parameter value [{}] ", param);
            }
        };
    }

/**
     * 根据传入的参数获取拒绝策略
     * @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());
    }
}
```

主启动类上加@EnableAsync注解开启异步
```java
@EnableAsync
@SpringBootApplication
public class DemoThreadpoolApplication {

public static void main(String[] args) {
        SpringApplication.run(DemoThreadpoolApplication.class, args);
    }
    
}
```

在Service类中需要异步的方法加上@Async("threadPoolTaskExecutor")注解并指定线程池Bean名
```java
@Service
public class ThreadPoolService {

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

@Async("threadPoolTaskExecutor") // 指定自定义线程池的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();
        }
    }

}
```
在Controller中调用测试
```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();
        }
    }
}
```
查看输出结果可知，因为配置了6个核心线程数，所以只有6个线程在同时工作，剩余4个任务会被放到阻塞队列中，当6个线程中有工作结束的空闲下来的时候，就会从阻塞队列中取出待执行的任务开始执行
![](/images/20220225021420730.png)