SpringBoot中怎么使用异步线程池实现生产环境批量数据推送

今天小编给大家分享一下SpringBoot中怎么使用异步线程池实现生产环境批量数据推送的相关知识点，内容详细，逻辑清晰，相信大部分人都还太了解这方面的知识，所以分享这篇文章给大家参考一下，希望大家阅读完这篇文章后有所收获，下面我们一起来了解一下吧。

前言

SpringBoot使用异步线程池：

1、编写线程池配置类，自定义一个线程池；

2、定义一个异步服务；

3、使用@Async注解指向定义的线程池；

这里以我工作中使用过的一个案例来做描述，我所在公司是医疗行业，敏感数据需要上报到某监管平台，所以有一个定时任务在流量较小时（一般是凌晨后）执行上报行为。但特殊时期会存在一定要在工作时间大批量上报数据的情况，且要求短时间内就要完成，此时就考虑写一个专门的异步上报接口手动执行，利用线程池上报，极大提高了速度。

编写线程池配置类

import lombok.extern.slf4j.Slf4j;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.scheduling.annotation.EnableAsync;import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import java.util.concurrent.Executor;import java.util.concurrent.ThreadPoolExecutor; /** * 类名称：ExecutorConfig * ******************************** * 
 * 类描述：线程池配置 * * @author guoj * @date 2021-09-07 09:00 */@Configuration@EnableAsync@Slf4jpublic class ExecutorConfig {    /**     * 定义数据上报线程池     * @return     */    @Bean("dataCollectionExecutor")    public Executor dataCollectionExecutor() {         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();         // 核心线程数量：当前机器的核心数        executor.setCorePoolSize(                Runtime.getRuntime().availableProcessors());         // 最大线程数        executor.setMaxPoolSize(                Runtime.getRuntime().availableProcessors() * 2);         // 队列大小        executor.setQueueCapacity(Integer.MAX_VALUE);         // 线程池中的线程名前缀        executor.setThreadNamePrefix("sjsb-");         // 拒绝策略：直接拒绝        executor.setRejectedExecutionHandler(                new ThreadPoolExecutor.AbortPolicy());         // 执行初始化        executor.initialize();         return executor;    }}

PS：

1）、需要注意，这里一定要自己定义ThreadPoolTaskExecutor线程池，否则springboot的异步注解会执行默认线程池，存在线程阻塞导致CPU飙高及内存溢出的风险。这一点可以参考阿里开发手册，线程池定义这块明确提到了这一点；

2）、在@Bean注解中定义线程池名称，后面异步注解会用到。

编写异步服务

/** * 异步方法的服务, 不影响主程序运行。 */@Servicepublic class AsyncService {     private final Logger log = LoggerFactory.getLogger(AsyncService.class);     /**     * 发送短信     */    @Async("sendMsgExecutor")    public void sendMsg(String access_token, Consult item, Map configMap) {        // 此处编写发送短信业务        // 1、buildConsultData();        // 2、sendMsg();    }     /**     * 发送微信订阅消息     */    @Async    public void sendSubscribeMsg(String access_token, Consult item, Map configMap) {        // 此处编写发送微信订阅消息业务        // 1、buildConsultData();        // 2、sendSubscribeMsg();    }     /**     * 数据并上报     */    @Async("dataCollectionExecutor")    public void buildAndPostData(String access_token, Consult item, Map configMap) {        // 此处编写上报业务，如拼接数据，然后执行上报。        // 1、buildConsultData();        // 2、postData();    }}

PS：
1）、以上是代码片段，个人经验认为专门定义一个异步service存放各个异步方法最佳，这样可以避免编码时一些误操作比如异步方法不是void或者是private修饰，导致@Async注解失效的情况，同时可以安排每个注解指向不同的自定义线程池更加灵活；
2）、@Async注解中的名称就是上面定义的自定义线程池名称，这样业务执行时就会从指定线程池中获取异步线程。

异步批量上报数据

@Autowiredprivate AsyncService asyncService; /** * 手动上报问诊记录，线程池方式。 */public void manualUploadConsultRecordsAsync(String channel, Date startTime, Date endTime) {     // 查询指定时间内的问诊记录   List consultList = consultService       .findPaidListByChannelAndTime(channel, startTime, endTime, configMap.get("serviceId"));    if (!CollectionUtils.isEmpty(consultList)) {        log.debug("[SendWZDataService][manualUploadConsultRecordsAsync]>>>> 手动上报问诊记录, 一共[{}]条", consultList.size());        consultList.forEach((item) -> {           try {               // 异步调用，使用线程池。               asyncService.buildAndPostData(access_token, item, configMap);           } catch (Exception ex) {               log.error("[SendWZDataService][manualUploadConsultRecordsAsync]>>>> 手动上报问诊记录发生异常: ", ex);           }       });   }}

以上就是"SpringBoot中怎么使用异步线程池实现生产环境批量数据推送"这篇文章的所有内容，感谢各位的阅读！相信大家阅读完这篇文章都有很大的收获，小编每天都会为大家更新不同的知识，如果还想学习更多的知识，请关注行业资讯频道。