RabbitMQ訊息佇列

同步通訊和異步通訊

微服務間通訊有同步和異步兩種方式：

同步通訊：就像打電話，需要實時回應，

異步通訊：就像發郵件，不需要馬上回復，

兩種方式各有優劣，打電話可以立即得到回應，但是你卻不能跟多個人同時通話，發送郵件可以同時與多個人收發郵件，但是往往回應會有延遲，

同步通訊

同步呼叫的優點：

時效性較強，可以立即得到結果

同步呼叫的問題：

耦合度高

性能和吞吐能力下降

有額外的資源消耗

有級聯失敗問題

我們之前學習的Feign呼叫就屬于同步方式，雖然呼叫可以實時得到結果，但存在下面的問題：

異步通訊

好處：

吞吐量提升：無需等待訂閱者處理完成，回應更快速

故障隔離：服務沒有直接呼叫，不存在級聯失敗問題

呼叫間沒有阻塞，不會造成無效的資源占用

耦合度極低，每個服務都可以靈活插拔，可替換

流量削峰：不管發布事件的流量波動多大，都由Broker接收，訂閱者可以按照自己的速度去處理事件

缺點：

架構復雜了，業務沒有明顯的流程線，不好管理

需要依賴于Broker的可靠、安全、性能

我們以購買商品為例，用戶支付后需要呼叫訂單服務完成訂單狀態修改，呼叫物流服務，從倉庫分配回應的庫存并準備發貨，

在事件模式中，支付服務是事件發布者（publisher），在支付完成后只需要發布一個支付成功的事件（event），事件中帶上訂單id，

訂單服務和物流服務是事件訂閱者（Consumer），訂閱支付成功的事件，監聽到事件后完成自己業務即可，

為了解除事件發布者與訂閱者之間的耦合，兩者并不是直接通信，而是有一個中間人（Broker），發布者發布事件到Broker，不關心誰來訂閱事件，訂閱者從Broker訂閱事件，不關心誰發來的訊息，

Broker 是一個像資料總線一樣的東西，所有的服務要接收資料和發送資料都發到這個總線上，這個總線就像協議一樣，讓服務間的通訊變得標準和可控，

常用的訊息佇列組件對比

MQ，中文是訊息佇列（MessageQueue），字面來看就是存放訊息的佇列，也就是事件驅動架構中的Broker，

比較常見的MQ實作：

ActiveMQ
RabbitMQ
RocketMQ
Kafka

幾種常見MQ的對比：

	RabbitMQ	ActiveMQ	RocketMQ	Kafka
公司/社區	Rabbit	Apache	阿里	Apache
開發語言	Erlang	Java	Java	Scala&Java
協議支持	AMQP，XMPP，SMTP，STOMP	OpenWire,STOMP，REST,XMPP,AMQP	自定義協議	自定義協議
可用性	高	一般	高	高
單機吞吐量	一般	差	高	非常高
訊息延遲	微秒級	毫秒級	毫秒級	毫秒以內
訊息可靠性	高	一般	高	一般

追求可用性：Kafka、 RocketMQ 、RabbitMQ

追求可靠性：RabbitMQ、RocketMQ

追求吞吐能力：RocketMQ、Kafka

追求訊息低延遲：RabbitMQ、Kafka

初步入門

1. 安裝RabbitMQ

1.1 單機部署

我們在Centos7虛擬機中使用Docker來安裝，

① 下載鏡像

方式一：在線拉取

docker pull rabbitmq:3-management

方式二：從本地加載

在課前資料已經提供了鏡像包：

上傳到虛擬機中后，使用命令加載鏡像即可：

#啟動docker
systemctl start docker

docker load -i mq.tar

② 安裝MQ

執行下面的命令來運行MQ容器：

docker run \
 -e RABBITMQ_DEFAULT_USER=itcast \
 -e RABBITMQ_DEFAULT_PASS=123321 \
 -v mq-plugins:/plugins \
 --name mq \
 --hostname mq1 \
 -p 15672:15672 \
 -p 5672:5672 \
 -d \
 rabbitmq:3.8-management

-v mq-plugins:/plugins \ ：將MQ的插件目錄掛載出去

然后訪問：http://192.168.194.131:15672 賬號：itcast 密碼：123321 進入到RabbitMQ的后臺管理UI界面

1.2 集群部署

后面的常見問題有說如何集群部署

2. RabbitMQ訊息模型

MQ的基本結構：

RabbitMQ中的一些角色：

publisher：生產者
consumer：消費者
exchange個：交換機，負責訊息路由
queue：佇列，存盤訊息
virtualHost：虛擬主機，隔離不同租戶的exchange、queue、訊息的隔離

RabbitMQ官方提供了5個不同的Demo示例，對應了不同的訊息模型：

3. 匯入Demo工程

課前資料提供了一個Demo工程，mq-demo:

本地idea匯入后可以看到結構如下：

包括三部分：

mq-demo：父工程，管理專案依賴
publisher：訊息的發送者
consumer：訊息的消費者

4. 入門案例

簡單佇列模式的模型圖：

官方的HelloWorld是基于最基礎的訊息佇列模型來實作的，只包括三個角色：

publisher：訊息發布者，將訊息發送到佇列queue
queue：訊息佇列，負責接受并快取訊息
consumer：訂閱佇列，處理佇列中的訊息

4.1 publisher實作

思路：

建立連接
創建Channel
宣告佇列
發送訊息
關閉連接和channel

代碼實作：

package cn.itcast.mq.helloworld;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class PublisherTest {
    @Test
    public void testSendMessage() throws IOException, TimeoutException {
        // 1.建立連接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.設定連接引數，分別是：主機名、埠號、vhost、用戶名、密碼
        factory.setHost("192.168.150.101");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("itcast");
        factory.setPassword("123321");
        // 1.2.建立連接
        Connection connection = factory.newConnection();

        // 2.創建通道Channel
        Channel channel = connection.createChannel();

        // 3.創建佇列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.發送訊息
        String message = "hello, rabbitmq!";
        channel.basicPublish("", queueName, null, message.getBytes());
        System.out.println("發送訊息成功：【" + message + "】");

        // 5.關閉通道和連接
        channel.close();
        connection.close();

    }
}

4.2 consumer實作

代碼思路：

建立連接
創建Channel
宣告佇列
訂閱訊息

代碼實作：

package cn.itcast.mq.helloworld;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConsumerTest {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.建立連接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.設定連接引數，分別是：主機名、埠號、vhost、用戶名、密碼
        factory.setHost("192.168.150.101");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("itcast");
        factory.setPassword("123321");
        // 1.2.建立連接
        Connection connection = factory.newConnection();

        // 2.創建通道Channel
        Channel channel = connection.createChannel();

        // 3.創建佇列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.訂閱訊息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 5.處理訊息
                String message = new String(body);
                System.out.println("接收到訊息：【" + message + "】");
            }
        });
        System.out.println("等待接收訊息，，，，");
    }
}

5. 總結

基本訊息佇列的訊息發送流程：

建立connection
創建channel
利用channel宣告佇列
利用channel向佇列發送訊息

基本訊息佇列的訊息接收流程：

建立connection
創建channel
利用channel宣告佇列
定義consumer的消費行為handleDelivery()
利用channel將消費者與佇列系結

SpringAMQP

基本使用

無論使用哪種模型，訊息發送者和接收者都需要設定如下配置：

0. 部署并創建運行RabbitMQ容器

在前面安裝RabbitMQ中的單機部署有教

1. 匯入依賴

<!--AMQP依賴，包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2. 添加配置yaml

記得改主機名為虛擬機ip地址

