SpringBoot：Event實(shí)現(xiàn)發(fā)布/訂閱模式
作者：拔土豆的程序員
不管是否基于spring boot 的發(fā)布訂閱模型，最終都是開啟了線程執(zhí)行任務(wù)，和使用第三方的MQ消息組件，問題在于重啟服務(wù)器或者未知原因崩潰的時(shí)候，消息的恢復(fù)機(jī)制要自行處理。
如圖所示支付業(yè)務(wù)中，用戶支付成功之后，后續(xù)還有很多的業(yè)務(wù)流程，但是對于用戶來講是透明的，所以為了提高接口的響應(yīng)速率，提高用戶體驗(yàn)，后續(xù)操作都會選擇異步執(zhí)行。
異步執(zhí)行方式
異步執(zhí)行主體
@Service
public class OrderService {
public void orderSuccess(){
// 訂單完成異步任務(wù)開啟 可以再統(tǒng)一封裝
Order order = new Order();
order.setOrderNo(String.valueOf(System.currentTimeMillis()));
Map<String, OrderSuccessService> orderSuccessServiceMap = SpringContextUtil.getBeansOfType(OrderSuccessService.class);
orderSuccessServiceMap.values().forEach(service -> {
service.orderSuccess(order);
});
}
}
異步執(zhí)行接口
public interface OrderSuccessService {
/**
* 訂單支付成功
* @param order
*/
public CompletableFuture<Boolean> orderSuccess(Order order);
}
@Slf4j
@Service
public class MerchantNoticeServiceImpl implements OrderSuccessService {
@Override
@Async("taskExecutor")
public CompletableFuture<Boolean> orderSuccess(Order order) {
log.info("{}商戶通知:{}",Thread.currentThread(),order);
// 返回異步調(diào)用的結(jié)果
return CompletableFuture.completedFuture(true);
}
}
@Slf4j
@Service
public class MerchantNoticeServiceImpl implements OrderSuccessService {
@Override
@Async("taskExecutor")
public CompletableFuture<Boolean> orderSuccess(Order order) {
log.info("{}商戶通知:{}",Thread.currentThread(),order);
// 返回異步調(diào)用的結(jié)果
return CompletableFuture.completedFuture(true);
}
}
@Slf4j
@Service
public class MerchantNoticeServiceImpl implements OrderSuccessService {
@Override
@Async("taskExecutor")
public CompletableFuture<Boolean> orderSuccess(Order order) {
log.info("{}商戶通知:{}",Thread.currentThread(),order);
// 返回異步調(diào)用的結(jié)果
return CompletableFuture.completedFuture(true);
}
}
自定義線程池，線程池隔離，開啟異步任務(wù)執(zhí)行
@Configuration // 配置類
@EnableAsync // @Async注解能夠生效
public class TaskConfiguration {
@Bean("taskExecutor")
public Executor taskExecutor(){
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
// 線程池創(chuàng)建時(shí)候初始化的線程數(shù)
executor.setCorePoolSize(5);
// 線程池最大的線程數(shù)，只有在緩沖隊(duì)列滿了之后，才會申請超過核心線程數(shù)的線程
executor.setMaxPoolSize(10);
// 用來緩沖執(zhí)行任務(wù)的隊(duì)列
executor.setQueueCapacity(200);
// 當(dāng)超過了核心線程之外的線程，在空閑時(shí)間到達(dá)之后會被銷毀
executor.setKeepAliveSeconds(60);
// 可以用于定位處理任務(wù)所在的線程池
executor.setThreadNamePrefix("taskExecutor-orderSuccess-");
// 這里采用CallerRunsPolicy策略，當(dāng)線程池沒有處理能力的時(shí)候，該策略會直接在execute方法的調(diào)用線程中運(yùn)行被拒絕的任務(wù)；
// 如果執(zhí)行程序已關(guān)閉，則會丟棄該任務(wù)
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
// 設(shè)置 線程池關(guān)閉 的時(shí)候 等待 所有任務(wù)都完成后，再繼續(xù) 銷毀 其他的 Bean，
// 這樣這些 異步任務(wù) 的 銷毀 就會先于 數(shù)據(jù)庫連接池對象 的銷毀。
executor.setWaitForTasksToCompleteOnShutdown(true);
// 該方法用來設(shè)置線程池中 任務(wù)的等待時(shí)間，如果超過這個(gè)時(shí)間還沒有銷毀就 強(qiáng)制銷毀，以確保應(yīng)用最后能夠被關(guān)閉，而不是阻塞住。
executor.setAwaitTerminationSeconds(60);
return executor;
}
}
Spring Event實(shí)現(xiàn)發(fā)布/訂閱模式
自定義事件：通過繼承ApplicationEve?
nt，并重寫構(gòu)造函數(shù)，實(shí)現(xiàn)事件擴(kuò)展。
public class OrderApplicationEvent extends ApplicationEvent {
public OrderApplicationEvent(OrderData orderData){
super(orderData);
}
}
定義事件的消息體
@Data
public class OrderData {
/**
* 訂單號
*/
private String orderNo;
}
事件監(jiān)聽
@Slf4j
@Service
public class MerchantNoticeListener {
@Async("asyncEventTaskExecutor")
@EventListener
public CompletableFuture<Boolean> orderSuccess(OrderApplicationEvent event) {
log.info("{}商戶通知:{}",Thread.currentThread(),event);
// 返回異步調(diào)用的結(jié)果
return CompletableFuture.completedFuture(true);
}
}
@Slf4j
@Service
public class UserNoticeListener implements ApplicationListener<OrderApplicationEvent> {
@Override
@Async("asyncEventTaskExecutor")
public void onApplicationEvent(OrderApplicationEvent event) {
log.info("{}用戶通知:{}",Thread.currentThread(),event);
}
}
@Slf4j
@Service
public class UserNoticeListener implements ApplicationListener<OrderApplicationEvent> {
@Override
@Async("asyncEventTaskExecutor")
public void onApplicationEvent(OrderApplicationEvent event) {
log.info("{}用戶通知:{}",Thread.currentThread(),event);
}
}
@Slf4j
@Service
public class UserNoticeListener implements ApplicationListener<OrderApplicationEvent> {
@Override
@Async("asyncEventTaskExecutor")
public void onApplicationEvent(OrderApplicationEvent event) {
log.info("{}用戶通知:{}",Thread.currentThread(),event);
}
}
自定義線程池
@Configuration
@Slf4j
@EnableAsync // @Async注解能夠生效
public class AsyncConfiguration implements AsyncConfigurer {
@Bean("asyncEventTaskExecutor")
public ThreadPoolTaskExecutor executor(){
//Spring封裝的一個(gè)線程池
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(5);
executor.setMaxPoolSize(50);
executor.setQueueCapacity(30);
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.setThreadNamePrefix("asyncEventTaskExecutor--orderSuccess-");
executor.initialize();
return executor;
}
@Override
public Executor getAsyncExecutor(){
return executor();
}
/**
* 異常處理
* @return
@Override
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler(){
return (ex, method, params) -> log.error(String.format("[async] task{} error:", method), ex);
}
}
事件發(fā)布
@Service
@Slf4j
public class OrderEventService {
private final ApplicationEventPublisher applicationEventPublisher;
public OrderEventService(ApplicationEventPublisher applicationEventPublisher){
this.applicationEventPublisher = applicationEventPublisher;
}
public void success(){
OrderData orderData = new OrderData();
orderData.setOrderNo(String.valueOf(System.currentTimeMillis()));
// 消息
OrderApplicationEvent orderApplicationEvent = new OrderApplicationEvent(orderData);
// 發(fā)布事件
applicationEventPublisher.publishEvent(orderApplicationEvent);
}
}
寫在最后：不管是否基于spring boot 的發(fā)布訂閱模型，最終都是開啟了線程執(zhí)行任務(wù)，和使用第三方的MQ消息組件，問題在于重啟服務(wù)器或者未知原因崩潰的時(shí)候，消息的恢復(fù)機(jī)制要自行處理。
建議使用在一些邊緣業(yè)務(wù)，比如記錄日志，這些要求沒有那么高的業(yè)務(wù)。
責(zé)任編輯：武曉燕
來源：
今日頭條
