【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模式 — 消費端