Spring Boot的消息事件機制

時間 2019-11-21

原文原文鏈接

Spring Boot自帶了消息機制，能夠讓咱們在一個地方發佈消息，多個地方同時接收消息並處理消息，固然這是在同一個JVM內存下進行的，不一樣的進程還須要使用MQ來實現。我以爲該消息模式跟觀察者模式有必定的區別，觀察者模式通常觀察的是一個對象內部屬性發生變化的時候使用。而該消息機制能夠在任意地方使用。app

消息事件自己是一個對象，繼承於ApplicationEvent異步

@Data
public class DemoEvent extends ApplicationEvent {
    private String type;
    private List<Map> msg;

    public DemoEvent(Object source,String type,List<Map> msg) {
        super(source);
        this.type = type;
        this.msg = msg;
    }
}

還須要有一個消息事件發佈者，將這個消息事件給發佈出去async

@Component
public class DemoPublisher {
    @Autowired
    ApplicationContext applicationContext;

    public void publish(DemoEvent demoEvent) {
        applicationContext.publishEvent(demoEvent);
    }
}

而後就是咱們的偵聽者，偵聽者能夠有任意個根據業務不一樣作不一樣的處理，他的寫法分兩種，一個是實現了ApplicationListener接口，一個是在方法上打上@EventListener標籤ide

@Component
@Slf4j
public class DemoListener implements ApplicationListener<DemoEvent> {
    @Override
    @Async
    public void onApplicationEvent(DemoEvent demoEvent) {
        log.info("接收到publisher發送到消息，時間" + Time.getTime());
        List<Map> msg = demoEvent.getMsg();
        String type = demoEvent.getType();
        try {
            Thread.sleep(3000);
        }catch (Exception e) {
            e.printStackTrace();
        }
        log.info("類型" + type + ",消息內容:" + msg);
    }
}

@Component
@Slf4j
public class DemoListener1 {
    @EventListener
    public void onDemoEvent(DemoEvent demoEvent) {
        log.info("listener1經過註解接收到了publisher發送的消息，時間" + Time.getTime());
        String type = demoEvent.getType();
        List<Map> msg = demoEvent.getMsg();
        try {
            Thread.sleep(2000);
        }catch (Exception e) {
            e.printStackTrace();
        }
        log.info("listener1:類型" + type + ",消息內容：" + msg);
    }
}

可是咱們須要知道的是，多個消息監聽是同步執行的，他們會發生阻塞，因此咱們須要進行異步監聽，實現異步監聽只須要在方法上打上@Async標籤，同時在Springboot主程序中開啓容許異步測試

@EnableAsync
@SpringBootApplication
public class LanmdaApplication {

   public static void main(String[] args) {
      SpringApplication.run(LanmdaApplication.class, args);
   }

}

最後寫一個測試的Controllerthis

@Slf4j
@RestController
public class TestController {
    @Autowired
    private DemoPublisher publisher;

    @GetMapping("/test")
    public String testListener() {
        List<Map> list = new ArrayList<>();
        Map<String,String> m1 = new HashMap<>();
        m1.put("1","2");
        Map<String,String> m2 = new HashMap<>();
        m2.put("3","4");
        Map<String,String> m3 = new HashMap<>();
        m3.put("5","6");
        list.add(m1);
        list.add(m2);
        list.add(m3);
        log.info("開始發佈消息:" + Time.getTime());
        publisher.publish(new DemoEvent(this,"測試消息",list));
        log.info("消息發佈結束:" + Time.getTime());
        return "消息發佈成功";
    }
}

執行後，日誌以下spa

2019-07-21 10:42:39.686 INFO 1756 --- [nio-8080-exec-1] c.g.lanmda.controller.TestController     : 開始發佈消息:10:42:39
2019-07-21 10:42:39.687 INFO 1756 --- [nio-8080-exec-1] com.guanjian.lanmda.event.DemoListener1 : listener1經過註解接收到了publisher發送的消息，時間10:42:39
2019-07-21 10:42:41.690 INFO 1756 --- [nio-8080-exec-1] com.guanjian.lanmda.event.DemoListener1 : listener1:類型測試消息,消息內容：[{1=2}, {3=4}, {5=6}]
2019-07-21 10:42:41.695 INFO 1756 --- [nio-8080-exec-1] .s.a.AnnotationAsyncExecutionInterceptor : No task executor bean found for async processing: no bean of type TaskExecutor and no bean named 'taskExecutor' either
2019-07-21 10:42:41.697 INFO 1756 --- [nio-8080-exec-1] c.g.lanmda.controller.TestController     : 消息發佈結束:10:42:41
2019-07-21 10:42:41.697 INFO 1756 --- [cTaskExecutor-1] com.guanjian.lanmda.event.DemoListener   : 接收到publisher發送到消息，時間10:42:41
2019-07-21 10:42:44.701 INFO 1756 --- [cTaskExecutor-1] com.guanjian.lanmda.event.DemoListener   : 類型測試消息,消息內容:[{1=2}, {3=4}, {5=6}].net