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 保證消息可靠性的問題、產生原因、解決方案等。不足之處，歡迎指正。
本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/OuOk6QBnGbm1TlWvpoO6Uw