Spring Boot（十三）RabbitMQ安裝與集成

一、前言

RabbitMQ是一個開源的消息代理軟件（面向消息的中間件），它的核心作用就是創建消息隊列，異步接收和發送消息，MQ的全程是：Message Queue中文的意思是消息隊列。

1.1 使用場景

1. 削峰填谷：用於應對間歇性流量提升對於系統的“破壞”，比如秒殺活動，可以把請求先發送到消息隊列在平滑的交由系統去處理，當訪問量大於一定數量的時候，還可以直接屏蔽后續操作，給前台的用戶友好的顯示；

2. 延遲處理：可以進行事件后置，比如訂單超時業務，用戶下單30分鍾未支付取消訂單；

3. 系統解耦：消息隊列也可以幫開發人員完成業務的解耦，比如用戶上傳頭像的功能，最初的設計是用戶上傳完之后才能發帖，后面有增加了經驗系統，需要在上傳頭像之后增加經驗值，到后來又上線了金幣系統，上傳頭像之后可以增加金幣，像這種需求的不斷升級，如果在業務代碼里面寫死每次該業務代碼是很不優雅的，這個時候如果使用消息隊列，那么只需要增加一個訂閱器用於介紹用戶上傳頭像的消息，再執行經驗的增加和金幣的增加是非常簡單的，並且在不改動業務模塊業務代碼的基礎上可以輕松實現，如果后期需要撤銷某個模塊了，只需要刪除訂閱器即可，就這樣就降低了系統開發的耦合性；

1.2 為什么使用RabbitMQ？

現在市面上比較主流的消息隊列還有Kafka、RocketMQ、RabbitMQ，它們的介紹和區別如下：

Kafka： 是LinkedIn開源的分布式發布-訂閱消息系統，目前歸屬於Apache定級項目。Kafka主要特點是基於Pull的模式來處理消息消費，追求高吞吐量，一開始的目的就是用於日志收集和傳輸。0.8版本開始支持復制，對消息的重復、丟失、錯誤沒有嚴格要求，適合產生大量數據的互聯網服務的數據收集業務。

RabbitMQ： 是使用Erlang語言開發的開源消息隊列系統，基於AMQP協議來實現。AMQP的主要特征是面向消息、隊列、路由（包括點對點和發布/訂閱）、可靠性、安全。AMQP協議更多用在企業系統內，對數據一致性、穩定性和可靠性要求很高的場景，對性能和吞吐量的要求還在其次。

RocketMQ： 是阿里開源的消息中間件，它是純Java開發，具有高吞吐量、高可用性、適合大規模分布式系統應用的特點。RocketMQ思路起源於Kafka，但並不是Kafka的一個Copy，它對消息的可靠傳輸及事務性做了優化，目前在阿里集團被廣泛應用於交易、充值、流計算、消息推送、日志流式處理、binglog分發等場景。

簡單總結： Kafka的性能最好，適用於對消息吞吐量達，對消息丟失不敏感的系統；RocketMQ借鑒了Kafka並提高了消息的可靠性，修復了Kafka的不足；RabbitMQ性能略低於Kafka，並實現了AMQP（Advanced Message Queuing Protocol）高級消息隊列協議的標准，有非常好的穩定性。

支持語言對比

• RocketMQ 支持語言：Java、C++、Golang
• Kafka 支持語言：Java、Scala
• RabbitMQ 支持語言：C#、Java、Js/NodeJs、Python、Ruby、Erlang、Perl、Clojure、Golang

1.3 RabbitMQ特點

RabbitMQ的特點是易用、擴展性好（集群訪問）、高可用，具體如下：

• 可靠性：持久化、消息確認、事務等保證了消息的可靠性；
• 伸縮性：集群服務，可以很方便的添加服務器來提高系統的負載；
• 高可用：集群狀態下部分節點出現問題依然可以運行；
• 多語言支持：RabbitMQ幾乎支持了所有的語言，比如Java、.Net、Nodejs、Golang等；
• 易用的管理頁面：RabbitMQ提供了易用了網頁版的管理監控系統，可以很方便的完成RabbitMQ的控制和查看；
• 插件機制：RabbitMQ提供了許多插件，可以豐富和擴展Rabbit的功能，用戶也可編寫自己的插件；

1.4 RabbitMQ基礎知識

在了解消息通訊之前首先要了解3個概念：生產者、消費者和代理。

