一文看懂RabbitMQ消息丟失如何防止

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

第一種:

生產者弄丟瞭數據。生產者將數據發送到 RabbitMQ 的時候,可能數據就在半路給搞丟瞭,因為網絡問題啥的,都有可能。

第二種:

RabbitMQ 弄丟瞭數據。MQ還沒有持久化自己掛瞭

第三種:

消費端弄丟瞭數據。剛消費到,還沒處理,結果進程掛瞭,比如重啟瞭。

二.RabbitMQ消息丟失解決方案

1.針對生產者

方案1 :開啟RabbitMQ事務

可以選擇用 RabbitMQ 提供的事務功能,就是生產者發送數據之前開啟 RabbitMQ 事務channel.txSelect,然後發送消息,如果消息沒有成功被 RabbitMQ 接收到,那麼生產者會收到異常報錯,此時就可以回滾事務channel.txRollback,然後重試發送消息;如果收到瞭消息,那麼可以提交事務channel.txCommit。

// 開啟事務
channel.txSelect
try {
      // 這裡發送消息
} catch (Exception e) {
      channel.txRollback

// 這裡再次重發這條消息

}

// 提交事務
channel.txCommit

缺點:
RabbitMQ 事務機制是同步的,你提交一個事務之後會阻塞在那兒,采用這種方式基本上吞吐量會下來,因為太耗性能。

方案2: 使用confirm機制 

事務機制和 confirm 機制最大的不同在於,事務機制是同步的,你提交一個事務之後會阻塞在那兒,但是 confirm 機制是異步的

在生產者開啟瞭confirm模式之後,每次寫的消息都會分配一個唯一的id,然後如果寫入瞭rabbitmq之中,rabbitmq會給你回傳一個ack消息,告訴你這個消息發送OK瞭;如果rabbitmq沒能處理這個消息,會回調你一個nack接口,告訴你這個消息失敗瞭,你可以進行重試。而且你可以結合這個機制知道自己在內存裡維護每個消息的id,如果超過一定時間還沒接收到這個消息的回調,那麼你可以進行重發。

//開啟confirm
channel.confirm();
//發送成功回調
public void ack(String messageId){
      
}

// 發送失敗回調
public void nack(String messageId){
//重發該消息
}

2.針對RabbitMQ

說三點:

(1)要保證rabbitMQ不丟失消息,那麼就需要開啟rabbitMQ的持久化機制,即把消息持久化到硬盤上,這樣即使rabbitMQ掛掉在重啟後仍然可以從硬盤讀取消息;

(2)如果rabbitMQ單點故障怎麼辦,這種情況倒不會造成消息丟失,這裡就要提到rabbitMQ的3種安裝模式,單機模式、普通集群模式、鏡像集群模式,這裡要保證rabbitMQ的高可用就要配合HAPROXY做鏡像集群模式

(3)如果硬盤壞掉怎麼保證消息不丟失

(1)消息持久化

RabbitMQ 的消息默認存放在內存上面,如果不特別聲明設置,消息不會持久化保存到硬盤上面的,如果節點重啟或者意外crash掉,消息就會丟失。

所以就要對消息進行持久化處理。如何持久化,下面具體說明下:

要想做到消息持久化,必須滿足以下三個條件,缺一不可。

1) Exchange 設置持久化

2)Queue 設置持久化

3)Message持久化發送:發送消息設置發送模式deliveryMode=2,代表持久化消息

(2)設置集群鏡像模式

我們先來介紹下RabbitMQ三種部署模式:

1)單節點模式:最簡單的情況,非集群模式,節點掛瞭,消息就不能用瞭。業務可能癱瘓,隻能等待。
2)普通模式:消息隻會存在與當前節點中,並不會同步到其他節點,當前節點宕機,有影響的業務會癱瘓,隻能等待節點恢復重啟可用(必須持久化消息情況下)。
3)鏡像模式:消息會同步到其他節點上,可以設置同步的節點個數,但吞吐量會下降。屬於RabbitMQ的HA方案

為什麼設置鏡像模式集群,因為隊列的內容僅僅存在某一個節點上面,不會存在所有節點上面,所有節點僅僅存放消息結構和元數據。下面自己畫瞭一張圖介紹普通集群丟失消息情況:

如果想解決上面途中問題,保證消息不丟失,需要采用HA 鏡像模式隊列。

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

1)同步至所有的
2)同步最多N個機器
3)隻同步至符合指定名稱的nodes

命令處理HA策略模版:rabbitmqctl set_policy [-p Vhost] Name Pattern Definition [Priority]

1)為每個以“rock.wechat”開頭的隊列設置所有節點的鏡像,並且設置為自動同步模式
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)為每個以“rock.wechat.”開頭的隊列設置兩個節點的鏡像,並且設置為自動同步模式
rabbitmqctl set_policy -p rock ha-exacly "^rock.wechat" \
'{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

3)為每個以“node.”開頭的隊列分配指定的節點做鏡像
rabbitmqctl set_policy ha-nodes "^nodes\." \
'{"ha-mode":"nodes","ha-params":["[email protected]", "[email protected]"]}'

但是:HA 鏡像隊列有一個很大的缺點就是:   系統的吞吐量會有所下降

(3)消息補償機制

為什麼還要消息補償機制呢?難道消息還會丟失,沒錯,系統是在一個復雜的環境,不要想的太簡單瞭,雖然以上的三種方案,基本可以保證消息的高可用不丟失的問題,

但是作為有追求的程序員來講,要絕對保證我的系統的穩定性,有一種危機意識。

比如:持久化的消息,保存到硬盤過程中,當前隊列節點掛瞭,存儲節點硬盤又壞瞭,消息丟瞭,怎麼辦?

1)生產端首先將業務數據以及消息數據入庫,需要在同一個事務中,消息數據入庫失敗,則整體回滾。

2)根據消息表中消息狀態,失敗則進行消息補償措施,重新發送消息處理。

3.針對消費者

方案一:ACK確認機制

多個消費者同時收取消息,比如消息接收到一半的時候,一個消費者死掉瞭(邏輯復雜時間太長,超時瞭或者消費被停機或者網絡斷開鏈接),如何保證消息不丟?

使用rabbitmq提供的ack機制,服務端首先關閉rabbitmq的自動ack,然後每次在確保處理完這個消息之後,在代碼裡手動調用ack。這樣就可以避免消息還沒有處理完就ack。才把消息從內存刪除。

這樣就解決瞭,即使一個消費者出瞭問題,但不會同步消息給服務端,會有其他的消費端去消費,保證瞭消息不丟的case。 

三.總結

如果需要保證消息在整條鏈路中不丟失,那就需要生產端、mq自身與消費端共同去保障。

生產端:對生產的消息進行狀態標記,開啟confirm機制,依據mq的響應來更新消息狀態,使用定時任務重新投遞超時的消息,多次投遞失敗進行報警。

mq自身:開啟持久化,並在落盤後再進行ack。如果是鏡像部署模式,需要在同步到多個副本之後再進行ack。

消費端:開啟手動ack模式,在業務處理完成後再進行ack,並且需要保證冪等。

通過以上的處理,理論上不存在消息丟失的情況,但是系統的吞吐量以及性能有所下降。

在實際開發中,需要考慮消息丟失的影響程度,來做出對可靠性以及性能之間的權衡。

到此這篇關於一文看懂RabbitMQ消息丟失如何防止的文章就介紹到這瞭,更多相關RabbitMQ消息丟失如何防止內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!

推薦閱讀: