RabbitMQ服務異步通信

消息隊列在使用過程中，面臨著很多實際問題需要思考：

在這里插入圖片描述

1. 消息可靠性

消息從發送，到消費者接收，會經理多個過程：

在這里插入圖片描述

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

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

針對這些問題，RabbitMQ分別給出了解決方案：

生產者確認機制
mq持久化
消費者確認機制
失敗重試機制

下面我們就通過案例來演示每一個步驟。

創建Demo工程

在這里插入圖片描述

1.1. 生產者消息確認

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

返回結果有兩種方式：

publisher-confirm，發送者確認
- 消息成功投遞到交換機，返回ack
- 消息為投遞到交換機，返回nack
publisher-return，發送者回執
- 消息投遞到交換機了，但是沒有路由到隊列，返回ack及路由失敗原因

在這里插入圖片描述

注意：確認機制發送消息時，需要給每個消息設置一個全局唯一id，以區分不同消息，避免ack沖突

1.1.1. 修改配置

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

spring:rabbitmq:publisher-confirm-type: correlatedpublisher-returns: truetemplate:mandatory: true

說明：

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

1.1.2. 定義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 {@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {// 獲取RabbitTemplate對象RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);// 設置ReturnCallbackrabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {// 投遞失敗，記錄日志log.info("消息發送失敗，應答碼{}，原因{}，交換機{}，路由鍵{},消息{}",replyCode, replyText, exchange, routingKey, message);});}
}

1.1.3. 定義ConfirmCallback

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

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

package cn.itcast.mq.spring;import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.connection.CorrelationData;
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;import java.util.UUID;@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testSendMessage2SimpleQueue() throws InterruptedException {String routingKey = "simple";// 準備消息String message = "hello, spring amqp!";// 全局唯一的消息ID，需要封裝到CorrelationData中CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());// 添加callbackcorrelationData.getFuture().addCallback(result -> {assert result != null;if (result.isAck()) {// ack消息成功log.debug("消息發送成功，消息ID：{}", correlationData.getId());} else {// nack消息失敗log.error("消息發送失敗，消息ID：{}，原因:{}", correlationData.getId(), result.getReason());}},ex -> log.error("消息發送異常，消息ID：{}，原因:{}", correlationData.getId(), ex.getMessage()));// 發送消息rabbitTemplate.convertAndSend("amq.topic", routingKey, message, correlationData);// 休眠一會兒，等待回調Thread.sleep(2000);}
}

1.2. 消息持久化

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

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

交換機持久化
隊列持久化
消息持久化

1.2.1. 交換機持久化

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

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

@Bean
public DirectExchange simpleExchange(){// 三個參數：交換機名稱、是否持久化、當沒有queue與其綁定時是否自動刪除return new DirectExchange("simple.direct", true, false);
}

事實上，默認情況下，由SpringAMQP聲明的交換機都是持久化的。

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

在這里插入圖片描述

1.2.2. 隊列持久化

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

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

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

事實上，默認情況下，由SpringAMQP聲明的隊列都是持久化的。

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

在這里插入圖片描述

1.2.3. 消息持久化

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

1：非持久化
2：持久化

用java代碼指定：

在這里插入圖片描述

默認情況下，SpringAMQP發出的任何消息都是持久化的，不用特意指定。

1.3. 消費者消息確認

RabbitMQ是閱后即焚機制，RabbitMQ確認消息被消費者消費后會立刻刪除。

而RabbitMQ是通過消費者回執來確認消費者是否成功處理消息的：消費者獲取消息后，應該向RabbitMQ發送ACK回執，表明自己已經處理消息。

設想這樣的場景：

1）RabbitMQ投遞消息給消費者
2）消費者獲取消息后，返回ACK給RabbitMQ
3）RabbitMQ刪除消息
4）消費者宕機，消息尚未處理

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

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

manual：手動ack，需要在業務代碼結束后，調用api發送ack。
auto：自動ack，由spring監測listener代碼是否出現異常，沒有異常則返回ack；拋出異常則返回nack
none：關閉ack，MQ假定消費者獲取消息后會成功處理，因此消息投遞后立即被刪除

由此可知：

none模式下，消息投遞是不可靠的，可能丟失
auto模式類似事務機制，出現異常時返回nack，消息回滾到mq；沒有異常，返回ack
manual：自己根據業務情況，判斷什么時候該ack

一般，我們都是使用默認的auto即可。

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 # 關閉ack

在異常位置打斷點，再次發送消息，程序卡在斷點時，可以發現此時消息狀態為unacked（未確定狀態）：

在這里插入圖片描述

拋出異常后，因為Spring會自動返回nack，所以消息恢復至Ready狀態，并且沒有被RabbitMQ刪除：

在這里插入圖片描述

1.4. 消費失敗重試機制

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

在這里插入圖片描述

怎么辦呢？

1.4.1. 本地重試

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

修改consumer服務的application.yml文件，添加內容：

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

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

在重試3次后，SpringAMQP會拋出異常AmqpRejectAndDontRequeueException，說明本地重試觸發了
查看RabbitMQ控制臺，發現消息被刪除了，說明最后SpringAMQP返回的是ack，mq刪除消息了

結論：

開啟本地重試時，消息處理過程中拋出異常，不會requeue到隊列，而是在消費者本地重試
重試達到最大次數后，Spring會返回ack，消息會被丟棄

1.4.2. 失敗策略

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

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

RejectAndDontRequeueRecoverer：重試耗盡后，直接reject，丟棄消息。默認就是這種方式
ImmediateRequeueMessageRecoverer：重試耗盡后，返回nack，消息重新入隊
RepublishMessageRecoverer：重試耗盡后，將失敗消息投遞到指定的交換機

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

1）在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");
}

