【Message Queue - RabbitMQ】 如何保證消息可靠性?

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

Producer端送到RabbitMQ時丟失:

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

RabbitMQ儲存的訊息丟失:

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

RabbitMQ送到Consumer時丟失:

處理方式:

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

AMQP Transaction機制

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

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

Confirm模式 — 生產端

channel.confirmSelect();

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

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

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

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

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

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(...) {
    ...
}

Confirm模式 — 消費端