RabbitMQ如何保證消息可靠性？
本篇文章不再介紹RabbitMQ具體實現原理，直接介紹如何保證消息的可靠性問題。所謂可靠性，指消息不重不漏。
文章導讀
圖片
生產者消費者模型
生產者-消費者模型用于描述兩類進程（生產者和消費者）之間的數據交互。可以被認為是獨立的服務，生產者負責生成數據，消費者負責處理這些數據。在分布式系統中，隊列在其中扮演了消息（數據）傳遞的功能。
圖片
關于消息隊列的作用，一般解讀為：
解耦：生產者和消費者獨立運作，無需知道對方的運行狀態。
異步：并非實時，生產者不必關注消費端的消費情況。
削峰：限制流量，防止消費者過載。
消息丟失
這其實不難理解，就像生活中下單-快遞-簽收的過程。這個過程和上邊的生產者-消費者模型恰有異曲同工之妙。
圖片
這個過程中，
- 下單用戶（生產者）
- 快遞小哥（隊列）
- 簽收人（消費者）
- 快件（消息）
如果包裹被粗略的認為是一條消息，那么快件在郵寄過程中丟失了，就是消息丟失。快件從發貨到簽收，我們不用去關心中間發生了什么。但是要是沒收到貨，那得給我個理由。
如何排查？
就上邊的快件丟失問題，怎么知道快遞為何沒有收到？很簡單，一段一段的排查：
- 商家是否有發貨？
- 快遞公司是否攬收？
- 查看快遞小哥是否放入代收點
相應的，如果生產環境中突然發現諸如：告警、服務宕機、數據流轉異常等問題時，我們也會在鏈路上（A、B、C三處）逐一排查。
產生原因及解決方案
1、生產端可靠性投遞
為確保消息從生產端可靠地投遞到RabbitMQ，我們需要考慮以下幾個關鍵點：
網絡故障：消息可能在傳輸過程中因網絡問題而丟失。
RabbitMQ故障：如果RabbitMQ宕機，消息也可能丟失。
圖片
對應解決方案：
- 開啟事務機制
事務在RabbitMQ中可能會影響性能，因為它們需要在所有節點上同步狀態。因此，RabbitMQ盡量避免使用事務。核心代碼：
private static void executeTransaction(Channel channel) throws IOException {
boolean transactionSuccess = false;
try {
// 開啟事務
channel.txSelect();
// 執行一系列消息操作，例如：channel.basicPublish(exchange, routingKey, message);
// 提交事務
channel.txCommit();
transactionSuccess = true;
} catch (ShutdownSignalException | IOException e) {
// 回滾事務
if (!transactionSuccess) {
channel.txRollback();
}
throw e;
}
}- 生產者確認機制
發布者確認機制允許發布者知道消息是否已經被RabbitMQ成功接收：
public static void sendPersistentMessage(String host, String queueName, String message) {
try (Connection connection = new ConnectionFactory().setHost(host).newConnection();
Channel channel = connection.createChannel()) {
// 啟用發布者確認
channel.confirmSelect();
// 將消息設置為持久化
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
.deliveryMode(2)
.build();
// 添加確認監聽器
channel.addConfirmListener(new ConfirmListener() {
@Override
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
System.out.println("消息已確認: " + deliveryTag);
// 消息正確到達Broker時的處理邏輯
}
@Override
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
System.out.println("消息未確認: " + deliveryTag);
// 因為內部錯誤導致消息丟失時的處理邏輯
}
});
channel.basicPublish("", queueName, properties, message.getBytes());
// 等待消息確認，或者超時
boolean allConfirmed = channel.waitForConfirms();
if (allConfirmed) {
//所有消息都已確認
} else {
//超時或其它
}
} catch (IOException | TimeoutException | InterruptedException e) {
e.printStackTrace();
}
}2、消息持久化
在RabbitMQ中，消息的持久化它確保消息不僅存儲在內存中，而且也安全地保存在磁盤上。這樣，即使在RabbitMQ服務崩潰或重啟的情況下，消息也不會丟失，可以從磁盤恢復。
圖片
消息到達RabbitMQ后通過Exchange交換機，路由給queue隊列，最后發送給消費端。
圖片
從RabbitMQ設計上看，消息的持久化應該從以下方面入手：
- Exchange持久化：
// 設置 durable = true;
channel.exchangeDeclare(exchangeName, "direct", durable);- 消息持久化：
// 設置 MessageProperties.PERSISTENT_TEXT_PLAIN
channel.basicPublish(exchangeName, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());- Queue持久化：
//設置 boolean durable = true;
channel.queueDeclare(queueName, durable, exclusive, false, null);這樣，如果RabbitMQ收到消息后掛了，重啟后會自行從磁盤上恢復消息。
3、消費者確認機制
如果上述生產端、消息隊列都正確投遞，那么問題出現在消費端是否可以正確消費？
圖片
消費者在成功處理了一條消息后通知RabbitMQ，這樣RabbitMQ在收到確認后才會移除隊列中的消息。
默認情況下，以下3種原因導致消息丟失：
1、 網絡故障：消費端還沒接收到消息之前，發生網絡故障導致消息丟失；
2、 未接收消息前服務宕機：消費端突然掛機未接收到消息，此時消息會丟失；
3、 處理過程中服務宕機：消費端正確接收到消息，但在處理消息的過程中發生異常或宕機了，消息也會丟失。
這是因為RabbitMQ的自動ack機制，即默認RabbitMQ在消息發出后，不管消費端是否接收到，是否處理完，就立即刪除這條消息，導致消息丟失。
圖片
應對方案：
- 將自動ack機制改為手動ack機制。
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
try {
//接收消息，業務處理
//設置手動確認
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
} catch (Exception e) {
//發生異常時，可以選擇重新發送消息或進行錯誤處理
// 例如，可以選擇負確認（nack），讓消息重回隊列
// channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
}
};
//設置autoAck為false，表示關閉自動確認機制，改為手動確認
channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> {});4、消息補償機制
以上3種解決辦法理論上可靠，但是系統的異常或者故障比較偶然，我們沒法做到100%消息不丟失。因此需要介入補償機制或者人工干預。這是我們的最后一道防線。
如何做消息補償呢？其實就是將消息入庫，通過定時任務重新發送失敗的消息。詳細流程如下：
圖片
- 生產端發送消息;
- 確認失敗，將消息保存到數據庫中，并設置初始狀態0;
- 定時任務以一定頻率掃描數據庫中status=0 的消息（失敗消息）;
- 重發消息，可多次;
- 重發成功，更新數據庫：status=1;
- 超過固定次數重發仍然失敗，人工干預。
標注：
超過最大失敗次數后，對于無法被正常消費的消息可移入死信隊列。
- 可人工干預手動排查
- 也可自動重試，需要實現一個消費者來從死信隊列中獲取消息，并根據業務邏輯來決定是否以及如何重新發送消息。這里涉及到消息去重、冪等性處理等。
以上，我們知道了消息丟失問題如何處理？那么對于消息重復的問題，下面做個介紹。
消息重復消費
消息重復消費是指在消息隊列中，同一條消息被不同的消費者多次消費處理。
產生原因：
- 網絡問題：消費者處理完消息后，因網絡問題導致確認信息未能成功發送回消息隊列。
- 服務中斷：消費者在確認消息之前服務崩潰，消息隊列未收到確認信號。
- 確認機制：自動確認模式下，如果確認在消息處理完成前發生，消息可能會被重復消費
對應解決方案：
1. 冪等性設計
設計消費者的消息處理邏輯時，要保證即使消息被多次消費，也不會對系統狀態產生不良影響。冪等性可以通過以下方式實現：
- 數據庫唯一約束：使用數據庫的主鍵約束或唯一索引防止插入重復記錄。
- 業務邏輯檢查：在執行業務操作前，先檢查是否已經處理過該消息。
2. 消息去重策略
使用唯一標識符（如訂單號、massageID）來識別消息，并在消費者中實現去重邏輯：
- 緩存檢查：使用內存緩存（如Redis）存儲已處理的消息ID。
- 持久化存儲：將消息ID與處理狀態保存在數據庫中，以便跨服務重啟后仍然有效。
3. 手動確認與重試機制
通過手動確認消息，控制消息何時從隊列中移除：
- 手動確認：在消息成功處理后，顯式調用channel.basicAck()方法確認消息。
- 重試機制：如果消息處理失敗，可以選擇將消息重新入隊（channel.basicReject(requeue=true)）或丟棄（channel.basicReject(requeue=false)）。
代碼演示:
消費者端去重邏輯
@RabbitListener(queues = "queueName", acknowledgeMode = "MANUAL")
public void receiveMessage(Message message, Channel channel) throws IOException {
String messageId = message.getMessageProperties().getMessageId();
// 檢查消息是否已消費
if (messageAlreadyProcessed(messageId)) {
// 消息已消費，確認消息并返回
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
return;
}
// 處理消息
try {
processMessage(message);
// 消息處理成功，持久化消息ID并確認消息
persistMessageId(messageId);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
// 處理失敗，可以選擇重新入隊或丟棄
boolean requeue = shouldRequeue(message);
channel.basicReject(message.getMessageProperties().getDeliveryTag(), requeue);
}
}生產者端發布確認
void sendWithConfirm(AmqpTemplate amqpTemplate, Message message) throws IOException {
ConfirmCallback confirmCallback = (correlationData, ack, cause) -> {
if (!ack) {
// 處理消息發送失敗的邏輯
// ...
}
};
amqpTemplate.setConfirmCallback(confirmCallback);
amqpTemplate.convertAndSend("exchangeName", "routingKey", message);
}具體實現需要根據實際業務邏輯和RabbitMQ配置進行調整。
總結
以上介紹了RabbitMQ保證消息可靠性的問題、產生原因、解決方案等。不足之處，歡迎指正。
