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

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
 * ********************************
 * <p>
 * 类描述：线程池配置
 *
 * @author guoj
 * @date 2021-09-07 09:00
 */
@Configuration
@EnableAsync
@Slf4j
public 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;
    }
}

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

@Autowired
private AsyncService asyncService;
 
/**
 * 手动上报问诊记录，线程池方式。
 */
public void manualUploadConsultRecordsAsync(String channel, Date startTime, Date endTime) {
 
    // 查询指定时间内的问诊记录
   List<Consult> 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);
           }
       });
   }
}