spring:
  rabbitmq:
    host: 192.168.150.101 # 主機名
    port: 5672 # 埠
    virtual-host: / # 虛擬主機
    username: itcast # 用戶名
    password: 123321 # 密碼

這個是只有在任務模型時的接收者需要配置的：

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1 # 每次只能獲取一條訊息，處理完成才能獲取下一個訊息

3. 配置訊息轉換器

在publisher和consumer兩個服務中都引入依賴：

<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-xml</artifactId>
    <version>2.9.10</version>
</dependency>

發送者和接收者都在啟動類中添加一個Bean即可：

@Bean
public MessageConverter jsonMessageConverter(){
    return new Jackson2JsonMessageConverter();
}

4. 根據不同的模型寫發送類和接收類

寫完這些類后記得，先啟動接收者服務，再運行發送者測驗，因為佇列和交換機寫在接收者了，所以需要先啟動創建佇列和交換機，發送者才能成功發送到佇列，【同理也可以把佇列和交換機寫在發送者中】

五大模型

Basic Queue 簡單佇列模型

宣告佇列

@Configuration
public class BasicConfig {
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("simple.queue");
    }
}

訊息發送

在publisher服務中撰寫測驗類SpringAmqpTest，并利用RabbitTemplate實作訊息發送：

package cn.itcast.mq.spring;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSimpleQueue() {
        // 佇列名稱
        String queueName = "simple.queue";
        // 訊息
        String message = "hello, spring amqp!";
        // 發送訊息
        rabbitTemplate.convertAndSend(queueName, message);
    }
}

訊息接收

在consumer服務的cn.itcast.mq.listener包中新建一個類SpringRabbitListener，代碼如下：

package cn.itcast.mq.listener;

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

@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException {
        System.out.println("spring 消費者接收到訊息：【" + msg + "】");
    }
}

WorkQueue 任務模型

Work模型的使用：

多個消費者系結到一個佇列，同一條訊息只會被一個消費者處理

通過設定prefetch來控制消費者預取的訊息數量

Work queues，也被稱為（Task queues），任務模型，簡單來說就是讓多個消費者系結到一個佇列，共同消費佇列中的訊息，

當訊息處理比較耗時的時候，可能生產訊息的速度會遠遠大于訊息的消費速度，長此以往，訊息就會堆積越來越多，無法及時處理，

此時就可以使用work 模型，多個消費者共同處理訊息處理，速度就能大大提高了，

宣告佇列

@Configuration
public class WorkConfig {
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("simple.queue");
    }
}

訊息發送

這次我們回圈發送，模擬大量訊息堆積現象，

在publisher服務中的SpringAmqpTest類中添加一個測驗方法：

/**
     * workQueue
     * 向佇列中不停發送訊息，模擬訊息堆積，
     */
@Test
public void testWorkQueue() throws InterruptedException {
    // 佇列名稱
    String queueName = "simple.queue";
    // 訊息
    String message = "hello, message_";
    for (int i = 0; i < 50; i++) {
        // 發送訊息
        rabbitTemplate.convertAndSend(queueName, message + i);
        Thread.sleep(20);
    }
}

訊息接收

要模擬多個消費者系結同一個佇列，我們在consumer服務的SpringRabbitListener中添加2個新的方法：

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
    System.out.println("消費者1接收到訊息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(20);
}

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
    System.err.println("消費者2........接收到訊息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(200);
}

注意到這個消費者sleep了1000秒，模擬任務耗時，

發布/訂閱模型

發布訂閱的模型如圖：

可以看到，在訂閱模型中，多了一個exchange角色，而且程序略有變化：

Publisher：生產者，也就是要發送訊息的程式，但是不再發送到佇列中，而是發給X（交換機）
Exchange：交換機，圖中的X，一方面，接收生產者發送的訊息，另一方面，知道如何處理訊息，例如遞交給某個特別佇列、遞交給所有佇列、或是將訊息丟棄，到底如何操作，取決于Exchange的型別，Exchange有以下3種型別：
- Fanout：廣播，將訊息交給所有系結到交換機的佇列
- Direct：定向，把訊息交給符合指定routing key 的佇列
- Topic：通配符，把訊息交給符合routing pattern（路由模式）的佇列
Consumer：消費者，與以前一樣，訂閱佇列，沒有變化
Queue：訊息佇列也與以前一樣，接收訊息、快取訊息，

Exchange（交換機）只負責轉發訊息，不具備存盤訊息的能力，因此如果沒有任何佇列與Exchange系結，或者沒有符合路由規則的佇列，那么訊息會丟失！

Fanout廣播模型

Fanout，英文翻譯是扇出，我覺得在MQ中叫廣播更合適，

在廣播模式下，訊息發送流程是這樣的：

1）可以有多個佇列
2）每個佇列都要系結到Exchange（交換機）
3）生產者發送的訊息，只能發送到交換機，交換機來決定要發給哪個佇列，生產者無法決定
4）交換機把訊息發送給系結過的所有佇列
5）訂閱佇列的消費者都能拿到訊息

我們的計劃是這樣的：

創建一個交換機 itcast.fanout，型別是Fanout
創建兩個佇列fanout.queue1和fanout.queue2，系結到交換機itcast.fanout

宣告佇列和交換機

Spring提供了一個介面Exchange，來表示所有不同型別的交換機：

在接收者consumer中創建一個配置類，宣告佇列和交換機：把系結代碼寫在接收者的代碼上，這樣交換機和佇列可以根據需求系結

@Configuration
public class FanoutConfig {
    /**
     * 宣告交換機
     * @return Fanout型別交換機
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("itcast.fanout");
    }

    /**
     * 第1個佇列
     */
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }

    /**
     * 第2個佇列
     */
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }
}

訊息發送

在publisher服務的SpringAmqpTest類中添加測驗方法：

@Test
public void testFanoutExchange() {
    // 佇列名稱
    String exchangeName = "itcast.fanout";
    // 訊息
    String message = "hello, everyone!";
    rabbitTemplate.convertAndSend(exchangeName, "", message); //注意第二個引數為空字串，且必須要傳
}

訊息接收

