一.RabbitMQ消息丟失的三種情況

第一種：生產(chǎn)者弄丟了數(shù)據(jù)。生產(chǎn)者將數(shù)據(jù)發(fā)送到 RabbitMQ 的時(shí)候，可能數(shù)據(jù)就在半路給搞丟了，因?yàn)榫W(wǎng)絡(luò)問(wèn)題啥的，都有可能。

第二種：RabbitMQ 弄丟了數(shù)據(jù)。MQ還沒有持久化自己掛了。

第三種：消費(fèi)端弄丟了數(shù)據(jù)。剛消費(fèi)到，還沒處理，結(jié)果進(jìn)程掛了，比如重啟了。

二.RabbitMQ消息丟失解決方案

1.針對(duì)生產(chǎn)者

方案1 ：開啟RabbitMQ事務(wù)

可以選擇用 RabbitMQ 提供的事務(wù)功能，就是生產(chǎn)者發(fā)送數(shù)據(jù)之前開啟 RabbitMQ 事務(wù)channel.txSelect，然后發(fā)送消息，如果消息沒有成功被 RabbitMQ 接收到，那么生產(chǎn)者會(huì)收到異常報(bào)錯(cuò)，此時(shí)就可以回滾事務(wù)channel.txRollback，然后重試發(fā)送消息；如果收到了消息，那么可以提交事務(wù)channel.txCommit。

// 開啟事務(wù)  
channel.txSelect();  
try {  
   // 這里發(fā)送消息  
} catch (Exception e) {  
   channel.txRollback(); 
// 這里再次重發(fā)這條消息
}
// 提交事務(wù)  
channel.txCommit();  

缺點(diǎn)：

RabbitMQ 事務(wù)機(jī)制是同步的，你提交一個(gè)事務(wù)之后會(huì)阻塞在那兒，采用這種方式基本上吞吐量會(huì)下來(lái)，因?yàn)樘男阅堋?/p>

方案2：使用confirm機(jī)制

事務(wù)機(jī)制和 confirm 機(jī)制最大的不同在于，事務(wù)機(jī)制是同步的，你提交一個(gè)事務(wù)之后會(huì)阻塞在那兒，但是 confirm 機(jī)制是異步的

在生產(chǎn)者開啟了confirm模式之后，每次寫的消息都會(huì)分配一個(gè)唯一的id，然后如果寫入了rabbitmq之中，rabbitmq會(huì)給你回傳一個(gè)ack消息，告訴你這個(gè)消息發(fā)送OK了；如果rabbitmq沒能處理這個(gè)消息，會(huì)回調(diào)你一個(gè)nack接口，告訴你這個(gè)消息失敗了，你可以進(jìn)行重試。而且你可以結(jié)合這個(gè)機(jī)制知道自己在內(nèi)存里維護(hù)每個(gè)消息的id，如果超過(guò)一定時(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ā)該消息  
}

2.針對(duì)RabbitMQ

主要需要應(yīng)對(duì)三點(diǎn)：

• 要保證rabbitMQ不丟失消息，那么就需要開啟rabbitMQ的持久化機(jī)制，即把消息持久化到硬盤上，這樣即使rabbitMQ掛掉在重啟后仍然可以從硬盤讀取消息；

• 如果rabbitMQ單點(diǎn)故障怎么辦，這種情況倒不會(huì)造成消息丟失，這里就要提到rabbitMQ的3種安裝模式，單機(jī)模式、普通集群模式、鏡像集群模式，這里要保證rabbitMQ的高可用就要配合HAPROXY做鏡像集群模式；

• 如果硬盤壞掉怎么保證消息不丟失。

（1）消息持久化

RabbitMQ 的消息默認(rèn)存放在內(nèi)存上面，如果不特別聲明設(shè)置，消息不會(huì)持久化保存到硬盤上面的，如果節(jié)點(diǎn)重啟或者意外crash掉，消息就會(huì)丟失。

所以就要對(duì)消息進(jìn)行持久化處理。如何持久化，下面具體說(shuō)明下。要想做到消息持久化，必須滿足以下三個(gè)條件，缺一不可。

• Exchange 設(shè)置持久化

• Queue 設(shè)置持久化

• Message持久化發(fā)送：發(fā)送消息設(shè)置發(fā)送模式deliveryMode=2，代表持久化消息

（2）設(shè)置集群鏡像模式

先來(lái)介紹下RabbitMQ三種部署模式：

• 單節(jié)點(diǎn)模式：最簡(jiǎn)單的情況，非集群模式，節(jié)點(diǎn)掛了，消息就不能用了。業(yè)務(wù)可能癱瘓，只能等待。 

• 普通模式：消息只會(huì)存在與當(dāng)前節(jié)點(diǎn)中，并不會(huì)同步到其他節(jié)點(diǎn)，當(dāng)前節(jié)點(diǎn)宕機(jī)，有影響的業(yè)務(wù)會(huì)癱瘓，只能等待節(jié)點(diǎn)恢復(fù)重啟可用（必須持久化消息情況下）。 

• 鏡像模式：消息會(huì)同步到其他節(jié)點(diǎn)上，可以設(shè)置同步的節(jié)點(diǎn)個(gè)數(shù)，但吞吐量會(huì)下降。屬于RabbitMQ的HA方案

