一、數(shù)據(jù)丟失場景

MQ 消息數(shù)據(jù)完整的鏈路為：從 Producer 發(fā)送消息到 RabbitMQ 服務(wù)器中，再由 Broker 服務(wù)的 Exchange 根據(jù) Routing_Key 路由到指定的 Queue 隊列中，最后投送到消費者中完成消費。

所以消息在上面三個節(jié)點都可能存在消息丟失的情況：

• 生產(chǎn)者丟失消息：生產(chǎn)者將消息發(fā)送到服務(wù)器過程中，由于網(wǎng)絡(luò)問題或服務(wù)器問題可能會導(dǎo)致消息發(fā)送失敗而導(dǎo)致消息丟失；
• MQ 隊列丟失消息：消息是存放在 MQ 服務(wù)器的消息隊列中的，但由于 MQ 服務(wù)故障導(dǎo)致崩潰或服務(wù)重啟，就可能會導(dǎo)致消息隊列中的數(shù)據(jù)丟失；
• 消費者丟失消息：消費者收到消息后，處理過程中可能因為程序出錯導(dǎo)致消息的消費失敗，或中途消費者掛了導(dǎo)致消息沒有完成消費，這些都會導(dǎo)致消息丟失。

二、數(shù)據(jù)可靠性方案

上面已經(jīng)了解到了消息數(shù)據(jù)可能丟失的環(huán)節(jié)，所以，我們需要針對每個環(huán)節(jié)進行處理，以防止數(shù)據(jù)的丟失。

1、生產(chǎn)者丟失消息解決方案

對于生產(chǎn)者消息丟失的問題，我們有常用的兩種方案：

• 開啟消息發(fā)送事務(wù)功能；
• 開啟 Confirm 消息確認(rèn)機制。

1-1、開啟消息發(fā)送事務(wù)功能

我們可以選擇使用 RabbitMQ 提供的事務(wù)功能：生產(chǎn)者在發(fā)送數(shù)據(jù)之前開啟事物，然后再發(fā)送消息。

如果消息沒有成功被 RabbitMQ 接收到，那么生產(chǎn)者會受到異常報錯，這時就可以回滾事務(wù)，然后嘗試重新發(fā)送；如果收到了消息，那么就可以提交事務(wù)。

偽代碼如下：

channel.txSelect();// 開啟事物
try{
	...
}catch(Exection e){
    channel.txRollback()；// 回滾事物
    // 重新提交
}

這種方案有個比較大的缺點：RabbitMQ 事務(wù)一旦開啟，就會變?yōu)橥阶枞僮?，生產(chǎn)者會阻塞等待是否發(fā)送成功，由于比較耗性能而會造成吞吐量的下降。所以并不推薦這種方案。

1-2、開啟 Confirm 消息確認(rèn)機制

在生產(chǎn)者中開啟了Confirm 模式，為每次寫的消息分配一個唯一的 ID，然后再發(fā)送給 RabbitMQ 服務(wù)

• 如果成功寫入到了 RabbitMQ 之中，RabbitMQ 會給你回傳一個 ACK 消息，告訴你這個消息發(fā)送 OK 了；
• 如果 RabbitMQ 沒能處理這個消息，就會回調(diào)你一個 NACK 接口，告訴你這個消息失敗了，你可以進行重試。

同時也可以結(jié)合這個機制知道自己在內(nèi)存里維護每個消息的 ID，如果超過一定時間還沒接收到這個消息的回調(diào)，那么可以嘗試進行重發(fā)。

偽代碼如下：

//開啟confirm
channel.confirm();
//發(fā)送成功回調(diào)
public void ack(String messageId){
}
// 發(fā)送失敗回調(diào)
public void nack(String messageId){
    //重發(fā)該消息
}

由于事務(wù)機制是同步阻塞的，而 Confirm 機制是異步的，在發(fā)送消息之后可以接著發(fā)送下一個消息，最后通過 RabbitMQ 的回調(diào)告知成功與否，所以，生產(chǎn)者消息丟失方案一般都是采用 Confirm 確認(rèn)機制。

2、MQ 隊列丟失消息解決方案

對于 MQ 隊列丟失消息的問題，我們可以開啟消息的持久化，當(dāng)然隊列本身也要開啟持久化，畢竟隊列如果不存在了，哪怕消息持久化也沒有用。

開啟了消息隊列的持久化后，可以將消息的持久化和生產(chǎn)者的 Confirm 機制配合起來，只有消息持久化到了磁盤，才會個生產(chǎn)者發(fā)送 ACK，這樣就算是在持久化之前 RabbitMQ 掛了，數(shù)據(jù)丟了，生產(chǎn)者收不到 ACK 回調(diào)也會進行消息重發(fā)。

持久化有個關(guān)鍵的問題需要注意：

消息在正確存入 RabbitMQ 之后，還需要有一段時間（這個時間很短，但不可忽視）才能存入磁盤之中。因為 RabbitMQ 并不是為每條消息都做 fsync 的處理，可能僅僅保存到 cache 中而不是物理磁盤上，在這段時間內(nèi) RabbitMQ 的 broker 發(fā)生 crash，消息保存到 cache 但是還沒來得及落盤，那么這些消息將會丟失。

解決這個問題的方案是 RabbitMQ 開啟鏡像隊列，鏡像隊列相當(dāng)于配置了副本，當(dāng) master 在此特殊時間內(nèi) crash 掉，可以自動切換到 slave，這樣有效地保障了數(shù)據(jù)的丟失。

3、消費者丟失消息解決方案

針對消費者丟失消息問題，我們可以使用 RabbitMQ 提供的 ACK 應(yīng)答機制，首先需要將 自動應(yīng)答標(biāo)志位 autoAck 設(shè)置為 false 來關(guān)閉 RabbitMQ 的自動ack，這是為了防止 Consumer 收到消息后，還沒來得及處理完成就 crash 掉了。所以我們采用手動應(yīng)答的方式：

String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;

然后在消費者執(zhí)行完畢之后手動應(yīng)答： channel.basicAck 。

總結(jié)

RabbitMQ 消息的可靠性涉及 producer 端的確認(rèn)機制、broker 服務(wù)的持久化與鏡像隊列的配置、consumer 端的確認(rèn)機制。

要想確保消息的可靠性越高，那么性能也會隨之而降，所以需要根據(jù)實際情況進行選擇和取舍。

到此這篇關(guān)于RabbitMQ之消息的可靠性方案詳解的文章就介紹到這了,更多相關(guān)RabbitMQ可靠性方案內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論