2）定義一個RepublishMessageRecoverer，關聯隊列和交換機

@Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
}

完整代碼：

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 {@Beanpublic DirectExchange errorMessageExchange(){return new DirectExchange("error.direct");}@Beanpublic Queue errorQueue(){return new Queue("error.queue", true);}@Beanpublic Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");}@Beanpublic MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");}
}

1.5. 總結

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

開啟生產者確認機制，確保生產者的消息能到達隊列
開啟持久化功能，確保消息未消費前在隊列中不會丟失
開啟消費者確認機制為auto，由spring確認消息處理成功后完成ack
開啟消費者失敗重試機制，并設置MessageRecoverer，多次重試失敗后將消息投遞到異常交換機，交由人工處理

2. 死信交換機

2.1. 初始死信交換機

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.2. 利用死信交換機接收死信

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

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

在這里插入圖片描述

我們在consumer服務中，定義一組死信交換機、死信隊列：

// 聲明普通的 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.1.3. 總結

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

消息被消費者reject或者返回nack
消息超時未消費
隊列滿了

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

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

2.2. TTL

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

消息所在的隊列設置了超時時間
消息本身設置了超時時間

在這里插入圖片描述

2.2.1.接收超時死信的死信交換機

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

@RabbitListener(bindings = @QueueBinding(value = @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屬性：

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

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

聲明交換機，將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 = 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 = new CorrelationData(UUID.randomUUID().toString());// 發送消息rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);log.debug("發送消息成功");
}

查看發送消息日志：

在這里插入圖片描述

接收消息日志：

在這里插入圖片描述

這次，發送與接收的延遲只有5秒。說明當隊列、消息都設置了TTL時，任意一個到期就會成為死信。

2.2.4. 總結

消息超時的兩種方式是？

給隊列設置ttl屬性，進入隊列后超過ttl時間的消息變為死信
給消息設置ttl屬性，隊列接收到消息超過ttl時間后變為死信

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

給消息的目標隊列指定死信交換機
將消費者監聽的隊列綁定到死信交換機
發送消息時給消息設置超時時間為20秒

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插件

1）上傳插件

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

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

docker volume inspect mq-plugins

可以得到下面結果：

在這里插入圖片描述

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

2）安裝插件

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

docker exec -it mq bash

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

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

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

結果如下：

在這里插入圖片描述

3）使用插件

DelayExchange插件的原理是對官方原生的Exchange做了功能的升級：

將DelayExchange接收到的消息暫存在內存中（官方的Exchange是無法存儲消息的）
在DelayExchange中計時，超時后才投遞消息隊列中

在RabbitMQ的管理平臺聲明一個DelayExchange：

在這里插入圖片描述

消息的延遲時間需要在發送消息的時候指定：

在這里插入圖片描述

2.3.2. DelayExchange原理

DelayExchange需要將一個交換機聲明為delayed類型。當我們發送消息到delayExchange時，流程如下：

接收消息
判斷消息是否具備x-delay屬性
如果有x-delay屬性，說明是延遲消息，持久化到硬盤，讀取x-delay值，作為延遲時間
返回routing not found結果給消息發送者
x-delay時間到期后，重新投遞消息到指定隊列

2.3.3. 使用DelayExchange

插件的使用也非常簡單：聲明一個交換機，交換機的類型可以是任意類型，只需要設定delayed屬性為true即可，然后聲明隊列與其綁定即可。

1）聲明DelayExchange交換機

基于注解方式（推薦）：

在這里插入圖片描述

也可以基于@Bean的方式：

在這里插入圖片描述

2）發送消息

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

在這里插入圖片描述

2.3.4. 總結

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

聲明一個交換機，添加delayed屬性為true
發送消息時，添加x-delay頭，值為超時時間

3. 惰性隊列

3.1. 消息堆積問題

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

在這里插入圖片描述

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

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

要提升隊列容積，把消息保存在內存中顯然是不行的。

3.2. 惰性隊列

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

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

3.2.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.2.2. 基于@Bean聲明lazy-queue

在這里插入圖片描述

3.2.3. 基于@RabbitListener聲明LazyQueue

在這里插入圖片描述

3.3. 總結

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

隊列上綁定多個消費者，提高消費速度
使用惰性隊列，可以再mq中保存更多消息

惰性隊列的優點有哪些？

基于磁盤存儲，消息上限高
沒有間歇性的page-out，性能比較穩定

惰性隊列的缺點有哪些？

基于磁盤存儲，消息時效性會降低
性能受限于磁盤的IO

4. MQ集群

4.1. 集群分類

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

普通集群：是一種分布式集群，將隊列分散到集群的各個節點，從而提高整個集群的并發能力。
鏡像集群：是一種主從集群，普通集群的基礎上，添加了主從備份功能，提高集群的數據可用性。

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

4.2. 普通集群

4.2.1. 集群結構和特征

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

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

結構如圖：

在這里插入圖片描述

4.2.2. 部署

1）集群分類

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

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

我們先來看普通模式集群，我們的計劃部署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

2）獲取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

3）準備集群配置

在/tmp目錄新建一個配置文件 rabbitmq.conf：

cd /tmp
# 創建文件
touch 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

再創建一個文件，記錄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

4）啟動集群

創建一個網絡：

docker network create mq-net

docker volume create

運行命令

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

5）測試

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

在這里插入圖片描述

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

在這里插入圖片描述

6）數據共享測試

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

在這里插入圖片描述

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

在這里插入圖片描述

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

在這里插入圖片描述

7）可用性測試

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

docker stop mq1

然后登錄mq2或mq3的控制臺，發現simple.queue也不可用了：

在這里插入圖片描述

說明數據并沒有拷貝到mq2和mq3。

4.3. 鏡像集群

4.3.1.集群結構和特征

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

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

結構如圖：

在這里插入圖片描述

4.3.2. 部署

在剛剛的案例中，一旦創建隊列的主機宕機，隊列就會不可用。不具備高可用能力。如果要解決這個問題，必須使用官方提供的鏡像集群方案。

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

1）鏡像模式的特征

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

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

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

當主節點接收到消費者的ACK時，所有鏡像都會刪除節點中的數據。

總結如下：

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

2）鏡像模式的配置

鏡像模式的配置有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"]：策略參數，這里指定副本所在節點名稱

3）測試

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

運行下面的命令：

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

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

在這里插入圖片描述

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

在這里插入圖片描述

4）測試數據共享

給two.queue發送一條消息：

在這里插入圖片描述

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

在這里插入圖片描述

5）測試高可用

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

docker stop mq1

查看集群狀態：

在這里插入圖片描述

查看隊列狀態：

在這里插入圖片描述

發現依然是健康的！并且其主節點切換到了rabbit@mq2上。

4.4. 仲裁隊列

4.4.1. 集群特征

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

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

4.4.2. 部署

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

1）添加仲裁隊列

在任意控制臺添加一個隊列，一定要選擇隊列類型為Quorum類型。

在這里插入圖片描述

在任意控制臺查看隊列：

在這里插入圖片描述

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

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

2）測試

可以參考對鏡像集群的測試，效果是一樣的。

3）集群擴容

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

在這里插入圖片描述

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：

在這里插入圖片描述

4.4.3. Java代碼創建仲裁隊列

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

4.4.4. SpringAMQP連接MQ集群

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

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