Spring AMQP 通信

1. 概述

本教程將探討使用 Spring AMQP 框架通過 AMQP 進行基於消息的通信。首先，我們將介紹消息通信的關鍵概念。然後，我們將轉向一個實際示例。

2. 基於消息的通信

消息傳遞是一種在應用程序之間進行通信的技術。它依賴於異步消息傳遞，而不是基於同步請求-響應的架構。生產者和消息的消費者通過一箇中間消息層級——消息代理解耦。消息代理提供諸如持久消息存儲、消息過濾和消息轉換等功能。

在Java應用程序之間進行消息傳遞的案例中，通常使用JMS（Java消息服務）API。為了在不同廠商和平台之間實現互操作性，我們無法使用JMS客户端和代理。這時，AMQP就派上用場.

3. AMQP – 高級消息隊列協議

AMQP 是一種開放標準的數據傳輸規範，用於異步消息通信。它描述了消息的構建方式。

3.1. AMQP 與 JMS 的區別

由於 AMQP 是一種平台中立的二進制協議標準，因此可以使用不同的編程語言編寫庫，並在不同的環境中運行。

不存在基於供應商的協議鎖定，就像從一個 JMS 消息中間件遷移到另一個時所發生的情況。有關更多詳細信息，請參閲 JMS vs AMQP 和 Understanding AMQP。廣泛使用的 AMQP 消息中間件包括 RabbitMQ, OpenAMQ 和 StormMQ。

3.2. AMQP 實體

簡而言之，AMQP 由交換 (Exchanges)、隊列 (Queues) 和綁定 (Bindings) 組成：

交換 (Exchanges) 類似於郵局或郵箱，客户端將消息發佈到 AMQP 交換。內置四種交換類型如下：
1. 直接交換 (Direct Exchange) – 通過匹配完整的路由鍵將消息路由到隊列
2. 廣播交換 (Fanout Exchange) – 將消息路由到綁定到它的所有隊列
3. 主題交換 (Topic Exchange) – 通過將路由鍵與模式匹配，將消息路由到多個隊列
4. 頭部交換 (Headers Exchange) – 基於消息頭信息路由消息
隊列 (Queues) 使用路由鍵綁定到交換
消息 (Messages) 使用路由鍵發送到交換，交換會將消息的副本分發到隊列

欲瞭解更多信息，請參閲 AMQP 概念和路由拓撲。

3.3. Spring AMQP

Spring AMQP 包含兩個模塊：spring-amqp 和 spring-rabbit。它們共同提供對以下抽象：

AMQP 實體 – 我們使用 `Message, Queue, Binding, and Exchange` 類創建實體
連接管理 – 我們通過使用 `CachingConnectionFactory` 連接到我們的 RabbitMQ 代理
消息發佈 – 我們使用 `RabbitTemplate` 發送消息
消息消費 – 我們使用 `@RabbitListener` 從隊列中讀取消息

4. 設置 RabbitMQ 代理服務器

我們需要一個可用的 RabbitMQ 代理服務器，以便我們與之連接。最簡單的方法是使用 Docker 來獲取並運行 RabbitMQ 鏡像：

docker run -d -p 5672:5672 -p 15672:15672 --name my-rabbit rabbitmq:3-management

我們暴露端口 5672，以便我們的應用程序可以連接到 RabbitMQ。

同時，我們暴露端口 15672，以便我們可以通過管理 UI (http://localhost:15672) 或 HTTP API (http://localhost:15672/api/index.html) 監視 RabbitMQ 代理的運行狀態。

5. 創建我們的 Spring AMqp 應用程序

現在，讓我們創建一個應用程序，使用 Spring AMQP 發送和接收一個簡單的“Hello, world!” 消息。

5.1. Maven 依賴

為了將 spring-amqp 和 spring-rabbit 模塊添加到我們的項目中，我們在 pom.xml 中添加了 spring-boot-starter-amqp 依賴：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
        <version>2.2.2.RELEASE</version>
    </dependency>
</dependencies>

我們可以在 Maven Central 找到最新版本：Maven Central。

5.2. 連接到我們的 RabbitMQ 代理

我們將使用 Spring Boot 的自動配置功能來創建我們的 ConnectionFactory、RabbitTemplate 和 RabbitAdmin Bean。這樣，我們就能通過默認用户名和密碼 (“guest”) 使用默認端口 5672 連接到我們的 RabbitMQ 代理。因此，我們只需在應用程序上添加 @SpringBootApplication 註解：

@SpringBootApplication
public class HelloWorldMessageApp {
   // ...
}

5.3. 創建我們的隊列

為了創建我們的隊列，我們只需定義一個類型為 Queue 的 Bean。RabbitAdmin 將會找到它並將其綁定到默認交換機，routing key 為 “myQueue”：

@Bean
public Queue myQueue() {
    return new Queue("myQueue", false);
}

我們設置隊列為不可持久化，以便在 RabbitMQ 停止時，隊列及其所有消息將被刪除。然而，重啓我們的應用程序不會影響隊列。

5.4. 發送我們的信息

讓我們使用 RabbitTemplate 發送 “Hello, world!” 消息：

rabbitTemplate.convertAndSend("myQueue", "Hello, world!");

5.5. 消費我們的消息

我們將通過在方法上添加 @RabbitListener 註解來實現消息消費者：

@RabbitListener(queues = "myQueue")
public void listen(String in) {
    System.out.println("Message read from myQueue : " + in);
}

6. 運行我們的應用程序

首先，啓動 RabbitMQ 代理服務器：

docker run -d -p 5672:5672 -p 15672:15672 --name my-rabbit rabbitmq:3-management

然後，我們通過運行 HelloWorldMessage.java, 執行 main()方法來運行 Spring Boot 應用程序：

mvn spring-boot:run -Dstart-class=com.baeldung.springamqp.simple.HelloWorldMessageApp

在應用程序運行時，我們會看到以下情況：

應用程序會將消息發送到默認交換器，並將“myQueue”作為路由鍵
然後，隊列“myQueue”接收到該消息
最後，listen 方法從“myQueue”中消費該消息並在控制枱中打印它

我們還可以使用 RabbitMQ 管理頁面（在 http://localhost:15672 處訪問）來確認我們的消息已發送和已消費。

7. 結論

在本教程中，我們探討了基於消息的架構，並使用 Spring AMQP 協議通過 AMQP 協議在應用程序之間進行通信。

知識庫 / Spring RSS 訂閱