Springboot異步事件配置和使用示例詳解

更新時間：2024年10月24日 08:59:34 作者：冰雪女媧

Spring框架提供了一套事件處理機制,允許應用在各個組件之間傳遞狀態(tài)信息,自定義事件通常繼承自ApplicationEvent類,Springboot通過自動配置簡化了異步處理的配置,實現(xiàn)開箱即用,Spring事件模型核心是觀察者模式,適用于解耦和提高響應速度

核心架構(gòu)

先簡要看一下，在Spring中要實現(xiàn)自定義事件監(jiān)聽需要涉及哪些接口類，這里忽略異步的引用、注解的實現(xiàn)，后面會說到。

基本實現(xiàn)步驟

自定義事件：一般繼承自ApplicationEvent即可，注意里面要去定義和實現(xiàn)自己的事件方法，也就是具體這個事件要做什么事，一般就在事件類、或者基于事件類去實現(xiàn)即可。
事件發(fā)布：業(yè)務代碼中注入ApplicationEventPublisher類，然后再具體業(yè)務方法中調(diào)用publishEvent方法，傳入上面自定義的事件，以及自定義的必要參數(shù)等信息
實現(xiàn)事件監(jiān)聽：有了事件、也發(fā)布了，那必須有對應的監(jiān)聽來調(diào)用具體的事件，一般實現(xiàn)ApplicationListener泛型傳入自己的事件類型即可

注意事項

異常和事物：默認情況下事件的發(fā)布、監(jiān)聽處理都是和當前業(yè)務線程綁定到一起的，也就是在同一個線程中操作事件任務。因此無論是事件發(fā)布時導致異常，或者是具體事件任務實現(xiàn)的方法異常，都會導致當前業(yè)務異常；相應的如果當前業(yè)務有事物，那么異常了也會回滾。
事件類型：首先一定要自定義自己的事件，其次在監(jiān)聽的時候也是監(jiān)聽自己的事件，而不是監(jiān)聽基類或者接口然后去判斷，這樣反而失去了基于事件監(jiān)聽編程靈活性，同時也違法開閉原則，并不利于后期擴展。具體事件中可以定義其他一些額外的參數(shù)，這樣方便在具體方法中傳參使用
事件順序：一次可以發(fā)布多個事件，無論是同一個還是不同的，執(zhí)行順序默認也是按照發(fā)布順序。

場景應用

這里以訂單完成和推送給平臺訂單相關數(shù)據(jù)為業(yè)務模型來舉例說明。Spring4.2之后提供了注解來實現(xiàn)事件監(jiān)聽，非常的方便，這里我們使用注解的方式實現(xiàn)監(jiān)聽即可。

縮略的業(yè)務類：包含事件的發(fā)布

@Resource
private ApplicationEventPublisher publisher;
public void completeTrade(TradeOrder trade){
  tradeMapper.modifyStatus(trade);
  publisher.publishEvent(new TradeStatusEvent(this,new TradeStatusEvent.Params(trade,"完成訂單")));
}

具體事件的定義：繼承自ApplicationEvent

public class TradeStatusEvent extends ApplicationEvent {
    private static final Logger logger = LoggerFactory.getLogger(TradeStatusEvent.class);
    private Params params;
    public Params getParams(){
      return this.params;
    }
    public TradeStatusEvent(Object source,Params param) {
        super(source);
        this.param = param;
    }
    public void send(){
      try{
        HttpUtils.send("xx.oo", PlatformBean.Builder().note(this.params.note)..build());
      } catch(Exception e){
        logger.error("TradeStatusEvent處理異常：",e);
      } 
    }
    public static class Params {
        private TradeOrder trade;
        private String note;
        //get、set 定義其他參數(shù)等
    }
}

監(jiān)聽實現(xiàn)：使用注解，注意這里我使用了事務監(jiān)聽注解，按照具體業(yè)務場景可以選擇具體的注解，比如最常用的@EventListener。因為我這里的訴求是當前事物提交完成之后再去推送消息，而且實際情況是啟用了異步監(jiān)聽來實現(xiàn)，同時有的人在監(jiān)聽的方法中可能還執(zhí)行了回查，也就是去查詢業(yè)務中提交的數(shù)據(jù)，那如果這里不標記為事物提交之后執(zhí)行，在異步情況下無法獲取到數(shù)據(jù)

