SpringBoot：Event實(shí)現(xiàn)發(fā)布/訂閱模式

作者：拔土豆的程序員 2022-12-02 07:28:58

不管是否基于spring boot 的發(fā)布訂閱模型，最終都是開啟了線程執(zhí)行任務(wù)，和使用第三方的MQ消息組件，問題在于重啟服務(wù)器或者未知原因崩潰的時(shí)候，消息的恢復(fù)機(jī)制要自行處理。

如圖所示支付業(yè)務(wù)中，用戶支付成功之后，后續(xù)還有很多的業(yè)務(wù)流程，但是對(duì)于用戶來講是透明的，所以為了提高接口的響應(yīng)速率，提高用戶體驗(yàn)，后續(xù)操作都會(huì)選擇異步執(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ì)列滿了之后，才會(huì)申請超過核心線程數(shù)的線程
       executor.setMaxPoolSize(10);
       // 用來緩沖執(zhí)行任務(wù)的隊(duì)列
       executor.setQueueCapacity(200);
       // 當(dāng)超過了核心線程之外的線程，在空閑時(shí)間到達(dá)之后會(huì)被銷毀
       executor.setKeepAliveSeconds(60);
       // 可以用于定位處理任務(wù)所在的線程池
       executor.setThreadNamePrefix("taskExecutor-orderSuccess-");
       // 這里采用CallerRunsPolicy策略，當(dāng)線程池沒有處理能力的時(shí)候，該策略會(huì)直接在execute方法的調(diào)用線程中運(yùn)行被拒絕的任務(wù)；
       // 如果執(zhí)行程序已關(guān)閉，則會(huì)丟棄該任務(wù)
       executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
       // 設(shè)置 線程池關(guān)閉 的時(shí)候 等待 所有任務(wù)都完成后，再繼續(xù) 銷毀 其他的 Bean，
       // 這樣這些 異步任務(wù) 的 銷毀 就會(huì)先于 數(shù)據(jù)庫連接池對(duì)象 的銷毀。
       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 {
   /**
    * 訂單號(hào)
    */
   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);
  }
}