生產者：消息的創建者，負責創建和推送數據到消息服務器；

消費者：消息的接收方，用於處理數據和確認消息；

代理：就是RabbitMQ本身，用於扮演“快遞”的角色，本身不生產消息，只是扮演“快遞”的角色。

（一）消息發送原理

首先你必須連接到Rabbit才能發布和消費消息，那怎么連接和發送消息的呢？

你的應用程序和Rabbit Server之間會創建一個TCP連接，一旦TCP打開，並通過了認證，認證就是你試圖連接Rabbit之前發送的Rabbit服務器連接信息和用戶名和密碼，有點像程序連接數據庫，使用Java有兩種連接認證的方式，后面代碼會詳細介紹，一旦認證通過你的應用程序和Rabbit就創建了一條AMQP信道（Channel）。

信道是創建在“真實”TCP上的虛擬連接，AMQP命令都是通過信道發送出去的，每個信道都會有一個唯一的ID，不論是發布消息，訂閱隊列或者接收消息都是通過信道完成的。

（二）為什么不通過TCP直接發送命令？

對於操作系統來說創建和銷毀TCP會話是非常昂貴的開銷，假設高峰期每秒有成千上萬條連接，每個連接都要創建一條TCP會話，這就造成了TCP連接的巨大浪費，而且操作系統每秒能創建的TCP也是有限的，因此很快就會遇到系統瓶頸。

如果我們每個請求都使用一條TCP連接，既滿足了性能的需要，又能確保每個連接的私密性，這就是引入信道概念的原因。

（三）RabbitMQ名稱解釋

ConnectionFactory（連接管理器）： 應用程序與Rabbit之間建立連接的管理器，程序代碼中使用；

Channel（信道）： 消息推送使用的通道；

Exchange（交換器）： 用於接受、分配消息；

Queue（隊列）： 用於存儲生產者的消息；

RoutingKey（路由鍵）： 用於把生成者的數據分配到交換器上；

BindingKey（綁定鍵）： 用於把交換器的消息綁定到隊列上；

看到上面的解釋，最難理解的路由鍵和綁定鍵了，那么他們具體怎么發揮作用的，請看下圖：

1.5 交換器分類

RabbitMQ的Exchange（交換器）分為四類：

• direct（默認）
• headers
• fanout
• topic

其中headers交換器允許你匹配AMQP消息的header而非路由鍵，除此之外headers交換器和direct交換器完全一致，但性能卻很差，幾乎用不到，所以我們這里不做解釋。

1.5.1 direct交換器

direct為默認的交換器類型，也非常的簡單，如果路由鍵匹配的話，消息就投遞到相應的隊列，如下圖：

1.5.2 fanout交換器

fanout有別於direct交換器，fanout是一種發布/訂閱模式的交換器，當你發送一條消息的時候，交換器會把消息廣播到所有附加到這個交換器的隊列上。

注意： 對於fanout交換器來說routingKey（路由鍵）是無效的，這個參數是被忽略的。

1.5.3 topic交換器

topic交換器運行和fanout類似，但是可以更靈活的匹配自己想要訂閱的信息，這個時候routingKey路由鍵就排上用場了，使用路由鍵進行消息（規則）匹配。

topic路由器的關鍵在於定義路由鍵，定義routingKey名稱不能超過255字節，使用“.”作為分隔符，例如：com.mq.rabbit.error。

匹配規則

匹配表達式可以用“*”和“#”匹配任何字符，具體規則如下：

• “*”匹配一個分段(用“.”分割)的內容；
• “#”匹配所有字符；

例如發布了一個“cn.mq.rabbit.error”的消息：

能匹配上的路由鍵：

• cn.mq.rabbit.*
• cn.mq.rabbit.#
• #.error
• cn.mq.#
• #

不能匹配上的路由鍵：

• cn.mq.*
• *.error
• *

1.6 消息持久化

RabbitMQ隊列和交換器有一個不可告人的秘密，就是默認情況下重啟服務器會導致消息丟失，那么怎么保證Rabbit在重啟的時候不丟失呢？答案就是消息持久化。

當你把消息發送到Rabbit服務器的時候，你需要選擇你是否要進行持久化，但這並不能保證Rabbit能從崩潰中恢復，想要Rabbit消息能恢復必須滿足3個條件：