@RabbitListener(bindings = @QueueBinding(
    value = https://www.cnblogs.com/buchizicai/archive/2023/02/07/@Queue(name ="fanout.queue1"),
    exchange = @Exchange(name = "itcast.fanout", type = ExchangeTypes.Fanout)
))
public void listenDirectQueue1(String msg){
    System.out.println("消費者接收到fanout.queue1的訊息：【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
    value = https://www.cnblogs.com/buchizicai/archive/2023/02/07/@Queue(name ="fanout.queue2"),
    exchange = @Exchange(name = "itcast.fanout", type = ExchangeTypes.Fanout),
    key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){
    System.out.println("消費者接收到fanout.queue2的訊息：【" + msg + "】");
}

Direct指定模型

描述下Direct交換機與Fanout交換機的差異？

Fanout交換機將訊息路由給每一個與之系結的佇列

Direct交換機根據RoutingKey判斷路由給哪個佇列

如果多個佇列具有相同的RoutingKey，則與Fanout功能類似

基于@RabbitListener注解宣告佇列和交換機有哪些常見注解？

@Queue

@Exchange

在Fanout模式中，一條訊息，會被所有訂閱的佇列都消費，但是，在某些場景下，我們希望不同的訊息被不同的佇列消費，這時就要用到Direct型別的Exchange，

在Direct模型下：

佇列與交換機的系結，不能是任意系結了，而是要指定一個RoutingKey（路由key）
訊息的發送方在向 Exchange發送訊息時，也必須指定訊息的 RoutingKey，
Exchange不再把訊息交給每一個系結的佇列，而是根據訊息的Routing Key進行判斷，只有佇列的Routingkey與訊息的 Routing key完全一致，才會接收到訊息

案例需求如下：

利用@RabbitListener宣告Exchange、Queue、RoutingKey
在consumer服務中，撰寫兩個消費者方法，分別監聽direct.queue1和direct.queue2
在publisher中撰寫測驗方法，向itcast. direct發送訊息

宣告佇列和交換機

在接收者consumer中創建一個配置類，宣告佇列和交換機：把系結代碼寫在接收者的代碼上，這樣交換機和佇列可以根據需求系結

@Configuration
public class DirectConfig {
    /**
     * 宣告交換機
     * @return Direct型別交換機
     */
    @Bean
    public DirectExchange directExchange(){
        return new DirectExchange("itcast.direct");
    }

    /**
     * 第1個佇列
     */
    @Bean
    public Queue directQueue1(){
        return new Queue("direct.queue1");
    }

    /**
     * 第2個佇列
     */
    @Bean
    public Queue directQueue2(){
        return new Queue("direct.queue2");
    }
}

訊息發送

在publisher服務的SpringAmqpTest類中添加測驗方法：

@Test
public void testSendDirectExchange() {
    // 交換機名稱
    String exchangeName = "itcast.direct";
    // 訊息
    String message = "紅色警報！日本亂排核廢水，導致海洋生物變異，驚現哥斯拉！";
    // 發送訊息
    rabbitTemplate.convertAndSend(exchangeName, "red", message);
}

訊息接收

在consumer的SpringRabbitListener類中添加兩個消費者，同時基于注解來宣告佇列和交換機：

@RabbitListener(bindings = @QueueBinding(
    value = https://www.cnblogs.com/buchizicai/archive/2023/02/07/@Queue(name ="direct.queue1"),
    exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
    key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){
    System.out.println("消費者接收到direct.queue1的訊息：【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
    value = https://www.cnblogs.com/buchizicai/archive/2023/02/07/@Queue(name ="direct.queue2"),
    exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
    key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){
    System.out.println("消費者接收到direct.queue2的訊息：【" + msg + "】");
}

Topic 話題模型

描述下Direct交換機與Topic交換機的差異？

Topic交換機接收的訊息RoutingKey必須是多個單詞，以 **.** 分割

Topic交換機與佇列系結時的bindingKey可以指定通配符

#：代表0個或多個詞

*：代表1個詞

Topic型別的Exchange與Direct相比，都是可以根據RoutingKey把訊息路由到不同的佇列，只不過Topic型別Exchange可以讓佇列在系結Routing key 的時候使用通配符！

Routingkey 一般都是有一個或多個單詞組成，多個單詞之間以”.”分割，例如： item.insert

通配符規則：

#：匹配一個或多個詞

*：匹配不多不少恰好1個詞

舉例：

item.#：能夠匹配item.spu.insert 或者 item.spu

item.*：只能匹配item.spu

圖示：

解釋：

Queue1：系結的是china.# ，因此凡是以 china.開頭的routing key 都會被匹配到，包括china.news和china.weather
Queue2：系結的是#.news ，因此凡是以 .news結尾的 routing key 都會被匹配，包括china.news和japan.news

案例需求：

實作思路如下：

并利用@RabbitListener宣告Exchange、Queue、RoutingKey
在consumer服務中，撰寫兩個消費者方法，分別監聽topic.queue1和topic.queue2
在publisher中撰寫測驗方法，向itcast. topic發送訊息

宣告佇列和交換機

在接收者consumer中創建一個配置類，宣告佇列和交換機：把系結代碼寫在接收者的代碼上，這樣交換機和佇列可以根據需求系結

@Configuration
public class TopicConfig {
    /**
     * 宣告交換機
     * @return Topic型別交換機
     */
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange("itcast.topic");
    }

    /**
     * 第1個佇列
     */
    @Bean
    public Queue topicQueue1(){
        return new Queue("topic.queue1");
    }

    /**
     * 第2個佇列
     */
    @Bean
    public Queue topicQueue2(){
        return new Queue("topic.queue2");
    }
}

訊息發送

在publisher服務的SpringAmqpTest類中添加測驗方法：

/**
     * topicExchange
     */
@Test
public void testSendTopicExchange() {
    // 交換機名稱
    String exchangeName = "itcast.topic";
    // 訊息
    String message = "喜報！孫悟空大戰哥斯拉，勝!";
    // 發送訊息
    rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}

訊息接收

在consumer服務的SpringRabbitListener中添加方法：

@RabbitListener(bindings = @QueueBinding(
    value = https://www.cnblogs.com/buchizicai/archive/2023/02/07/@Queue(name ="topic.queue1"),
    exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
    key = "china.#"
))
public void listenTopicQueue1(String msg){
    System.out.println("消費者接收到topic.queue1的訊息：【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
    value = https://www.cnblogs.com/buchizicai/archive/2023/02/07/@Queue(name ="topic.queue2"),
    exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
    key = "#.news"
))
public void listenTopicQueue2(String msg){
    System.out.println("消費者接收到topic.queue2的訊息：【" + msg + "】");
}

訊息轉換器

之前說過，Spring會把你發送的訊息序列化為位元組發送給MQ，接收訊息的時候，還會把位元組反序列化為Java物件，

只不過，默認情況下Spring采用的序列化方式是JDK序列化，眾所周知，JDK序列化存在下列問題：

資料體積過大
有安全漏洞
可讀性差

我們來測驗一下，

測驗默認轉換器

我們修改訊息發送的代碼，發送一個Map物件：

@Test
public void testSendMap() throws InterruptedException {
    // 準備訊息
    Map<String,Object> msg = new HashMap<>();
    msg.put("name", "Jack");
    msg.put("age", 21);
    // 發送訊息
    rabbitTemplate.convertAndSend("simple.queue","", msg);
}

停止consumer服務

發送訊息后查看控制臺：

配置JSON轉換器【重要】

顯然，JDK序列化方式并不合適，我們希望訊息體的體積更小、可讀性更高，因此可以使用JSON方式來做序列化和反序列化，

在publisher和consumer兩個服務中都引入依賴：

<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-xml</artifactId>
    <version>2.9.10</version>
</dependency>

配置訊息轉換器，

發送者和接收者都在啟動類中添加一個Bean即可：

@Bean
public MessageConverter jsonMessageConverter(){
    return new Jackson2JsonMessageConverter();
}

常見問題

訊息佇列在使用程序中，面臨著很多實際問題需要思考：

1. 訊息可靠性問題

如何確保RabbitMQ訊息的可靠性？

開啟生產者確認機制；確保生產者的訊息能到達佇列

開啟持久化功能；確保訊息未消費前在佇列中不會丟失

開啟消費者確認機制為auto；由spring確認訊息處理成功后完成ack

開啟消費者失敗重試機制；并設定MessageRecoverer，多次重試失敗后將訊息投遞到例外交換機，交由人工處理

訊息從發送，到消費者接收，會經歷多個程序：

其中的每一步都可能導致訊息丟失，常見的丟失原因包括：

發送時丟失：
- 生產者發送的訊息未送達exchange
- 訊息到達exchange后未到達queue
MQ宕機，queue將訊息丟失
consumer接收到訊息后未消費就宕機

1.1 生產者確認機制

解決訊息可靠性的訊息發送環節可能會出的問題

RabbitMQ提供了publisher confirm機制來避免訊息發送到MQ程序中丟失，這種機制必須給每個訊息指定一個唯一ID，訊息發送到MQ以后，會回傳一個結果給發送者，表示訊息是否處理成功，

回傳結果有兩種方式：

publisher-confirm，發送者確認
- 訊息成功投遞到交換機，回傳ack
- 訊息未投遞到交換機，回傳nack
publisher-return，發送者回執
- 訊息投遞到交換機了，但是沒有路由到佇列，回傳publisher-confirm的ACK，及路由失敗原因，

注意：

1.1.1 修改配置

首先，修改publisher服務中的application.yml檔案，添加下面的內容：

spring:
  rabbitmq:
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true

說明：

publish-confirm-type：開啟publisher-confirm，這里支持兩種型別：
- simple：同步等待confirm結果，直到超時
- correlated：異步回呼，定義ConfirmCallback，MQ回傳結果時會回呼這個ConfirmCallback
publish-returns：開啟publish-return功能，同樣是基于callback機制，不過是定義ReturnCallback
template.mandatory：定義訊息路由失敗時的策略，true，則呼叫ReturnCallback；false：則直接丟棄訊息

1.1.2 定義Return回呼

publisher-return，發送者回執

每個RabbitTemplate只能配置一個ReturnCallback，因此需要在專案加載時配置：

修改publisher服務，添加一個：

package cn.itcast.mq.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;

@Slf4j
@Configuration
public class CommonConfig implements ApplicationContextAware {//實作該類就是在專案加載時配置
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        // 獲取RabbitTemplate
        RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
        // 設定ReturnCallback
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            // 投遞失敗，記錄日志
            log.info("訊息發送失敗，應答碼{}，原因{}，交換機{}，路由鍵{},訊息{}",
                     replyCode, replyText, exchange, routingKey, message.toString());
            // 如果有業務需要，可以重發訊息
        });
    }
}

