Spring Boot中使用RabbitMQ

很久沒有寫Spring Boot的內容了，正好最近在寫Spring Cloud Bus的內容，因為內容會有一些相關性，所以先補一篇關於AMQP的整合。

Message Broker與AMQP簡介

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

• 消息路由到一個或多個目的地
• 消息轉化為其他的表現方式
• 執行消息的聚集、消息的分解，並將結果發送到他們的目的地，然后重新組合相應返回給消息用戶
• 調用Web服務來檢索數據
• 響應事件或錯誤
• 使用發布-訂閱模式來提供內容或基於主題的消息路由

AMQP是Advanced Message Queuing Protocol的簡稱，它是一個面向消息中間件的開放式標准應用層協議。AMQP定義了這些特性：

• 消息方向
• 消息隊列
• 消息路由（包括：點到點和發布-訂閱模式）
• 可靠性
• 安全性

RabbitMQ

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

安裝

詳情見《安裝RabbitMQ》

Rabbit管理

我們可以直接通過配置文件的訪問進行管理，也可以通過Web的訪問進行管理。下面我們將介紹如何通過Web進行管理。

• 執行rabbitmq-plugins.bat enable rabbitmq_management命令，開啟Web管理插件，這樣我們就可以通過瀏覽器來進行管理了。

D:\myProgram\RabbitMQ Server\rabbitmq_server-3.6.9\sbin>rabbitmq-plugins.bat enable rabbitmq_management
The following plugins have been enabled:
  amqp_client
  cowlib
  cowboy
  rabbitmq_web_dispatch
  rabbitmq_management_agent
  rabbitmq_management

Applying plugin configuration to rabbit@SF0001107252A... started 6 plugins.

D:\myProgram\RabbitMQ Server\rabbitmq_server-3.6.9\sbin>

 

• 打開瀏覽器並訪問：http://localhost:15672/，並使用默認用戶guest登錄，密碼也為guest。我們可以看到如下圖的管理頁面：

從圖中，我們可以看到之前章節中提到的一些基本概念，比如：Connections、Channels、Exchanges、Queue等。第一次使用的讀者，可以都點開看看都有些什么內容，熟悉一下RabbitMQ Server的服務端。

• • 點擊Admin標簽，在這里可以進行用戶的管理。

 

Spring Boot整合

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

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

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

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.dxz</groupId>
    <artifactId>rabbitmq-hello</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <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>
</project>

• 在application.properties中配置關於RabbitMQ的連接和用戶信息，用戶可以回到上面的安裝內容，在管理頁面中創建用戶。

spring.application.name=rabbitmq-hello
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

• 創建消息生產者Sender。通過注入AmqpTemplate接口的實例來實現消息的發送，AmqpTemplate接口定義了一套針對AMQP協議的基礎操作。在Spring Boot中會根據配置來注入其具體實現。在該生產者，我們會產生一個字符串，並發送到名為hello的隊列中。

package com.dxz;

import java.util.Date;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@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隊列的消費，消費操作為輸出消息的字符串內容。

package com.dxz;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = "hello")
public class Receiver {

    @RabbitHandler
    public void process(String hello) {
        System.out.println("Receiver : " + hello);
    }

}

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

package com.dxz;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.rabbitmq.client.impl.AMQImpl.Queue;

@Configuration
public class RabbitConfig {

    @Bean
    public Queue helloQueue() {
        return new Queue();
    }
}

• 創建應用主類：

package com.dxz;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class RabbitmqHelloApplication {

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

}

• 創建單元測試類，用來調用消息生產：

package com.dxz;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes = RabbitmqHelloApplication.class)
public class RabbitmqHelloApplicationTests {

    @Autowired
    private Sender sender;

    @Test
    public void hello() throws Exception {
        sender.send();
    }

}

啟動應用主類，從控制台中，我們看到如下內容，程序創建了一個訪問127.0.0.1:5672中springcloud的連接。完成程序編寫之后，下面開始嘗試運行。首先確保RabbitMQ Server已經開始，然后進行下面的操作：

2017-04-07 14:57:06.596  INFO 11120 --- [cTaskExecutor-1] o.s.a.r.c.CachingConnectionFactory       : Created new connection: SimpleConnection@185f5797 [delegate=amqp://guest@127.0.0.1:5672/]

 

運行單元測試類，我們可以看到控制台中輸出下面的內容，消息被發送到了RabbitMQ Server的hello隊列中。同時，我們通過RabbitMQ的控制面板，可以看到Connection和Channels中包含當前連接的條目。

1	Sender : hello Sun Sep 25 11:06:11 CST 2016

• 切換到應用主類的控制台，我們可以看到類似如下輸出，消費者對hello隊列的監聽程序執行了，並輸出了接受到的消息信息。

1	Receiver : hello Sun Sep 25 11:06:11 CST 2016

通過上面的示例，我們在Spring Boot應用中引入spring-boot-starter-amqp模塊，進行簡單配置就完成了對RabbitMQ的消息生產和消費的開發內容。然而在實際應用中，我們還有很多內容沒有演示，這里不做更多的講解，讀者可以自行查閱RabbitMQ的官方教程，有更全面的了解。

完整示例：Chapter5-2-1