@Component
public class TradeStatusEventListener {
    @TransactionalEventListener(phase= TransactionPhase.AFTER_COMMIT, fallbackExecution=true)
    void handlerAfterComplete(TradeStatusEvent event) {
        event.send();
    }
}

異步實現(xiàn)

所謂異步實現(xiàn)，一般是指異步監(jiān)聽，將主體業(yè)務邏輯和消息監(jiān)聽任務放到不同的線程去執(zhí)行，提高業(yè)務的響應速度。

Springboot中我們有多個辦法來實現(xiàn)異步監(jiān)聽執(zhí)行，最簡單、最直接的就和異步方法實現(xiàn)一模一樣，只需在監(jiān)聽方法上加上@Async注解（前提是啟用了異步執(zhí)行）

第一種辦法：Configuration配置類中加上注解@EnableAsync，啟用Spring的異步方法執(zhí)行能力。然后在監(jiān)聽方法上加上@Async注解，標明此方法是異步執(zhí)行。Over就這樣就行了【我們沒有配置異步線程對不對？那是會直接new Thread()來執(zhí)行異步任務嗎，當然不是，而是Spring默認提供并初始化了一個專門用來執(zhí)行異步任務的線程池ThreadPoolTaskExecutor，會接管所有的異步任務在同一個線程池中執(zhí)行。也支持定制化處理，后續(xù)我們會說到】

@Configuration
@EnableAsync
public class AppConfig{}
//````
@Component
public class TradeStatusEventListener {
    @Async
    @TransactionalEventListener(phase= TransactionPhase.AFTER_COMMIT, fallbackExecution=true)
    void handlerAfterComplete(TradeStatusEvent event) {
        event.send();
    }
}

第二種辦法：如果說不想全局開啟異步，只是想給事件監(jiān)聽的代碼實現(xiàn)異步任務呢？那最簡單就是直接在監(jiān)聽哪里new Thread().start()，不受控、不優(yōu)雅，但是業(yè)務場景簡單，訪問量小的情況下也不是不可以。那要規(guī)范一點呢，就是自己創(chuàng)建一個線程池，比如ExecutorService executorService = Executors.newCachedThreadPool()；然后在event.send哪里使用executorService.execute(..)執(zhí)行即可。
第三種辦法：優(yōu)雅點實現(xiàn)，創(chuàng)建SimpleApplicationEventMulticaster的Bean，然后創(chuàng)建一個線程池給塞進去，注意需要把自定義實現(xiàn)注入到Spring容器中。其他代碼不用做任何修改，就像同步邏輯一樣，在事件發(fā)布的時候廣播會使用multicastEvent調(diào)用taskExecutor獲取一個線程去執(zhí)行監(jiān)聽任務

@Configuration
public class AppConfig{
  @Bean
  public SimpleApplicationEventMulticaster simpleApplicationEventMulticaster(){
        SimpleApplicationEventMulticaster mu = new SimpleApplicationEventMulticaster();
        //這里我使用spring提供的任務構(gòu)造器創(chuàng)建了一個立即執(zhí)行的有界隊列任務線程池
        Executor taskExecutor = new TaskExecutorBuilder().corePoolSize(8).maxPoolSize(200).queueCapacity(20).threadNamePrefix("trade-send-").build();
        mu.setTaskExecutor(taskExecutor);
        //設置異常處理
        mu.setErrorHandler((t)->{
            //logger.error("==========調(diào)用平臺發(fā)送消息方法失敗,",t);
        });
    return mu;
  }
}

框架原理

為什么異步監(jiān)聽只需要@EnableAsync、以及在方法上加上@Async就可以了呢？

當我們使用Springboot，引入starter時會自動引入spring-boot-autoconfigure，此包里面實現(xiàn)了很多自動配置的功能（約定大于配置）名字都是xxxAutoConfiguration，比如我們這里要說的就是TaskExecutionAutoConfiguration，容器啟動的時候就會加載和創(chuàng)建默認的任務線程池，可以通過spring.task.execution開頭屬性來配置。需要注意的是，無論是否加入@EnableAsync注解TaskExecutionAutoConfiguration都會初始化一個默認的線程池，因為這個是全局的。