1.1.3定義ConfirmCallback

publisher-confirm，發送者確認

ConfirmCallback可以在發送訊息時指定，因為每個業務處理confirm成功或失敗的邏輯不一定相同，

在publisher服務的cn.itcast.mq.spring.SpringAmqpTest類中，定義一個單元測驗方法：

public void testSendMessage2SimpleQueue() throws InterruptedException {
    // 1.訊息體
    String message = "hello, spring amqp!";
    // 2.全域唯一的訊息ID，需要封裝到CorrelationData中
    CorrelationData correlationData = https://www.cnblogs.com/buchizicai/archive/2023/02/07/new CorrelationData(UUID.randomUUID().toString());
    // 3.添加callback
    correlationData.getFuture().addCallback(
        result -> {
            if(result.isAck()){
                // 3.1.ack，訊息成功
                log.debug("訊息發送成功, ID:{}", correlationData.getId());
            }else{
                // 3.2.nack，訊息失敗
                log.error("訊息發送失敗, ID:{}, 原因{}",correlationData.getId(), result.getReason());
            }
        },
        ex -> log.error("訊息發送例外, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
    );
    // 4.發送訊息
    rabbitTemplate.convertAndSend("task.direct", "task", message, correlationData);

    // 休眠一會兒，等待ack回執
    Thread.sleep(2000);
}

1.2 訊息持久化

解決訊息可靠性的 MQ中間可能會出的問題

Spring默認MQ都是持久化的（包括交換機、佇列、訊息等）；本節是告訴客戶端對不需要持久化的訊息可以設為非持久化

生產者確認可以確保訊息投遞到RabbitMQ的佇列中，但是訊息發送到RabbitMQ以后，如果突然宕機，也可能導致訊息丟失，

要想確保訊息在RabbitMQ中安全保存，必須開啟訊息持久化機制，

交換機持久化
佇列持久化
訊息持久化

1.2.1 交換機持久化

RabbitMQ中交換機默認是非持久化的，mq重啟后就丟失，

SpringAMQP中可以通過代碼指定交換機持久化：

@Bean
public DirectExchange simpleExchange(){
    // 三個引數：交換機名稱、是否持久化、當沒有queue與其系結時是否自動洗掉
    return new DirectExchange("simple.direct", true, false);
    //第二個引數：交換機是否持久化，第三個引數：交換機系結的佇列都不存在的時候該交換機自動洗掉
}

可以在RabbitMQ控制臺看到持久化的交換機都會帶上D的標示：

1.2.2 佇列持久化

RabbitMQ中佇列默認是非持久化的，mq重啟后就丟失，

SpringAMQP中可以通過代碼指定交換機持久化：

@Bean
public Queue simpleQueue(){
    // 使用QueueBuilder構建佇列，durable就是持久化的
    return QueueBuilder.durable("simple.queue").build();
}

可以在RabbitMQ控制臺看到持久化的佇列都會帶上D的標示：

1.2.3 訊息持久化

利用SpringAMQP發送訊息時，可以設定訊息的屬性（MessageProperties），指定delivery-mode：

1：非持久化
2：持久化

用java代碼指定：

1.3 消費者訊息確認

解決訊息可靠性的消費者接識訓節可能會出的問題

RabbitMQ是閱后即焚機制，RabbitMQ確認訊息被消費者消費后會立刻洗掉，而RabbitMQ是通過消費者回執來確認消費者是否成功處理訊息的：消費者獲取訊息后，應該向RabbitMQ發送ACK回執，表明自己已經處理訊息，

閱后即焚可能出現的問題：設想這樣的場景

1）RabbitMQ投遞訊息給消費者
2）消費者獲取訊息后，回傳ACK給RabbitMQ
3）RabbitMQ洗掉訊息
4）消費者宕機，訊息尚未處理

這樣，訊息就丟失了，因此消費者回傳ACK的時機非常重要，

而SpringAMQP則允許配置三種確認模式：

?manual：手動ack，需要在業務代碼結束后，呼叫api發送ack，

自己根據業務情況，判斷什么時候該ack

?auto：自動ack，由spring監測listener代碼是否出現例外，沒有例外則回傳ack；拋出例外則回傳nack

auto模式類似事務機制，出現例外時回傳nack，訊息回滾到mq；沒有例外，回傳ack【默認且常用】

?none：關閉ack，MQ假定消費者獲取訊息后會成功處理，因此訊息投遞后立即被洗掉

訊息投遞是不可靠的，可能丟失

1.3.1 演示none模式

修改consumer服務的application.yml檔案，添加下面內容：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: none # 關閉ack

修改consumer服務的SpringRabbitListener類中的方法，模擬一個訊息處理例外：

@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue(String msg) {
    log.info("消費者接收到simple.queue的訊息：【{}】", msg);
    // 模擬例外
    System.out.println(1 / 0);
    log.debug("訊息處理完成！");
}

測驗可以發現，當訊息處理拋例外時，訊息依然被RabbitMQ洗掉了，

1.3.2 演示auto模式

再次把確認機制修改為auto:

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto

在例外位置打斷點，再次發送訊息，程式卡在斷點時，可以發現此時訊息狀態為unack（未確定狀態）：

拋出例外后，因為Spring會自動回傳nack，所以訊息恢復至Ready狀態，并且沒有被RabbitMQ洗掉：

1.4 消費失敗重試機制

解決訊息可靠性的消費者接識訓節后訊息的回收處理問題

當消費者出現例外后，訊息會不斷requeue（重入隊）到佇列，再重新發送給消費者，然后再次例外，再次requeue，無限回圈，導致mq的訊息處理飆升，帶來不必要的壓力：

1.4.1 本地重試

結論：

開啟本地重試時，訊息處理程序中拋出例外，不會requeue到佇列，而是在消費者本地重試

重試達到最大次數后，Spring會回傳ack，訊息會被丟棄

我們可以利用Spring的retry機制，在消費者出現例外時利用本地重試，而不是無限制的requeue到mq佇列，

修改consumer服務的application.yml檔案，添加內容：

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true # 開啟消費者失敗重試
          initial-interval: 1000 # 初識的失敗等待時長為1秒
          multiplier: 1 # 失敗的等待時長倍數，下次等待時長 = multiplier * last-interval
          max-attempts: 3 # 最大重試次數
          stateless: true # true無狀態；false有狀態，如果業務中包含事務，這里改為false

重啟consumer服務，重復之前的測驗，可以發現：

在重試3次后，SpringAMQP會拋出例外AmqpRejectAndDontRequeueException，說明本地重試觸發了
查看RabbitMQ控制臺，發現訊息被洗掉了，說明最后SpringAMQP回傳的是ack，mq洗掉訊息了

1.4.2 失敗策略

本地重試失敗后的訊息可以通過失敗策略回收訊息并傳到指定的服務，該服務一般是人工處理的，

在之前的測驗中，達到最大重試次數后，訊息會被丟棄，這是由Spring內部機制決定的，

在開啟重試模式后，重試次數耗盡，如果訊息依然失敗，則需要有MessageRecovery介面來處理，它包含三種不同的實作：

RejectAndDontRequeueRecoverer：重試耗盡后，直接reject，丟棄訊息，【默認】
ImmediateRequeueMessageRecoverer：重試耗盡后，回傳nack，訊息重新入隊
RepublishMessageRecoverer：重試耗盡后，將失敗訊息投遞到指定的交換機（如下圖）【推薦】

error.queue的訊息在傳送到指定的人工處理客戶端，由人工來處理

比較優雅的一種處理方案是RepublishMessageRecoverer，失敗后將訊息投遞到一個指定的，專門存放例外訊息的佇列，后續由人工集中處理，

圖中consumer系結錯誤交換機的完整代碼：

package cn.itcast.mq.config;

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.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.context.annotation.Bean;

@Configuration
public class ErrorMessageConfig {
    //在consumer服務中定義處理失敗訊息的交換機和佇列
    @Bean
    public DirectExchange errorMessageExchange(){
        return new DirectExchange("error.direct");
    }
    @Bean
    public Queue errorQueue(){
        return new Queue("error.queue", true);
    }
    @Bean
    public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
        return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
    }

    //第三種策略的代碼:本地服務拋出例外到指定的錯誤交換機（定義一個RepublishMessageRecoverer，系結服務到錯誤交換機）
    @Bean
    public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
        return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
    }
}

