今日內容是訊息佇列。大佬講了三個訊息佇列分別是 Pegion、Kafka 和 RabbitMQ。其中 Pegion 是公司自己研發的元件。今天主要在 SpringBoot 中整合 RabbitMQ,將訊息儲存在訊息佇列中並消費的過程。

Message Broker 與 AMQP 簡介

Message Broker 是一種訊息驗證、傳輸、路由的架構模式，其設計目標主要應用於下面這些場景：

• 訊息路由到一個或多個目的地
• 訊息轉化為其他的表現方式
• 執行訊息的聚集、訊息的分解，並將結果傳送到他們的目的地，然後重新組合相應返回給訊息使用者
• 呼叫Web服務來檢索資料
• 響應事件或錯誤
• 使用釋出-訂閱模式來提供內容或基於主題的訊息路由

AMQP 是 Advanced Message Queuing Protocol 的簡稱，它是一個面向訊息中介軟體的開放式標準應用層協議。AMQP 定義了這些特性：

• 訊息方向
• 訊息佇列
• 訊息路由（包括：點到點和釋出-訂閱模式）
• 可靠性
• 安全性

RabbitMQ

本文要介紹的 RabbitMQ 就是以 AMQP 協議實現的一種中介軟體產品，它可以支援多種作業系統，多種程式語言，幾乎可以覆蓋所有主流的企業級技術平臺。

SpringBoot整合

下面，我們通過在 SpringBoot 應用中整合 RabbitMQ，並實現一個簡單的傳送、接收訊息的例子來對 RabbitMQ 有一個直觀的感受和理解。

在 SpringBoot 中整合 RabbitMQ 是一件非常容易的事，因為之前我們已經介紹過 Starter POMs，其中的 AMQP 模組就可以很好的支援 RabbitMQ，下面我們就來詳細說說整合過程：

• 新建一個 SpringBoot 工程，命名為：“rabbitmq-hello”。
• 在pom.xml中引入如下依賴內容，其中spring-boot-starter-amqp用於支援 RabbitMQ。

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>1.3.7.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>
複製程式碼
 

• 在application.properties中配置關於RabbitMQ的連線和使用者資訊，使用者可以回到上面的安裝內容，在管理頁面中建立使用者。

spring.application.name=rabbitmq-hello

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=spring
spring.rabbitmq.password=123456
複製程式碼

• 建立訊息生產者Sender。通過注入AmqpTemplate介面的例項來實現訊息的傳送，AmqpTemplate介面定義了一套針對AMQP協議的基礎操作。在 SpringBoot 中會根據配置來注入其具體實現。在該生產者，我們會產生一個字串，併傳送到名為hello的佇列中。

    @Component
    public class Sender {
    
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        public void send() {
            String context = "hello " + new Date();
            System.out.println("Sender : " + context);
            this.rabbitTemplate.convertAndSend("hello",context);
        }
    
}
複製程式碼

• 建立訊息消費者Receiver。通過@RabbitListener註解定義該類對hello佇列的監聽，並用@RabbitHandler註解來指定對訊息的處理方法。所以，該消費者實現了對hello佇列的消費，消費操作為輸出訊息的字串內容。

    @Component
    @RabbitListener(queues = "hello")
    public class Receiver {
    
        @RabbitHandler
        public void process(String hello) {
            System.out.println("Receiver : " + hello);
        }
    }
複製程式碼

• 建立RabbitMQ的配置類RabbitConfig，用來配置佇列、交換器、路由等高階資訊。這裡我們以入門為主，先以最小化的配置來定義，以完成一個基本的生產和消費過程。

    @Configuration
    public class RabbitConfig {
    
        @Bean
        public Queue helloQueue() {
            return new Queue("hello");
        }
    
    }
複製程式碼

• 建立應用主類：

    @SpringBootApplication
    public class HelloApplication {
    
        public static void main(String[] args) {
            SpringApplication.run(HelloApplication.class,args);
        }
    }
複製程式碼

• 建立單元測試類，用來呼叫訊息生產：

    @RunWith(SpringJUnit4ClassRunner.class)
    @SpringApplicationConfiguration(classes = HelloApplication.class)
    public class HelloApplicationTests {
    
        @Autowired
        private Sender sender;
    
        @Test
        public void hello() throws Exception {
            sender.send();
        }
    
   }
複製程式碼

完成程式編寫之後，下面開始嘗試執行。首先確保 RabbitMQ Server 已經開始，然後進行下面的操作：

• 啟動應用主類，從控制檯中，我們看到如下內容，程式建立了一個訪問127.0.0.1:5672中springcloud的連線。 o.s.a.r.c.CachingConnectionFactory : Created new connection: SimpleConnection@29836d32 [delegate=amqp://[email protected]:5672/] 同時，我們通過 RabbitMQ 的控制面板，可以看到 Connection 和 Channels 中包含當前連線的條目。

• 執行單元測試類，我們可以看到控制檯中輸出下面的內容，訊息被髮送到了 RabbitMQ Server 的hello佇列中。 Sender : hello Sun Sep 25 11:06:11 CST 2016

• 切換到應用主類的控制檯，我們可以看到類似如下輸出，消費者對hello佇列的監聽程式執行了，並輸出了接受到的訊息資訊。 Receiver : hello Sun Sep 25 11:06:11 CST 2016

大功告成！