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

更新时间:2023-04-04 21:50:38 阅读：评论：0

前言

springboot使用异步线程池：

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

2、定义一个异步服务；

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

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

编写线程池配置类

import lombok.ext壁虎吃什么ern.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* ********************************* <p>* 类描述：线程池配置** @author guoj* @date 2021-09-07 09:00*/@configuration@enableasync@slf4jpublic class executorconfig {  /**  * 定义数据上报线程池  * @return  */  @bean("datacollectionexecutor")  public executor datacollectionexecutor() {    threadpooltaskexecutor executor = new threadpooltaskexecutor();    // 核心线程数量：当前机器的核心数    executor.tcorepoolsize(        runtime.getruntime().availableprocessors());    // 最大线程数    executor.tmaxpoolsize(        runtime.getruntime().availableprocessors() * 2);    // 队列大小    executor.tqueuecapacity(integer.max_value);    // 线程池中的线程名前缀    executor.tthreadnameprefix("sjsb-");    // 拒绝策略：直接拒绝    executor.trejectedexecutionhandler(        new threadpoolexecutor.abortpolicy());    // 执行初始化    executor.initialize();    return executor;  }}

ps：

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

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

编写异步服务

/*** 异步方法的服务, 不影响主程序运行。*/@rvicepublic class asyncrvice {  private final logger log = loggerfactory.getlogger(asyncrvice.class);  /**  * 发送短信  */  @async("ndmsgexecutor")  public void ndmsg(string access_token, consult item, map<string, string> configmap) {    // 此处编写发送短信业务    // 1、buildconsultdata();    // 2、ndmsg();  }  /**  * 发送微信订阅消息  */  @async  public voi经典搞笑故事d ndsubscribemsg(string access_token, consult item, map<string, string> configmap) {    // 此处编写发送微信订阅消息业务    //忘不了的同学 1、buildconsultdata();    // 2、ndsubscribemsg();  }  /**  * 数据并上报  */  @async("datacollectionexecutor")  public void buildandpostdata(string access_token, consult item, map<string, string> configmap) {    // 此处编写上报业务，如拼接数据，然后执行上报。    // 1、buildconsultdata();    // 2、postdata();  }}

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

异步批量上报数据

@autowiredprivate asyncrvice asyncrvice;/*** 手动上报问诊记录，线程池方式。*/public void manualuploadconsultrecordsasync(string channel, date starttime, date endtime) {  // 查询指定时间内的问诊记录 list<consult> consultlist = consultrvice   .findpaidlistbychannelandtime(channel, starttime, endtime, configmap.get("rviceid")); if (!collectionutils.impty(consultlist)) {   log.debug("[ndwzdatarvice][manualuploadconsultrecordsasync]>>>> 手动上报问诊记录, 一共[{}]条", consultlist.size());   consultlist.foreach((item) -> {     try {       // 异七年级上册数学作业本2答案步调用，使用线程池。       asyncrvice.buildandpostdata(access_token, item, configmap);     } catch (exception ex) {       log.error("[ndwzdatarvice][manualuploadconsultrecordsasync]>>>> 手动上报问诊记录发生异常: ", ex);     }   }); }}