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
 * ********************************
 * <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;
    }
}

 PS：

1）、需要注意，這里一定要自己定義ThreadPoolTaskExecutor線程池，否則springboot的異步注解會執行默認線程池，存在線程阻塞導致CPU飆高及內存溢出的風險。這一點可以參考阿里開發手冊，線程池定義這塊明確提到了這一點；

2）、在@Bean注解中定義線程池名稱，后面異步注解會用到。

編寫異步服務

/**
 * 異步方法的服務, 不影響主程序運行。
 */
@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();
    }
}

PS：
1）、以上是代碼片段，個人經驗認為專門定義一個異步service存放各個異步方法最佳，這樣可以避免編碼時一些誤操作比如異步方法不是void或者是private修飾，導致@Async注解失效的情況，同時可以安排每個注解指向不同的自定義線程池更加靈活；
2）、@Async注解中的名稱就是上面定義的自定義線程池名稱，這樣業務執行時就會從指定線程池中獲取異步線程。

異步批量上報數據

@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);
           }
       });
   }
}

以上就是“SpringBoot中怎么使用異步線程池實現生產環境批量數據推送”這篇文章的所有內容，感謝各位的閱讀！相信大家閱讀完這篇文章都有很大的收獲，小編每天都會為大家更新不同的知識，如果還想學習更多的知識，請關注億速云行業資訊頻道。