五、 SpringBoot 注解 @Async

除了硬編碼的異步編程處理方式，SpringBoot 框架還提供了 注解式 解決方案，以 方法體 為邊界，方法體內(nèi)部的代碼邏輯全部按異步方式執(zhí)行。

首先，使用 @EnableAsync 啟用異步注解

@SpringBootApplication
@EnableAsync
public class StartApplication {

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

自定義線程池：

@Configuration
@Slf4j
public class ThreadPoolConfiguration {

    @Bean(name = "defaultThreadPoolExecutor", destroyMethod = "shutdown")
    public ThreadPoolExecutor systemCheckPoolExecutorService() {

        return new ThreadPoolExecutor(3, 10, 60, TimeUnit.SECONDS,
                new LinkedBlockingQueue(10000),
                new ThreadFactoryBuilder().setNameFormat("default-executor-%d").build(),
                (r, executor) -> log.error("system pool is full! "));
    }
}

在異步處理的方法上添加注解 @Async ，當對 execute 方法 調(diào)用時，通過自定義的線程池 defaultThreadPoolExecutor 異步化執(zhí)行 execute 方法

@Service
public class AsyncServiceImpl implements AsyncService {

    @Async("defaultThreadPoolExecutor")
    public Boolean execute(Integer num) {
        System.out.println("線程：" + Thread.currentThread().getName() + " , 任務(wù)：" + num);
        return true;
    }

}

用 @Async 注解標記的方法，稱為異步方法。在spring boot應用中使用 @Async 很簡單：

• 調(diào)用異步方法類上或者啟動類加上注解 @EnableAsync
• 在需要被異步調(diào)用的方法外加上 @Async
• 所使用的 @Async 注解方法的類對象應該是Spring容器管理的bean對象；

六、Spring ApplicationEvent 事件

事件機制在一些大型項目中被經(jīng)常使用，Spring 專門提供了一套事件機制的接口，滿足了架構(gòu)原則上的解耦。

ApplicationContext 通過 ApplicationEvent 類和 ApplicationListener 接口進行事件處理。如果將實現(xiàn) ApplicationListener 接口的 bean 注入到上下文中，則每次使用 ApplicationContext 發(fā)布 ApplicationEvent 時，都會通知該 bean。本質(zhì)上，這是標準的觀察者設(shè)計模式。

ApplicationEvent 是由 Spring 提供的所有 Event 類的基類

首先，自定義業(yè)務(wù)事件子類，繼承自 ApplicationEvent，通過泛型注入業(yè)務(wù)模型參數(shù)類。相當于 MQ 的消息體。

public class OrderEvent extends AbstractGenericEvent {
    public OrderEvent(OrderModel source) {
        super(source);
    }
}

然后，編寫事件監(jiān)聽器。ApplicationListener 接口是由 Spring 提供的事件訂閱者必須實現(xiàn)的接口，我們需要定義一個子類，繼承 ApplicationListener。相當于 MQ 的消費端

@Component
public class OrderEventListener implements ApplicationListener<OrderEvent> {
    @Override
    public void onApplicationEvent(OrderEvent event) {

        System.out.println("【OrderEventListener】監(jiān)聽器處理！" + JSON.toJSONString(event.getSource()));

    }
}

最后，發(fā)布事件，把某個事件告訴所有與這個事件相關(guān)的監(jiān)聽器。相當于 MQ 的生產(chǎn)端。

OrderModel orderModel = new OrderModel();
orderModel.setOrderId((long) i);
orderModel.setBuyerName("Tom-" + i);
orderModel.setSellerName("judy-" + i);
orderModel.setAmount(100L);
// 發(fā)布Spring事件通知
SpringUtils.getApplicationContext().publishEvent(new OrderEvent(orderModel));

加個餐：

[消費端]線程：http-nio-8090-exec-1，消費事件 {"amount":100.0,"buyerName":"Tom-1","orderId":1,"sellerName":"judy-1"}
[生產(chǎn)端]線程：http-nio-8090-exec-1，發(fā)布事件 1
[消費端]線程：http-nio-8090-exec-1，消費事件 {"amount":100.0,"buyerName":"Tom-2","orderId":2,"sellerName":"judy-2"}
[生產(chǎn)端]線程：http-nio-8090-exec-1，發(fā)布事件 2
[消費端]線程：http-nio-8090-exec-1，消費事件 {"amount":100.0,"buyerName":"Tom-3","orderId":3,"sellerName":"judy-3"}
[生產(chǎn)端]線程：http-nio-8090-exec-1，發(fā)布事件 3

上面是跑了個demo的運行結(jié)果，我們發(fā)現(xiàn)無論生產(chǎn)端還是消費端，使用了同一個線程 http-nio-8090-exec-1，Spring 框架的事件機制默認是同步阻塞的。只是在代碼規(guī)范方面做了解耦，有較好的擴展性，但底層還是采用同步調(diào)用方式。

那么問題來了，如果想實現(xiàn)異步調(diào)用，如何處理？

我們需要手動創(chuàng)建一個 SimpleApplicationEventMulticaster，并設(shè)置 TaskExecutor，此時所有的消費事件采用異步線程執(zhí)行。

@Component
public class SpringConfiguration {