2. 延遲訊息問題

2.1 前置知識：死信交換機

什么樣的訊息會成為死信？

訊息被消費者reject或者回傳nack
訊息超時未消費
佇列滿了

死信交換機的使用場景是什么？

如果佇列系結了死信交換機，死信會投遞到死信交換機；
可以利用死信交換機收集所有消費者處理失敗的訊息（死信），交由人工處理，進一步提高訊息佇列的可靠性，

【對于例外訊息以及兜底方式，還是建議使用前面失敗策略中的的例外處理交換機】

2.1.1 死信交換機是什么

當一個佇列中的訊息滿足下列情況之一時，可以成為死信（dead letter）：

消費者使用basic.reject或 basic.nack宣告消費失敗，并且訊息的requeue引數設定為false
訊息是一個過期訊息，超時無人消費
要投遞的佇列訊息滿了，無法投遞

如果這個包含死信的佇列配置了dead-letter-exchange屬性，指定了一個交換機，那么佇列中的死信就會投遞到這個交換機中，而這個交換機稱為死信交換機（Dead Letter Exchange，檢查DLX）

如圖，一個訊息被消費者拒絕了，變成了死信；simple.queue系結了死信交換機 dl.direct，因此死信會投遞給這個交換機；如果這個死信交換機也系結了一個佇列，則訊息最侄訓進入這個存放死信的佇列：

另外，佇列將死信投遞給死信交換機時，必須知道兩個資訊：

死信交換機名稱
死信交換機與死信佇列系結的RoutingKey

這樣才能確保投遞的訊息能到達死信交換機，并且正確的路由到死信佇列，

2.1.1 利用死信交換機接收死信

在失敗重試策略中，默認的RejectAndDontRequeueRecoverer會在本地重試次數耗盡后，發送reject給RabbitMQ，訊息變成死信，被丟棄，

我們可以給simple.queue添加一個死信交換機，給死信交換機系結一個佇列，這樣訊息變成死信后也不會丟棄，而是最終投遞到死信交換機，路由到與死信交換機系結的佇列，

我們在consumer服務中，定義一組死信交換機、死信佇列：（這里沒有寫出配置simple的交換機以及佇列）

// 宣告普通的 simple.queue佇列，并且為其指定死信交換機：dl.direct
@Bean
public Queue simpleQueue2(){
    return QueueBuilder.durable("simple.queue") // 指定佇列名稱，并持久化
        .deadLetterExchange("dl.direct") // 指定死信交換機
        .build();
}
// 宣告死信交換機 dl.direct
@Bean
public DirectExchange dlExchange(){
    return new DirectExchange("dl.direct", true, false);
}
// 宣告存盤死信的佇列 dl.queue
@Bean
public Queue dlQueue(){
    return new Queue("dl.queue", true);
}
// 將死信佇列 與 死信交換機系結
@Bean
public Binding dlBinding(){
    return BindingBuilder.bind(dlQueue()).to(dlExchange()).with("simple");
}

2.2 實作延遲訊息方法一：TTL

訊息超時的兩種方式是？

給佇列設定ttl屬性，進入佇列后超過ttl時間的訊息變為死信

給訊息設定ttl屬性，佇列接收到訊息超過ttl時間后變為死信

如何實作發送一個訊息20秒后消費者才收到訊息？

給訊息的目標佇列指定死信交換機

將消費者監聽的佇列系結到死信交換機

發送訊息時給訊息設定超時時間為20秒

一個佇列中的訊息如果超時未消費，則會變為死信，超時分為兩種情況：

當佇列、訊息都設定了TTL時，任意一個到期就會成為死信，

訊息所在的佇列設定了超時時間
訊息本身設定了超時時間

2.2.1 設定接收超時死信的死信交換機

在consumer服務的SpringRabbitListener中，定義一個新的消費者，并且宣告死信交換機、死信佇列：

@RabbitListener(bindings = @QueueBinding(
    value = https://www.cnblogs.com/buchizicai/archive/2023/02/07/@Queue(name ="dl.ttl.queue", durable = "true"),
    exchange = @Exchange(name = "dl.ttl.direct"),
    key = "ttl"
))
public void listenDlQueue(String msg){
    log.info("接收到 dl.ttl.queue的延遲訊息：{}", msg);
}

2.2.2 宣告佇列時，佇列設定TTL

要給佇列設定超時時間，需要在宣告佇列時配置x-message-ttl屬性：

注意，這個佇列設定了死信交換機為dl.ttl.direct

@Bean
public Queue ttlQueue(){
    return QueueBuilder.durable("ttl.queue") // 指定佇列名稱，并持久化
        .ttl(10000) // 設定佇列的超時時間，10秒
        .deadLetterExchange("dl.ttl.direct") // 指定死信交換機
        .build();
}

宣告交換機，將ttl與交換機系結：

@Bean
public DirectExchange ttlExchange(){
    return new DirectExchange("ttl.direct");
}
@Bean
public Binding ttlBinding(){
    return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl");
}

發送者發送訊息，但是不要指定TTL：

@Test
public void testTTLQueue() {
    // 創建訊息
    String message = "hello, ttl queue";
    // 訊息ID，需要封裝到CorrelationData中
    CorrelationData correlationData = https://www.cnblogs.com/buchizicai/archive/2023/02/07/new CorrelationData(UUID.randomUUID().toString());
    // 發送訊息
    rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);
    // 記錄日志
    log.debug("發送訊息成功");
}

