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;
?
/**
?* 類(lèi)名稱(chēng)：ExecutorConfig
?* ********************************
?* <p>
?* 類(lèi)描述：線(xiàn)程池配置
?*
?* @author guoj
?* @date 2021-09-07 09:00
?*/
@Configuration
@EnableAsync
@Slf4j
public class ExecutorConfig {
? ? /**
? ? ?* 定義數(shù)據(jù)上報(bào)線(xiàn)程池
? ? ?* @return
? ? ?*/
? ? @Bean("dataCollectionExecutor")
? ? public Executor dataCollectionExecutor() {
?
? ? ? ? ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
?
? ? ? ? // 核心線(xiàn)程數(shù)量：當(dāng)前機(jī)器的核心數(shù)
? ? ? ? executor.setCorePoolSize(
? ? ? ? ? ? ? ? Runtime.getRuntime().availableProcessors());
?
? ? ? ? // 最大線(xiàn)程數(shù)
? ? ? ? executor.setMaxPoolSize(
? ? ? ? ? ? ? ? Runtime.getRuntime().availableProcessors() * 2);
?
? ? ? ? // 隊(duì)列大小
? ? ? ? executor.setQueueCapacity(Integer.MAX_VALUE);
?
? ? ? ? // 線(xiàn)程池中的線(xiàn)程名前綴
? ? ? ? executor.setThreadNamePrefix("sjsb-");
?
? ? ? ? // 拒絕策略：直接拒絕
? ? ? ? executor.setRejectedExecutionHandler(
? ? ? ? ? ? ? ? new ThreadPoolExecutor.AbortPolicy());
?
? ? ? ? // 執(zhí)行初始化
? ? ? ? executor.initialize();
?
? ? ? ? return executor;
? ? }
}

/**
?* 異步方法的服務(wù), 不影響主程序運(yùn)行。
?*/
@Service
public class AsyncService {
?
? ? private final Logger log = LoggerFactory.getLogger(AsyncService.class);
?
? ? /**
? ? ?* 發(fā)送短信
? ? ?*/
? ? @Async("sendMsgExecutor")
? ? public void sendMsg(String access_token, Consult item, Map<String, String> configMap) {
? ? ? ? // 此處編寫(xiě)發(fā)送短信業(yè)務(wù)
? ? ? ? // 1、buildConsultData();
? ? ? ? // 2、sendMsg();
? ? }
?
? ? /**
? ? ?* 發(fā)送微信訂閱消息
? ? ?*/
? ? @Async
? ? public void sendSubscribeMsg(String access_token, Consult item, Map<String, String> configMap) {
? ? ? ? // 此處編寫(xiě)發(fā)送微信訂閱消息業(yè)務(wù)
? ? ? ? // 1、buildConsultData();
? ? ? ? // 2、sendSubscribeMsg();
? ? }
?
? ? /**
? ? ?* 數(shù)據(jù)并上報(bào)
? ? ?*/
? ? @Async("dataCollectionExecutor")
? ? public void buildAndPostData(String access_token, Consult item, Map<String, String> configMap) {
? ? ? ? // 此處編寫(xiě)上報(bào)業(yè)務(wù)，如拼接數(shù)據(jù)，然后執(zhí)行上報(bào)。
? ? ? ? // 1、buildConsultData();
? ? ? ? // 2、postData();
? ? }
}

@Autowired
private AsyncService asyncService;
?
/**
?* 手動(dòng)上報(bào)問(wèn)診記錄，線(xiàn)程池方式。
?*/
public void manualUploadConsultRecordsAsync(String channel, Date startTime, Date endTime) {
?
? ? // 查詢(xún)指定時(shí)間內(nèi)的問(wèn)診記錄
? ?List<Consult> consultList = consultService
? ? ? ?.findPaidListByChannelAndTime(channel, startTime, endTime, configMap.get("serviceId"));
?
? ?if (!CollectionUtils.isEmpty(consultList)) {
?
? ? ? ?log.debug("[SendWZDataService][manualUploadConsultRecordsAsync]>>>> 手動(dòng)上報(bào)問(wèn)診記錄, 一共[{}]條", consultList.size());
?
? ? ? ?consultList.forEach((item) -> {
? ? ? ? ? ?try {
? ? ? ? ? ? ? ?// 異步調(diào)用，使用線(xiàn)程池。
? ? ? ? ? ? ? ?asyncService.buildAndPostData(access_token, item, configMap);
? ? ? ? ? ?} catch (Exception ex) {
? ? ? ? ? ? ? ?log.error("[SendWZDataService][manualUploadConsultRecordsAsync]>>>> 手動(dòng)上報(bào)問(wèn)診記錄發(fā)生異常: ", ex);
? ? ? ? ? ?}
? ? ? ?});
? ?}
}