異步編程的幾種種實(shí)現(xiàn)方式（下）

五、 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 ，當(dāng)對(duì) execute 方法 調(diào)用時(shí)，通過自定義的線程池 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 注解標(biāo)記的方法，稱為異步方法。在spring boot應(yīng)用中使用 @Async 很簡(jiǎn)單：

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

六、Spring ApplicationEvent 事件

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

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

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

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

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

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

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

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

    }
}

最后，發(fā)布事件，把某個(gè)事件告訴所有與這個(gè)事件相關(guān)的監(jiān)聽器。相當(dāng)于 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));

加個(gè)餐：

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

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

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

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

@Component
public class SpringConfiguration {

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

}

我們看下改造后的運(yùn)行結(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
[消費(fèi)端]線程：default-executor-1，消費(fèi)事件 {"amount":100.0,"buyerName":"Tom-2","orderId":2,"sellerName":"judy-2"}
[消費(fèi)端]線程：default-executor-2，消費(fèi)事件 {"amount":100.0,"buyerName":"Tom-1","orderId":1,"sellerName":"judy-1"}
[消費(fèi)端]線程：default-executor-0，消費(fèi)事件 {"amount":100.0,"buyerName":"Tom-3","orderId":3,"sellerName":"judy-3"}

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

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

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

七、消息隊(duì)列

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

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

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

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

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

當(dāng)然市面上消息隊(duì)列框架非常多，常見的有RabbitMQ、Kafka、RocketMQ、ActiveMQ 和 Pulsar 等

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

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