發送訊息的日志：

查看下接收訊息的日志：

因為佇列的TTL值是10000ms，也就是10秒，可以看到訊息發送與接收之間的時差剛好是10秒，

2.2.3 發送訊息時，訊息設定TTL

在發送訊息時，也可以指定TTL：

@Test
public void testTTLMsg() {
    // 創建訊息
    Message message = MessageBuilder
        .withBody("hello, ttl message".getBytes(StandardCharsets.UTF_8))
        .setExpiration("5000")
        .build();
    // 訊息ID，需要封裝到CorrelationData中
    CorrelationData correlationData = https://www.cnblogs.com/buchizicai/archive/2023/02/07/new CorrelationData(UUID.randomUUID().toString());
    // 發送訊息
    rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);
    log.debug("發送訊息成功");
}

查看發送訊息日志：

接收訊息日志：

2.3 實作延遲訊息方法二：延遲佇列

利用TTL結合死信交換機，我們實作了訊息發出后，消費者延遲收到訊息的效果，這種訊息模式就稱為延遲佇列（Delay Queue）模式，

延遲佇列的使用場景包括：

延遲發送短信
用戶下單，如果用戶在15 分鐘內未支付，則自動取消
預約作業會議，20分鐘后自動通知所有參會人員

因為延遲佇列的需求非常多，所以RabbitMQ的官方也推出了一個插件，原生支持延遲佇列效果，

這個插件就是DelayExchange插件，參考RabbitMQ的插件串列頁面：https://www.rabbitmq.com/community-plugins.html

使用方式可以參考官網地址：https://blog.rabbitmq.com/posts/2015/04/scheduling-messages-with-rabbitmq

2.3.1 安裝DelayExchange插件

該插件為MQ的拓展，所以安裝插件前MQ的插件目錄需要掛載出去（單機部署里有）

官方的安裝指南地址為：https://blog.rabbitmq.com/posts/2015/04/scheduling-messages-with-rabbitmq

上述檔案是基于linux原生安裝RabbitMQ，然后安裝插件，

1）下載插件

RabbitMQ有一個官方的插件社區，地址為：https://www.rabbitmq.com/community-plugins.html

其中包含各種各樣的插件，包括我們要使用的DelayExchange插件：

大家可以去對應的GitHub頁面下載3.8.9版本的插件，地址為https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/tag/3.8.9這個對應RabbitMQ的3.8.5以上版本，

資料也提供了下載好的插件：

2）上傳插件

因為我們是基于Docker安裝，所以需要先查看RabbitMQ的插件目錄對應的資料卷，如果不是基于Docker的同學，請參考第一章部分，重新創建Docker容器，

我們之前設定的RabbitMQ的資料卷名稱為mq-plugins，所以我們使用下面命令查看資料卷：

docker volume inspect mq-plugins

可以得到下面結果：

接下來，將插件上傳到這個目錄即可：

3）安裝插件

最后就是安裝了，需要進入MQ容器內部來執行安裝，我的容器名為mq，所以執行下面命令：

docker exec -it mq bash

執行時，請將其中的 -it 后面的mq替換為你自己的容器名.

進入容器內部后，執行下面命令開啟插件：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

結果如下：

2.3.2 使用插件DelayExchange

延遲佇列插件的使用步驟包括哪些？

?宣告一個交換機，添加delayed屬性為true

?發送訊息時，添加x-delay頭，值為超時時間

插件的使用也非常簡單：宣告一個交換機，交換機的型別可以是任意型別，只需要設定delayed屬性為true即可，然后宣告佇列與其系結即可，

1）宣告DelayExchange交換機

基于注解方式（推薦）：

也可以基于@Bean的方式：

2）發送訊息

發送訊息時，一定要攜帶x-delay屬性，指定延遲的時間：

3. 訊息堆積問題

當生產者發送訊息的速度超過了消費者處理訊息的速度，就會導致佇列中的訊息堆積，直到佇列存盤訊息達到上限，之后發送的訊息就會成為死信，可能會被丟棄，這就是訊息堆積問題，

解決訊息堆積有兩種思路：

增加更多消費者，提高消費速度（也就是我們之前說的work queue模式）
擴大佇列容積，提高堆積上限

3.1 惰性佇列

訊息堆積問題的解決方案？

佇列上系結多個消費者，提高消費速度

使用惰性佇列，可以再mq中保存更多訊息

惰性佇列的優點有哪些？

基于磁盤存盤，訊息上限高

沒有間歇性的page-out，性能比較穩定

惰性佇列的缺點有哪些？

基于磁盤存盤，訊息時效性會降低

性能受限于磁盤的IO

從RabbitMQ的3.6.0版本開始，就增加了Lazy Queues的概念，也就是惰性佇列，惰性佇列的特征如下：

接收到訊息后直接存入磁盤而非記憶體
消費者要消費訊息時才會從磁盤中讀取并加載到記憶體
支持數百萬條的訊息存盤

3.1.1 基于命令列設定lazy-queue

而要設定一個佇列為惰性佇列，只需要在宣告佇列時，指定x-queue-mode屬性為lazy即可，可以通過命令列將一個運行中的佇列修改為惰性佇列：

rabbitmqctl set_policy Lazy "^lazy-queue$" '{"queue-mode":"lazy"}' --apply-to queues

命令解讀：

rabbitmqctl ：RabbitMQ的命令列工具
set_policy ：添加一個策略
Lazy ：策略名稱，可以自定義
"^lazy-queue$" ：用正則運算式匹配佇列的名字
'{"queue-mode":"lazy"}' ：設定佇列模式為lazy模式
--apply-to queues ：策略的作用物件，是所有的佇列

3.1.2 基于@Bean宣告lazy-queue

3.1.3 基于@RabbitListener宣告LazyQueue

4. 高可用問題 (集群部署)

RabbitMQ集群和ES集群原理一樣

4.1 集群分類

RabbitMQ的是基于Erlang語言撰寫，而Erlang又是一個面向并發的語言，天然支持集群模式，RabbitMQ的集群有兩種模式：

?普通集群：是一種分布式集群，將佇列分散到集群的各個節點，從而提高整個集群的并發能力，

?鏡像集群：是一種主從集群，普通集群的基礎上，添加了主從備份功能，提高集群的資料可用性，

兩種集群的原理

在RabbitMQ的官方檔案中，講述了兩種集群的配置方式的原理：

普通模式：普通模式集群不進行資料同步，每個MQ都有自己的佇列、資料資訊（其它元資料資訊如交換機等會同步），例如我們有2個MQ：mq1，和mq2，如果你的訊息在mq1，而你連接到了mq2，那么mq2會去mq1拉取訊息，然后回傳給你，如果mq1宕機，訊息就會丟失，
鏡像模式：與普通模式不同，佇列會在各個mq的鏡像節點之間同步，因此你連接到任何一個鏡像節點，均可獲取到訊息，而且如果一個節點宕機，并不會導致資料丟失，不過，這種方式增加了資料同步的帶寬消耗，

鏡像集群雖然支持主從，但主從同步并不是強一致的，某些情況下可能有資料丟失的風險，因此在RabbitMQ的3.8版本以后，推出了新的功能：仲裁佇列來代替鏡像集群，底層采用Raft協議確保主從的資料一致性，

4.2 普通集群

一旦主機宕機，佇列將不可用，不具備高可用能力，一般會在普通集群的基礎上再實作鏡像集群或者仲裁集群（推薦仲裁）

4.2.1 集群結構和特征

普通集群，或者叫標準集群（classic cluster），具備下列特征：

