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