1. 投遞消息的時候durable設置為true，消息持久化，代碼：channel.queueDeclare(x, true, false, false, null)，參數2設置為true持久化；
2. 設置投遞模式deliveryMode設置為2（持久），代碼：channel.basicPublish(x, x, MessageProperties.PERSISTENT_TEXT_PLAIN,x)，參數3設置為存儲純文本到磁盤；
3. 消息已經到達持久化交換器上；
4. 消息已經到達持久化的隊列；

持久化工作原理

Rabbit會將你的持久化消息寫入磁盤上的持久化日志文件，等消息被消費之后，Rabbit會把這條消息標識為等待垃圾回收。

持久化的缺點

消息持久化的優點顯而易見，但缺點也很明顯，那就是性能，因為要寫入硬盤要比寫入內存性能較低很多，從而降低了服務器的吞吐量，盡管使用SSD硬盤可以使事情得到緩解，但他仍然吸干了Rabbit的性能，當消息成千上萬條要寫入磁盤的時候，性能是很低的。

所以使用者要根據自己的情況，選擇適合自己的方式。

學習更多RabbitMQ知識，訪問：https://gitbook.cn/gitchat/activity/5b558d54c28306099b47ae9c

二、在Docker中安裝RabbitMQ

（1）下載鏡像

https://hub.docker.com/r/library/rabbitmq/tags/

• alpine 輕量版
• management 帶插件的版本

從鏡像的大小也可以很直觀的看出來alpine是輕量版。

使用命令：

docker pull rabbitmq:3.7.7-management

下載帶management插件的版本。

（2）運行RabbitMQ

使用命令：

docker run -d --hostname myrabbit --name rabbit -p 15672:15672 -p 5672:5672 rabbitmq:3.7.7-management

• -d 后台運行
• --hostname 主機名稱
• --name 容器名稱
• -p 15672:15672 http訪問端口，映射本地端口到容器端口
• -p 5672:5672 amqp端口，映射本地端口到容器端口

正常啟動之后，訪問：http://localhost:15672/

登錄網頁管理頁面，用戶名密碼：guest/guest，登錄成功如下圖：

三、RabbitMQ集成

3.1 添加依賴

如果用Idea創建新項目，可以直接在創建Spring Boot的時候，點擊“Integration”面板，選擇RabbitMQ集成，如下圖：

如果是老Maven項目，直接在pom.xml添加如下代碼：

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

3.2 配置RabbitMQ信息

在application.properties設置如下信息：

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=test
spring.rabbitmq.password=test

3.3 代碼

3.3 代碼實現

本節分別來看三種交換器：direct、fanout、topic的實現代碼。

3.3.1 Direct Exchange

3.3.1.1 配置隊列

創建DirectConfig.java代碼如下：

package com.example.rabbitmq.mq;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class DirectConfig {
    final static String QUEUE_NAME = "direct"; //隊列名稱
    final static String EXCHANGE_NAME = "mydirect"; //交換器名稱
    @Bean
    public Queue queue() {
        // 聲明隊列 參數一：隊列名稱；參數二：是否持久化
        return new Queue(DirectConfig.QUEUE_NAME, false);
    }
    // 配置默認的交換機，以下部分都可以不配置，不設置使用默認交換器（AMQP default）
    @Bean
    DirectExchange directExchange() {
        // 參數一：交換器名稱；參數二：是否持久化；參數三：是否自動刪除消息
        return new DirectExchange(DirectConfig.EXCHANGE_NAME, false, false);
    }
    // 綁定“direct”隊列到上面配置的“mydirect”路由器
    @Bean
    Binding bindingExchangeDirectQueue(Queue directQueue, DirectExchange directExchange) {
        return BindingBuilder.bind(directQueue).to(directExchange).with(DirectConfig.QUEUE_NAME);
    }
}

3.3.1.2 發送消息

創建Sender.java代碼如下：

package com.example.rabbitmq.mq;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Date;
/**
 * 消息發送者-生產消息
 */
@Component
public class Sender {
    @Autowired
    private AmqpTemplate rabbitTemplate;
    public void driectSend(String message) {
        System.out.println("Direct 發送消息：" + message);
        //參數一：交換器名稱，可以省略（省略存儲到AMQP default交換器）；參數二：路由鍵名稱（direct模式下路由鍵=隊列名稱）；參數三：存儲消息
        this.rabbitTemplate.convertAndSend("direct", message);
    }
}