會在集群的各個節點間共享部分資料，包括：交換機、佇列元資訊，不包含佇列中的訊息，
當訪問集群某節點時，如果佇列不在該節點，會從資料所在節點傳遞到當前節點并回傳
佇列所在節點宕機，佇列中的訊息就會丟失

結構如圖：

4.2.2 部署

普通模式集群，我們的計劃部署3節點的mq集群：

主機名	控制臺埠	amqp通信埠
mq1	8081 ---> 15672	8071 ---> 5672
mq2	8082 ---> 15672	8072 ---> 5672
mq3	8083 ---> 15672	8073 ---> 5672

集群中的節點標示默認都是：rabbit@[hostname]，因此以上三個節點的名稱分別為：

rabbit@mq1
rabbit@mq2
rabbit@mq3

1）取cookie

RabbitMQ底層依賴于Erlang，而Erlang虛擬機就是一個面向分布式的語言，默認就支持集群模式，集群模式中的每個RabbitMQ 節點使用 cookie 來確定它們是否被允許相互通信，

要使兩個節點能夠通信，它們必須具有相同的共享秘密，稱為Erlang cookie，cookie 只是一串最多 255 個字符的字母數字字符，

每個集群節點必須具有相同的 cookie，實體之間也需要它來相互通信，

我們先在之前啟動的mq容器中獲取一個cookie值，作為集群的cookie，執行下面的命令：

docker exec -it mq cat /var/lib/rabbitmq/.erlang.cookie

可以看到cookie值如下：

FXZMCVGLBIXZCDEMMVZQ

接下來，停止并洗掉當前的mq容器，我們重新搭建集群，

docker rm -f mq

2）準備集群配置

在/tmp目錄新建一個組態檔 rabbitmq.conf：

cd /tmp
# 創建檔案
touch rabbitmq.conf
# 寫入以下內容
vim rabbitmq.conf

檔案內容如下：

loopback_users.guest = false
listeners.tcp.default = 5672
cluster_formation.peer_discovery_backend = rabbit_peer_discovery_classic_config
cluster_formation.classic_config.nodes.1 = rabbit@mq1
cluster_formation.classic_config.nodes.2 = rabbit@mq2
cluster_formation.classic_config.nodes.3 = rabbit@mq3

loopback_users.guest = false：禁用guest用戶，防止黑客侵入

listeners.tcp.default = 5672：監聽埠5672，MQ訊息通信使用的埠

再創建一個檔案，記錄cookie

cd /tmp
# 創建cookie檔案
touch .erlang.cookie
# 寫入cookie
echo "FXZMCVGLBIXZCDEMMVZQ" > .erlang.cookie
# 修改cookie檔案的權限（防止他人修改該檔案）
chmod 600 .erlang.cookie

準備三個目錄,mq1、mq2、mq3：

cd /tmp
# 創建目錄
mkdir mq1 mq2 mq3

然后拷貝rabbitmq.conf、cookie檔案到mq1、mq2、mq3：

# 進入/tmp
cd /tmp
# 拷貝
cp rabbitmq.conf mq1
cp rabbitmq.conf mq2
cp rabbitmq.conf mq3
cp .erlang.cookie mq1
cp .erlang.cookie mq2
cp .erlang.cookie mq3

3）啟動集群

創建一個網路：

docker network create mq-net

運行命令：根據三個不同的mq的組態檔創建三個mq

docker run -d --net mq-net \
-v ${PWD}/mq1/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USER=itcast \
-e RABBITMQ_DEFAULT_PASS=123321 \
--name mq1 \
--hostname mq1 \
-p 8071:5672 \
-p 8081:15672 \
rabbitmq:3.8-management

docker run -d --net mq-net \
-v ${PWD}/mq2/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USER=itcast \
-e RABBITMQ_DEFAULT_PASS=123321 \
--name mq2 \
--hostname mq2 \
-p 8072:5672 \
-p 8082:15672 \
rabbitmq:3.8-management

docker run -d --net mq-net \
-v ${PWD}/mq3/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USER=itcast \
-e RABBITMQ_DEFAULT_PASS=123321 \
--name mq3 \
--hostname mq3 \
-p 8073:5672 \
-p 8083:15672 \
rabbitmq:3.8-management

4.2.3 測驗

在mq1這個節點上添加一個佇列：

如圖，在mq2和mq3兩個控制臺也都能看到：

1）資料共享測驗

點擊這個佇列，進入管理頁面：

然后利用控制臺發送一條訊息到這個佇列：

結果在mq2、mq3上都能看到這條訊息：

2）可用性測驗

我們讓其中一臺節點mq1宕機：

docker stop mq1

然后登錄mq2或mq3的控制臺，發現simple.queue也不可用了：說明資料并沒有拷貝到mq2和mq3

4.3 鏡像集群

普通集群+鏡像集群

默認情況下，佇列只保存在創建該佇列的節點上，而鏡像模式下，創建佇列的節點被稱為該佇列的主節點，佇列還會拷貝到集群中的其它節點，也叫做該佇列的鏡像節點，

但是，不同佇列可以在集群中的任意節點上創建，因此不同佇列的主節點可以不同，甚至，一個佇列的主節點可能是另一個佇列的鏡像節點，

用戶發送給佇列的一切請求，例如發送訊息、訊息回執默認都會在主節點完成，如果是從節點接收到請求，也會路由到主節點去完成，鏡像節點僅僅起到備份資料作用，

當主節點接收到消費者的ACK時，所有鏡像都會洗掉節點中的資料，

總結如下：

鏡像佇列結構是一主多從（從就是鏡像）
所有操作都是主節點完成，然后同步給鏡像節點
主宕機后，鏡像節點會替代成新的主（如果在主從同步完成前，主就已經宕機，可能出現資料丟失）
不具備負載均衡功能，因為所有操作都會有主節點完成（但是不同佇列，其主節點可以不同，可以利用這個提高吞吐量）

4.3.1 集群結構和特征

鏡像集群：本質是主從模式，具備下面的特征：

交換機、佇列、佇列中的訊息會在各個mq的鏡像節點之間同步備份，
創建佇列的節點被稱為該佇列的主節點，備份到的其它節點叫做該佇列的鏡像節點，
一個佇列的主節點可能是另一個佇列的鏡像節點
所有操作都是主節點完成，然后同步給鏡像節點
主宕機后，鏡像節點會替代成新的主

結構如圖：

4.3.2 部署

官方檔案地址：https://www.rabbitmq.com/ha.html

鏡像模式的配置有3種模式：

ha-mode	ha-params	效果
準確模式exactly	佇列的副本量count	集群中佇列副本（主服務器和鏡像服務器之和）的數量，count如果為1意味著單個副本：即佇列主節點，count值為2表示2個副本：1個佇列主和1個佇列鏡像，換句話說：count = 鏡像數量 + 1，如果群集中的節點數少于count，則該佇列將鏡像到所有節點，如果有集群總數大于count+1，并且包含鏡像的節點出現故障，則將在另一個節點上創建一個新的鏡像，
all	(none)	佇列在群集中的所有節點之間進行鏡像，佇列將鏡像到任何新加入的節點，鏡像到所有節點將對所有群集節點施加額外的壓力，包括網路I / O，磁盤I / O和磁盤空間使用情況，推薦使用exactly，設定副本數為（N / 2 +1），
nodes	node names	指定佇列創建到哪些節點，如果指定的節點全部不存在，則會出現例外，如果指定的節點在集群中存在，但是暫時不可用，會創建節點到當前客戶端連接到的節點，

這里我們以rabbitmqctl命令作為案例來講解配置語法，

語法示例：

1）exactly模式

rabbitmqctl set_policy ha-two "^two\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

