SpringBoot：详解@EnableAsync + @Async 实现共享线程池

锲子

平时的开发中，可以看到很多地方要使用多线程技术，

比如：
1.处理大数据量的数据时，可以采用线程池，充分利用多核优势
2.用户触发一个较长的流程时，可以将一部分处理逻辑，另起一个线程异步处理，减少用户等待时间

不过线程是一种宝贵的资源，一个系统运行在服务器上，要根据CPU的数量来合理设置并发线程数量。
如果一个系统中每个线程使用者都自己定义线程或者线程池，有一些可见的不良后果
比如：
1.系统各处启线程太多，导致CPU切换上下文的消耗
2.定义线程池的参数不一致，导致各种不同实现共存难以处理和排查问题
所以本文介绍一种安全又干净的方式：在Springboot中，使用@EnableAsync + @Async注解实现公用线程池，这里的详解就是对涉及的知识点进行一点研究和分析，网上现存的介绍多是一句话带过，比如

使用@EnableAsync来开启异步的支持，使用@Async来对某个方法进行异步执行

知识点详解

学习一个注解，同样是直接看源码和注释
为了避免篇幅过长，我分开写了两篇：

@EnableAsync

@EnableAsync 详解

@Async

@Async 详解

线程池参数

关于线程池参数等知识，美团技术团队这批文章很好，建议学习
Java线程池实现原理及其在美团业务中的实践

实战&举例

由知识点详解中的内容可知，有两种方式可以实现线程池

方式一继承AsyncConfigurer接口，直接使用@Async

线程池配置类

@Configuration
@EnableAsync
public class AsyncExecutor implements AsyncConfigurer {

    private static final Logger logger = LoggerFactory.getLogger(AsyncExecutor.class);

    //核心线程数
    private static final int CORE_POOL_SIZE = 5;

    //最大线程数
    private static final int MAX_POOL_SIZE = 5;

    //队列大小
    private static final int QUEUE_CAPACITY = 50;

    //线程池中的线程的名称前缀
    private static final String THREAD_NAME = "MyExecutor-";


    @Override
    public ThreadPoolTaskExecutor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //配置核心线程数
        executor.setCorePoolSize(CORE_POOL_SIZE);
        //配置最大线程数
        executor.setMaxPoolSize(MAX_POOL_SIZE);
        //配置队列大小
        executor.setQueueCapacity(QUEUE_CAPACITY);
        //配置线程池中的线程的名称前缀
        executor.setThreadNamePrefix(THREAD_NAME);
        //配置线程池拒绝策略，我设置为CallerRunsPolicy，当线程和队列都满了，由发起线程的主线程自己执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new MyAsyncUncaughtExceptionHandler();
    }

    private class MyAsyncUncaughtExceptionHandler implements AsyncUncaughtExceptionHandler {
        @Override
        public void handleUncaughtException(Throwable throwable, Method method, Object... obj) {
            //手动处理的逻辑
            logger.error("输出报错信息");
        }
    }

方式二不继承AsyncConfigurer接口，使用@Async(“指定线程池”)

线程池配置类

@Configuration
@EnableAsync
public class AsyncExecutor2 {

    private static final Logger logger = LoggerFactory.getLogger(AsyncExecutor2.class);

    //核心线程数
    private static final int CORE_POOL_SIZE = 5;

    //最大线程数
    private static final int MAX_POOL_SIZE = 5;

    //队列大小
    private static final int QUEUE_CAPACITY = 50;

    //线程池中的线程的名称前缀
    private static final String THREAD_NAME = "MyExecutor-";


    @Bean
    public ThreadPoolTaskExecutor getAsyncExecutorMethod() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //配置核心线程数
        executor.setCorePoolSize(CORE_POOL_SIZE);
        //配置最大线程数
        executor.setMaxPoolSize(MAX_POOL_SIZE);
        //配置队列大小
        executor.setQueueCapacity(QUEUE_CAPACITY);
        //配置线程池中的线程的名称前缀
        executor.setThreadNamePrefix(THREAD_NAME);
        //配置线程池拒绝策略，我设置为CallerRunsPolicy，当线程和队列都满了，由发起线程的主线程自己执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}

测试

注解方法

@Service
public class AsyncService {
    private static final Logger logger = LoggerFactory.getLogger(AsyncService.class);
    @Async
    public void test1() throws InterruptedException {
        logger.info("---------test1 start: {}", Thread.currentThread().getName());
        Thread.sleep(1000);
        logger.info("---------test1 end: {}", Thread.currentThread().getName());
    }
    @Async
    public void test2() throws InterruptedException {
        logger.info("---------test2 start: {}", Thread.currentThread().getName());
        Thread.sleep(1000);
        logger.info("---------test2 end: {}", Thread.currentThread().getName());
    }
    @Async("getAsyncExecutorMethod")
    public void test3() throws InterruptedException {
        logger.info("---------test3 start: {}", Thread.currentThread().getName());
        Thread.sleep(5000);
        logger.info("---------test3 end: {}", Thread.currentThread().getName());
    }
    @Async("getAsyncExecutorMethod")
    public void test4() throws InterruptedException {
        logger.info("---------test4 start: {}", Thread.currentThread().getName());
        Thread.sleep(1000);
        logger.info("---------test4 end: {}", Thread.currentThread().getName());
    }
}

调用的接口

@RestController
@RequestMapping("api")
public class AsyncController {
    private static final Logger logger = LoggerFactory.getLogger(AsyncService.class);
    @Autowired
    AsyncService asyncService;

    @GetMapping("/async")
    public void ac() {
        try {
            asyncService.test1();
            asyncService.test2();
        } catch (InterruptedException e) {
            logger.error("ac error");
        }
    }

    @GetMapping("/async2")
    public void ac2() {
        try {
            asyncService.test3();
            asyncService.test4();
        } catch (InterruptedException e) {
            logger.error("ac2 error");
        }
    }
}