RabbitMQ — 如何保證消息可靠性?

阿Han

2020 年 8 月 11 日

IPFS

訊息傳遞的過程中有三種可能遺失的情境:

Producer端送到RabbitMQ時丟失:

外界環境問題導致: 發生網路丟包、網路故障等造成訊息丟失。
程式碼層面、配置層面導致訊息丟失。

RabbitMQ儲存的訊息丟失:

訊息沒有持久化。
磁碟意外損壞導致訊息同步失敗。

RabbitMQ送到Consumer時丟失:

消費者接收訊息後還沒來得及處理就當機。

處理方式:

為了確保訊息正確的送達RabbitMQ及Consumer正確的收到訊息，RabbitMQ為我們提供了兩種方式:

透過AMQP Transaction機制實現。
將Channel設定為confirm模式。

AMQP Transaction機制

主要有三個方法分別為:

txSelect(): 將當前的channel設定為transaction模式。
txCommit(): 提交事務。
txRollback(): 回滾事務。

可以看到下面流程中多了四個步驟，如此一來當資料量大的時候處理起來會比較沒有效率，因此一般應用狀況下很少採用Transaction機制。

try {    channel.txSelect();    channel.basicPublish(...);    channel.txCommit();} catch (e) {    channel.txRollback();}

Confirm模式 — 生產端

首先我們一樣將通道設定為Confirm模式，Confirm模式與AMQP Transaction機制只能選一個使用。

channel.confirmSelect();

Confirm模式又分為以下三種方式:

1. 普通Confirm模式: 每發送一條消息就等待服務端回應一次。

這種方式較沒效率，每次丟一個訊息就得等待通知。

try {  // 發送一條訊息  channel.basicPublish(...);  // 等待確認訊號  if(channel.waitForConfirms()){     // 發送成功  } else {     // 重送或其他處置  }} catch(...) {    ...}

2. 批次Confirm模式: 發送一批消息後再等待服務端回應。

這種方式可以發送一批後再等待通知，乍看之下蠻有效率的，但如果訊息常常丟失，那麼我們也得批次重傳。

try {  // 發送多條訊息  for (i = 0; i < batchSize; i++) {     channel.basicPublish(...);  }  // 等待確認訊號  if(channel.waitForConfirms()){     // 發送成功  } else {     // 重送或其他處置  }} catch(...) {    ...}

3. 異步Confirm模式: 以Listener方式等待服務端回應。

這邊以Listener的方式來獲取通知訊息，RabbitMQ會發送以下兩種通知:

Ack: 成功將訊息送到Queue。
Nack: Queue達到上限、MQ異常、磁碟寫滿…等造成未正確將訊息送到Queue。

這邊會有一種狀況是Ack、Nack都沒收到，如網路瞬斷…等，這時候就得搭配其他機制去進行重送。

try {  // 開啟Confirm模式  channel.confirmSelect();  channel.addConfirmListener(new ConfirmListener() {  public void handleAck(long deliveryTag, boolean multiple) throws IOException {         // 成功將訊息送到Queue。  }  public void handleNack(long deliveryTag, boolean multiple) throws IOException {	       // Queue達到上限、MQ異常、磁碟寫滿...等。  }  });} catch(...) {    ...}

以上都不保證100%投送到MQ，只是盡量的保證能夠送達而已，如果對於投送訊息的部份有強烈需求則可以考慮搭配Redis…等DB來維持是否送達的狀態，但勢必會捨棄一些效能。

Confirm模式 — 消費端

消費端的部份就相對單純，處理完訊息後發送ack確認訊號給RabbitMQ即可，RabbitMQ收到後就認為已經消費完成會將該訊息刪除。