rabbitmqctl set_policy：固定寫法
ha-two：策略名稱，自定義
"^two\."：匹配佇列的正則運算式，符合命名規則的佇列才生效，這里是任何以two.開頭的佇列名稱
'{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}': 策略內容
- "ha-mode":"exactly"：策略模式，此處是exactly模式，指定副本數量
- "ha-params":2：策略引數，這里是2，就是副本數量為2，1主1鏡像
- "ha-sync-mode":"automatic"：同步策略，默認是manual，即新加入的鏡像節點不會同步舊的訊息，如果設定為automatic，則新加入的鏡像節點會把主節點中所有訊息都同步，會帶來額外的網路開銷

2）all模式

rabbitmqctl set_policy ha-all "^all\." '{"ha-mode":"all"}'

ha-all：策略名稱，自定義
"^all\."：匹配所有以all.開頭的佇列名
'{"ha-mode":"all"}'：策略內容
- "ha-mode":"all"：策略模式，此處是all模式，即所有節點都會稱為鏡像節點

3）nodes模式

rabbitmqctl set_policy ha-nodes "^nodes\." '{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}'

rabbitmqctl set_policy：固定寫法
ha-nodes：策略名稱，自定義
"^nodes\."：匹配佇列的正則運算式，符合命名規則的佇列才生效，這里是任何以nodes.開頭的佇列名稱
'{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}': 策略內容
- "ha-mode":"nodes"：策略模式，此處是nodes模式
- "ha-params":["rabbit@mq1", "rabbit@mq2"]：策略引數，這里指定副本所在節點名稱

4.3.2 測驗

我們使用exactly模式的鏡像，因為集群節點數量為3，因此鏡像數量就設定為2.

運行下面的命令：

docker exec -it mq1 rabbitmqctl set_policy ha-two "^two\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

下面，我們創建一個新的佇列：

在任意一個mq控制臺查看佇列：

1）測驗資料共享

給two.queue發送一條訊息：

然后在mq1、mq2、mq3的任意控制臺查看訊息：

2）測驗高可用

現在，我們讓two.queue的主節點mq1宕機：

docker stop mq1

查看集群狀態：

查看佇列狀態：發現依然是健康的！并且其主節點切換到了rabbit@mq2上

4.4 仲裁佇列

普通集群+仲裁佇列

從RabbitMQ 3.8版本開始，引入了新的仲裁佇列，他具備與鏡像隊里類似的功能，但使用更加方便

4.4.1 集群特征

仲裁佇列：仲裁佇列是3.8版本以后才有的新功能，用來替代鏡像佇列，具備下列特征：

與鏡像佇列一樣，都是主從模式，支持主從資料同步
使用非常簡單，沒有復雜的配置
主從同步基于Raft協議，強一致

4.4.2 網頁方式部署

在任意控制臺添加一個佇列，一定要選擇佇列型別為Quorum型別，

在任意控制臺查看佇列：

可以看到，仲裁佇列的 + 2字樣，代表這個佇列有2個鏡像節點，

因為仲裁佇列默認的鏡像數為5，如果你的集群有7個節點，那么鏡像數肯定是5；而我們集群只有3個節點，因此鏡像數量就是3.

4.4.3 Java代碼方式部署

修改組態檔：SpringAMQP連接MQ集群

注意，這里用address來代替host、port方式

spring:
  rabbitmq:
    addresses: 192.168.150.105:8071, 192.168.150.105:8072, 192.168.150.105:8073
    username: itcast
    password: 123321
    virtual-host: /

Java代碼創建仲裁佇列

@Bean
public Queue quorumQueue() {
    return QueueBuilder
        .durable("quorum.queue") // 持久化
        .quorum() // 仲裁佇列
        .build();
}

4.4.4 測驗

和鏡像集群一樣的測驗方法

4.5 集群擴容

4.5.1 加入集群

1）啟動一個新的MQ容器：

docker run -d --net mq-net \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USER=itcast \
-e RABBITMQ_DEFAULT_PASS=123321 \
--name mq4 \
--hostname mq5 \
-p 8074:15672 \
-p 8084:15672 \
rabbitmq:3.8-management

2）進入容器控制臺：

docker exec -it mq4 bash

3）停止mq行程

rabbitmqctl stop_app

4）重置RabbitMQ中的資料：

rabbitmqctl reset

5）加入mq1：

rabbitmqctl join_cluster rabbit@mq1

6）再次啟動mq行程

rabbitmqctl start_app

4.5.2 增加仲裁佇列副本

我們先查看下quorum.queue這個佇列目前的副本情況，進入mq1容器：

docker exec -it mq1 bash

執行命令：

rabbitmq-queues quorum_status "quorum.queue"

結果：

現在，我們讓mq4也加入進來：

rabbitmq-queues add_member "quorum.queue" "rabbit@mq4"

結果：

再次查看：

rabbitmq-queues quorum_status "quorum.queue"

查看控制臺，發現quorum.queue的鏡像數量也從原來的 +2 變成了 +3：

本文來自博客園，作者：不吃紫菜，遵循CC 4.0 BY-SA著作權協議，轉載請附上原文出處鏈接：https://www.cnblogs.com/buchizicai/p/17093632.html及本宣告，

本文著作權歸作者所有，歡迎轉載，但未經作者同意必須保留此段宣告，且在文章頁面明顯位置給出原文連接，否則保留追究法律責任的權利，

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/543181.html

標籤：其他

上一篇：面向物件編程

下一篇：DataX二次開發——新增HiveReader插件

RabbitMQ訊息佇列入門及解決常見問題

RabbitMQ訊息佇列

同步通訊和異步通訊

同步通訊

異步通訊

常用的訊息佇列組件對比

初步入門

1. 安裝RabbitMQ

1.1 單機部署

1.2 集群部署

2. RabbitMQ訊息模型

3. 匯入Demo工程

4. 入門案例

4.1 publisher實作

4.2 consumer實作

5. 總結

SpringAMQP

基本使用

0. 部署并創建運行RabbitMQ容器

1. 匯入依賴

2. 添加配置yaml

3. 配置訊息轉換器

4. 根據不同的模型寫發送類和接收類

五大模型

Basic Queue 簡單佇列模型

宣告佇列

訊息發送

訊息接收

WorkQueue 任務模型

宣告佇列

訊息發送

訊息接收

Fanout廣播模型

宣告佇列和交換機

訊息發送

訊息接收

Direct指定模型

宣告佇列和交換機

訊息發送

訊息接收

Topic 話題模型

宣告佇列和交換機

訊息發送

訊息接收

訊息轉換器

測驗默認轉換器

配置JSON轉換器【重要】

常見問題

1. 訊息可靠性問題

1.1 生產者確認機制

1.1.1 修改配置

1.1.2 定義Return回呼

1.1.3定義ConfirmCallback

1.2 訊息持久化

1.2.1 交換機持久化

1.2.2 佇列持久化

1.2.3 訊息持久化

1.3 消費者訊息確認

1.3.1 演示none模式

1.3.2 演示auto模式

1.4 消費失敗重試機制

1.4.1 本地重試

1.4.2 失敗策略

2. 延遲訊息問題

2.1 前置知識：死信交換機

2.1.1 死信交換機是什么

2.1.1 利用死信交換機接收死信

2.2 實作延遲訊息方法一：TTL

2.2.1 設定接收超時死信的死信交換機

2.2.2 宣告佇列時，佇列設定TTL

2.2.3 發送訊息時，訊息設定TTL

2.3 實作延遲訊息方法二：延遲佇列

2.3.1 安裝DelayExchange插件

1）下載插件

2）上傳插件

3）安裝插件

2.3.2 使用插件DelayExchange

1）宣告DelayExchange交換機

2）發送訊息

3. 訊息堆積問題