注意：

• 在direct交換器中，路由鍵名稱就是隊列的名稱；
• 發送消息“convertAndSend”的時候，第一個參數為交換器的名稱，非必填可以忽略，如果忽略則會把消息發送到默認交換器“AMQP default”；

3.3.1.3 消費消息

創建Receiver.java代碼如下：

package com.example.rabbitmq.mq;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 消息接收者-消費消息
 */
@Component
@RabbitListener(queues = "direct")
public class Receiver {
    @Autowired
    private AmqpTemplate rabbitTemplate;
    @RabbitHandler
    /**
     * 監聽消費消息
     */
    public void process(String message) {
        System.out.println("Direct 消費消息：" + message);
    }
}

3.3.1.4 測試代碼

使用Spring Boot中的默認測試框架JUnit進行單元測試，不了解JUnit的可以參考我的上一篇文章，創建MQTest.java代碼如下：

package com.example.rabbitmq.mq;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import java.text.SimpleDateFormat;
import java.util.Date;
import static org.junit.Assert.*;

@RunWith(SpringRunner.class)
@SpringBootTest
public class MQTest {
    @Autowired
    private Sender sender;
    @Test
    public void driectTest() {
        SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd");
        sender.driectSend("Driect Data：" + sf.format(new Date()));
    }
}

執行之后，效果如下圖：

表示消息已經被發送並被消費了。

3.3.2 Fanout Exchange

3.3.2.1 配置隊列

創建FanoutConfig.java代碼如下：

package com.example.rabbitmq.mq;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutConfig {
    final static String QUEUE_NAME = "fanout"; //隊列名稱
    final static String QUEUE_NAME2 = "fanout2"; //隊列名稱
    final static String EXCHANGE_NAME = "myfanout"; //交換器名稱
    @Bean
    public Queue queueFanout() {
        return new Queue(FanoutConfig.QUEUE_NAME);
    }
    @Bean
    public Queue queueFanout2() {
        return new Queue(FanoutConfig.QUEUE_NAME2);
    }
    //配置交換器
    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange(FanoutConfig.EXCHANGE_NAME);
    }
    // 綁定隊列到交換器
    @Bean
    Binding bindingFanoutExchangeQueue(Queue queueFanout, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queueFanout).to(fanoutExchange);
    }
    // 綁定隊列到交換器
    @Bean
    Binding bindingFanoutExchangeQueue2(Queue queueFanout2, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queueFanout2).to(fanoutExchange);
    }
}

3.3.2.2 發送消息

創建FanoutSender.java代碼如下：

package com.example.rabbitmq.mq;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class FanoutSender {
    @Autowired
    private AmqpTemplate rabbitTemplate;
    public void send(String message) {
        System.out.println("發送消息：" + message); this.rabbitTemplate.convertAndSend(FanoutConfig.EXCHANGE_NAME,FanoutConfig.QUEUE_NAME, message);
    }
    public void send2(String message) {
        System.out.println("發送消息2：" + message); this.rabbitTemplate.convertAndSend(FanoutConfig.EXCHANGE_NAME,FanoutConfig.QUEUE_NAME2, message);
    }
}

3.3.2.3 消費消息

創建兩個監聽類，第一個FanoutReceiver.java代碼如下：

package com.example.rabbitmq.mq;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;

@Component
@RabbitListener(queues = "fanout")
public class FanoutReceiver {
    @RabbitHandler
    public void process(String msg) {
        System.out.println("Fanout（FanoutReceiver）消費消息：" + msg);
    }
}

第二個FanoutReceiver2.java代碼如下：

package com.example.rabbitmq.mq;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = "fanout2")
public class FanoutReceiver2 {
    @RabbitHandler
    public void process(String message) {
        System.out.println("Fanout（FanoutReceiver2）消費消息：" + message);
    }
}

3.3.2.4 測試代碼

創建FanoutTest.java代碼如下：

package com.example.rabbitmq.mq;
import com.example.rabbitmq.RabbitmqApplication;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import java.text.SimpleDateFormat;
import java.util.Date;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = RabbitmqApplication.class)
public class FanoutTest {
    @Autowired
    private FanoutSender sender;

    @Test
    public void Test() throws InterruptedException {
        SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd");
        sender.send("Time1 => " + sf.format(new Date()));
        sender.send2("Date2 => " + sf.format(new Date()));
    }
}