為什么設(shè)置鏡像模式集群，因?yàn)殛?duì)列的內(nèi)容僅僅存在某一個(gè)節(jié)點(diǎn)上面，不會(huì)存在所有節(jié)點(diǎn)上面，所有節(jié)點(diǎn)僅僅存放消息結(jié)構(gòu)和元數(shù)據(jù)。下面畫了一張圖介紹普通集群丟失消息情況：

如果想解決上面途中問(wèn)題，保證消息不丟失，需要采用HA 鏡像模式隊(duì)列。

下面介紹下三種HA策略模式：

• 同步至所有的

• 同步最多N個(gè)機(jī)器

• 只同步至符合指定名稱的nodes

命令處理HA策略模版：

rabbitmqctl set_policy [-p Vhost] Name Pattern Definition [Priority]

1）為每個(gè)以rock.wechat開頭的隊(duì)列設(shè)置所有節(jié)點(diǎn)的鏡像，并且設(shè)置為自動(dòng)同步模式

rabbitmqctl set_policy ha-all "^rock.wechat" '{"ha-mode":"all","ha-sync-mode":"automatic"}'  
rabbitmqctl set_policy -p rock ha-all "^rock.wechat" '{"ha-mode":"all","ha-sync-mode":"automatic"}'

2）為每個(gè)以rock.wechat.開頭的隊(duì)列設(shè)置兩個(gè)節(jié)點(diǎn)的鏡像，并且設(shè)置為自動(dòng)同步模式

rabbitmqctl set_policy -p rock ha-exacly "^rock.wechat" \
'{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

3）為每個(gè)以node.開頭的隊(duì)列分配指定的節(jié)點(diǎn)做鏡像

rabbitmqctl set_policy ha-nodes "^nodes\." \
'{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}'

但是：HA 鏡像隊(duì)列有一個(gè)很大的缺點(diǎn)就是系統(tǒng)的吞吐量會(huì)有所下降。

（3）消息補(bǔ)償機(jī)制

為什么還要消息補(bǔ)償機(jī)制呢？難道消息還會(huì)丟失，沒錯(cuò)，系統(tǒng)是在一個(gè)復(fù)雜的環(huán)境，不要想的太簡(jiǎn)單了，雖然以上的三種方案，基本可以保證消息的高可用不丟失的問(wèn)題。

但是作為有追求的程序員來(lái)講，要絕對(duì)保證我的系統(tǒng)的穩(wěn)定性，有一種危機(jī)意識(shí)。

比如：持久化的消息，保存到硬盤過(guò)程中，當(dāng)前隊(duì)列節(jié)點(diǎn)掛了，存儲(chǔ)節(jié)點(diǎn)硬盤又壞了，消息丟了，怎么辦？

1）生產(chǎn)端首先將業(yè)務(wù)數(shù)據(jù)以及消息數(shù)據(jù)入庫(kù)，需要在同一個(gè)事務(wù)中，消息數(shù)據(jù)入庫(kù)失敗，則整體回滾。

2）根據(jù)消息表中消息狀態(tài)，失敗則進(jìn)行消息補(bǔ)償措施，重新發(fā)送消息處理。

3.針對(duì)消費(fèi)者

方案一：ACK確認(rèn)機(jī)制

多個(gè)消費(fèi)者同時(shí)收取消息，比如消息接收到一半的時(shí)候，一個(gè)消費(fèi)者死掉了(邏輯復(fù)雜時(shí)間太長(zhǎng)，超時(shí)了或者消費(fèi)被停機(jī)或者網(wǎng)絡(luò)斷開鏈接)，如何保證消息不丟？

使用rabbitmq提供的ack機(jī)制，服務(wù)端首先關(guān)閉rabbitmq的自動(dòng)ack，然后每次在確保處理完這個(gè)消息之后，在代碼里手動(dòng)調(diào)用ack。這樣就可以避免消息還沒有處理完就ack。才把消息從內(nèi)存刪除。

這樣就解決了，即使一個(gè)消費(fèi)者出了問(wèn)題，但不會(huì)同步消息給服務(wù)端，會(huì)有其他的消費(fèi)端去消費(fèi)，保證了消息不丟的case。

總結(jié)

如果需要保證消息在整條鏈路中不丟失，那就需要生產(chǎn)端、mq自身與消費(fèi)端共同去保障。

• 生產(chǎn)端：對(duì)生產(chǎn)的消息進(jìn)行狀態(tài)標(biāo)記，開啟confirm機(jī)制，依據(jù)mq的響應(yīng)來(lái)更新消息狀態(tài)，使用定時(shí)任務(wù)重新投遞超時(shí)的消息，多次投遞失敗進(jìn)行報(bào)警。

• mq自身：開啟持久化，并在落盤后再進(jìn)行ack。如果是鏡像部署模式，需要在同步到多個(gè)副本之后再進(jìn)行ack。

• 消費(fèi)端：開啟手動(dòng)ack模式，在業(yè)務(wù)處理完成后再進(jìn)行ack，并且需要保證冪等。

通過(guò)以上的處理，理論上不存在消息丟失的情況，但是系統(tǒng)的吞吐量以及性能有所下降。在實(shí)際開發(fā)中，需要考慮消息丟失的影響程度，來(lái)做出對(duì)可靠性以及性能之間的權(quán)衡。

以上就是RabbitMQ消息丟失解決方案的詳細(xì)內(nèi)容，更多關(guān)于RabbitMQ 消息丟失的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論