手把手教你使用 Spring Boot 3 開發上線一個前後端分離的生產級系統（九）

Spring AMQP 介紹

AMQP（高階訊息佇列協定）是一個非同步訊息傳遞所使用的應用層協定規範，為訊息導向中介層設計，不受產品和開發語言的限制. Spring AMQP 將核心 Spring 概念應用於基於 AMQP 訊息傳遞解決方案的開發。

RabbitMQ 是基於 AMQP 協定的輕量級、可靠、可延伸、可移植的訊息中介軟體，Spring 使用 RabbitMQ 通過 AMQP 協定進行通訊。Spring Boot 為通過 RabbitMQ 使用 AMQP 提供了多種便利，包括 spring-boot-starter-amqp 「Starter」。

Spring AMQP 整合與設定

可通過如下 Docker 命令安裝 RabbiMQ：

 docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.10-management

登入 RabbiMQ 的 web 管理介面，建立虛擬主機novel:

專案中加入如下的 maven 依賴：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

在 application.yml 組態檔中加入 RabbitMQ 的連線設定：

spring:
  rabbitmq:
    addresses: "amqp://guest:[email protected]"
    virtual-host: novel
    template:
      retry:
        # 開啟重試
        enabled: true
        # 最大重試次數
        max-attempts: 3
        # 第一次和第二次重試之間的持續時間
        initial-interval: "3s"

此時已經可以在 Spring Beans 中注入 AmqpTemplate 傳送訊息了。

Spring AMQP 使用範例

在io.github.xxyopen.novel.core.constant包下建立 AMQP 相關常數類：

/**
 * AMQP 相關常數
 *
 * @author xiongxiaoyang
 * @date 2022/5/25
 */
public class AmqpConsts {

    /**
     * 小說資訊改變 MQ
     * */
    public static class BookChangeMq{

        /**
         * 小說資訊改變交換機
         * */
        public static final String EXCHANGE_NAME = "EXCHANGE-BOOK-CHANGE";

        /**
         * Elasticsearch book 索引更新的佇列
         * */
        public static final String QUEUE_ES_UPDATE = "QUEUE-ES-BOOK-UPDATE";

        /**
         * Redis book 快取更新的佇列
         * */
        public static final String QUEUE_REDIS_UPDATE = "QUEUE-REDIS-BOOK-UPDATE";

        // ... 其它的更新佇列

    }

}

在io.github.xxyopen.novel.core.config包下建立 AMQP 設定類，設定各個交換機、佇列以及繫結關係：

/**
 * AMQP 設定類
 *
 * @author xiongxiaoyang
 * @date 2022/5/25
 */
@Configuration
public class AmqpConfig {

    /**
     * 小說資訊改變交換機
     */
    @Bean
    public FanoutExchange bookChangeExchange() {
        return new FanoutExchange(AmqpConsts.BookChangeMq.EXCHANGE_NAME);
    }

    /**
     * Elasticsearch book 索引更新佇列
     */
    @Bean
    public Queue esBookUpdateQueue() {
        return new Queue(AmqpConsts.BookChangeMq.QUEUE_ES_UPDATE);
    }

    /**
     * Elasticsearch book 索引更新佇列繫結到小說資訊改變交換機
     */
    @Bean
    public Binding esBookUpdateQueueBinding() {
        return BindingBuilder.bind(esBookUpdateQueue()).to(bookChangeExchange());
    }

    // ... 其它的更新佇列以及繫結關係

}

在io.github.xxyopen.novel.manager.mq包下建立 AMQP 訊息管理類，用來傳送各種 AMQP 訊息：

/**
 * AMQP 訊息管理類
 *
 * @author xiongxiaoyang
 * @date 2022/5/25
 */
@Component
@RequiredArgsConstructor
public class AmqpMsgManager {

    private final AmqpTemplate amqpTemplate;

    @Value("${spring.amqp.enable}")
    private String enableAmqp;

    /**
     * 傳送小說資訊改變訊息
     */
    public void sendBookChangeMsg(Long bookId) {
        if (Objects.equals(enableAmqp, CommonConsts.TRUE)) {
            sendAmqpMessage(amqpTemplate, AmqpConsts.BookChangeMq.EXCHANGE_NAME, null, bookId);
        }
    }

    private void sendAmqpMessage(AmqpTemplate amqpTemplate, String exchange, String routingKey, Object message) {
        // 如果在事務中則在事務執行完成後再傳送，否則可以直接傳送
        if (TransactionSynchronizationManager.isActualTransactionActive()) {
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                @Override
                public void afterCommit() {
                    amqpTemplate.convertAndSend(exchange, routingKey, message);
                }
            });
            return;
        }
        amqpTemplate.convertAndSend(exchange, routingKey, message);
    }

}

在小說資訊更新後，傳送 AMQP 訊息：

@Transactional(rollbackFor = Exception.class)
@Override
public RestResp<Void> saveBookChapter(ChapterAddReqDto dto) {
    // 1) 儲存章節相關資訊到小說章節表
    //  a) 查詢最新章節號
   
    //  b) 設定章節相關資訊並儲存   

    // 2) 儲存章節內容到小說內容表

    // 3) 更新小說表最新章節資訊和小說總字數資訊
    //  a) 更新小說表關於最新章節的資訊
    
    //  b) 傳送小說資訊更新的 MQ 訊息
    amqpMsgManager.sendBookChangeMsg(dto.getBookId());
    return RestResp.ok();
}

在io.github.xxyopen.novel.core.listener包下建立 Rabbit 佇列監聽器，監聽各個 RabbitMQ 佇列的訊息並處理：

/**
 * Rabbit 佇列監聽器
 *
 * @author xiongxiaoyang
 * @date 2022/5/25
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class RabbitQueueListener {

    private final BookInfoMapper bookInfoMapper;

    private final ElasticsearchClient esClient;

    /**
     * 監聽小說資訊改變的 ES 更新佇列，更新最新小說資訊到 ES
     * */
    @RabbitListener(queues = AmqpConsts.BookChangeMq.QUEUE_ES_UPDATE)
    @SneakyThrows
    public void updateEsBook(Long bookId) {
        BookInfo bookInfo = bookInfoMapper.selectById(bookId);
        IndexResponse response = esClient.index(i -> i
                .index(EsConsts.BookIndex.INDEX_NAME)
                .id(bookInfo.getId().toString())
                .document(EsBookDto.build(bookInfo))
        );
        log.info("Indexed with version " + response.version());
    }

    // ... 監聽其它佇列，重新整理其它副本資料

}

此時，如果需要更新其它小說副本資料，只需要設定更新佇列和增加監聽器，不需要在小說資訊變更的地方增加任何業務程式碼，而且任意小說副本的資料重新整理之間互不影響，真正實現了模組間的解耦。