運行測試代碼，輸出結果如下：

發送消息：Time1 => 2018-09-11
發送消息2：Date2 => 2018-09-11
Fanout（FanoutReceiver2）消費消息：Time1 => 2018-09-11
Fanout（FanoutReceiver2）消費消息：Date2 => 2018-09-11
Fanout（FanoutReceiver）消費消息：Time1 => 2018-09-11
Fanout（FanoutReceiver）消費消息：Date2 => 2018-09-11

總結： 可以看出fanout會把消息分發到所有訂閱到該交換器的隊列，fanout模式是忽略路由鍵的。

3.3.3 Topic Exchange

3.3.3.1 配置隊列

@Configuration
public class TopicConfig {
    final static String QUEUE_NAME = "log";
    final static String QUEUE_NAME2 = "log.all";
    final static String QUEUE_NAME3 = "log.all.error";
    final static String EXCHANGE_NAME = "topicExchange"; //交換器名稱
    @Bean
    public Queue queuetopic() {
        return new Queue(TopicConfig.QUEUE_NAME);
    }
    @Bean
    public Queue queuetopic2() {
        return new Queue(TopicConfig.QUEUE_NAME2);
    }
    @Bean
    public Queue queuetopic3() {
        return new Queue(TopicConfig.QUEUE_NAME3);
    }
    // 配置交換器
    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange(TopicConfig.EXCHANGE_NAME);
    }
    // 綁定隊列到交換器，並設置路由鍵（log.#）
    @Bean
    Binding bindingtopicExchangeQueue(Queue queuetopic, TopicExchange topicExchange) {
        return BindingBuilder.bind(queuetopic).to(topicExchange).with("log.#");
    }
    // 綁定隊列到交換器，並設置路由鍵（log.*）
    @Bean
    Binding bindingtopicExchangeQueue2(Queue queuetopic2, TopicExchange topicExchange) {
        return BindingBuilder.bind(queuetopic2).to(topicExchange).with("log.*");
    }
    // 綁定隊列到交換器，並設置路由鍵（log.*.error）
    @Bean
    Binding bindingtopicExchangeQueue3(Queue queuetopic3, TopicExchange topicExchange) {
        return BindingBuilder.bind(queuetopic3).to(topicExchange).with("log.*.error");
    }
}

3.3.3.2 發布消息

@Component
public class TopicSender {
    @Autowired
    private AmqpTemplate rabbitTemplate;
    public void topicSender(String message) {
        String routingKey = "log.all.error";
        System.out.println(routingKey + " 發送消息：" + message);
        this.rabbitTemplate.convertAndSend(TopicConfig.EXCHANGE_NAME, routingKey, message);
    }
}

3.3.3.3 消費消息

@Component
@RabbitListener(queues = "log")
public class TopicReceiver {
    @RabbitHandler
    public void process(String msg) {
        System.out.println("log.# 消費消息：" + msg);
    }
}

@Component
@RabbitListener(queues = "log.all")
public class TopicReceiver2 {
    @RabbitHandler
    public void process(String msg) {
        System.out.println("log.* 消費消息：" + msg);
    }
}

@Component
@RabbitListener(queues = "log.all.error")
public class TopicReceiver3 {
    @RabbitHandler
    public void process(String msg) {
        System.out.println("log.*.error 消費消息：" + msg);
    }
}

3.3.3.4 測試代碼

@RunWith(SpringRunner.class)
@SpringBootTest(classes = RabbitmqApplication.class)
public class FanoutTest {
    @Autowired
    private FanoutSender fanoutSender;
    @Test
    public void Test() {
        SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd");
        fanoutSender.send("Time1 => " + sf.format(new Date()));
        fanoutSender.send2("Date2 => " + sf.format(new Date()));
    }
}

輸出結果：

log.all.error 發送消息：time => 2018-09-11
log.# 消費消息：time => 2018-09-11
log.*.error 消費消息：time => 2018-09-11

總結： 在Topic Exchange中“#”可以匹配所有內容，而“*”則是匹配一個字符段的內容。

以上示例代碼Github地址：https://github.com/vipstone/springboot-example/tree/master/springboot-rabbitmq

參考文檔

阿里 RocketMQ 優勢對比：https://juejin.im/entry/5a0abfb5f265da43062a4a91