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

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

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

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

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

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

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

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

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

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

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

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

偽代碼如下：

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

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

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

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

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

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

偽代碼如下：

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

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

2、MQ 隊(duì)列丟失消息解決方案

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

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

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

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

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

3、消費(fèi)者丟失消息解決方案

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

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

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

總結(jié)

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

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

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

最新評(píng)論