    @Bean
    public SimpleApplicationEventMulticaster applicationEventMulticaster(@Qualifier("defaultThreadPoolExecutor") ThreadPoolExecutor defaultThreadPoolExecutor) {
        SimpleApplicationEventMulticaster simpleApplicationEventMulticaster = new SimpleApplicationEventMulticaster();
        simpleApplicationEventMulticaster.setTaskExecutor(defaultThreadPoolExecutor);
        return simpleApplicationEventMulticaster;
    }

}

我們看下改造后的運行結(jié)果：

[生產(chǎn)端]線程：http-nio-8090-exec-1，發(fā)布事件 1
[生產(chǎn)端]線程：http-nio-8090-exec-1，發(fā)布事件 2
[生產(chǎn)端]線程：http-nio-8090-exec-1，發(fā)布事件 3
[消費端]線程：default-executor-1，消費事件 {"amount":100.0,"buyerName":"Tom-2","orderId":2,"sellerName":"judy-2"}
[消費端]線程：default-executor-2，消費事件 {"amount":100.0,"buyerName":"Tom-1","orderId":1,"sellerName":"judy-1"}
[消費端]線程：default-executor-0，消費事件 {"amount":100.0,"buyerName":"Tom-3","orderId":3,"sellerName":"judy-3"}

SimpleApplicationEventMulticaster 這個我們自己實例化的 Bean 與系統(tǒng)默認的加載順序如何？會不會有沖突？

查了下 Spring 源碼，處理邏輯在 AbstractApplicationContext#initApplicationEventMulticaster 方法中，通過 beanFactory 查找是否有自定義的 Bean，如果沒有，容器會自己 new 一個 SimpleApplicationEventMulticaster 對象注入到容器中。

代碼地址：https://github.com/aalansehaiyang/wx-project

七、消息隊列

異步架構(gòu)是互聯(lián)網(wǎng)系統(tǒng)中一種典型架構(gòu)模式，與同步架構(gòu)相對應。而消息隊列天生就是這種異步架構(gòu)，具有超高吞吐量和超低時延。

消息隊列異步架構(gòu)的主要角色包括消息生產(chǎn)者、消息隊列和消息消費者。

消息生產(chǎn)者就是主應用程序，生產(chǎn)者將調(diào)用請求封裝成消息發(fā)送給消息隊列。

消息隊列的職責就是緩沖消息，等待消費者消費。根據(jù)消費方式又分為點對點模式和發(fā)布訂閱模式兩種。

消息消費者，用來從消息隊列中拉取、消費消息，完成業(yè)務(wù)邏輯處理。

當然市面上消息隊列框架非常多，常見的有RabbitMQ、Kafka、RocketMQ、ActiveMQ 和 Pulsar 等

不同的消息隊列的功能特性會略有不同，但整體架構(gòu)類似，這里就不展開了。

我們只需要記住一個關(guān)鍵點，借助消息隊列這個中間件可以高效的實現(xiàn)異步編程。