RabbitMQ 訊息怎麼才不會掉？Ack、Prefetch、延遲重試與死信佇列

訊息可能在哪裡掉？

Exchange、Queue、Binding 的基本關係可以先參考 RabbitMQ 的 Exchange、Queue、Binding 這篇。一則訊息從發出到處理完，會經過三段，每一段都可能出事：

第一段：確定訊息真的送到了

BasicPublish 呼叫成功，只代表訊息被寫進了 TCP 連線，不代表 Broker 收到了。要確認 Broker 真的接手，需要開啟 Publisher Confirm：

channel.ConfirmSelect();

channel.BasicPublish("house.events", "house.created.franchise", props, body);

// 等待 Broker 確認；逾時或被拒絕會拋出例外，由呼叫端決定重送或記錄
channel.WaitForConfirmsOrDie(TimeSpan.FromSeconds(5));

每發一筆就同步等待確認，吞吐量會下降很多。訊息量大時，可以改成批次發送後再一起等待確認，或是監聽 BasicAcks／BasicNacks 事件做非同步確認。

第二段：Broker 重啟也不能掉

RabbitMQ 預設把訊息放在記憶體裡。要讓訊息在 Broker 重啟後還在，需要三個條件同時成立：

• Exchange 是 Durable：重啟後 Exchange 本身還在
• Queue 是 Durable：重啟後 Queue 本身還在
• 訊息是 Persistent（DeliveryMode = 2）：訊息會寫到磁碟

channel.ExchangeDeclare("house.events", ExchangeType.Topic, durable: true);
channel.QueueDeclare("franchise-sync", durable: true, exclusive: false, autoDelete: false);

var props = channel.CreateBasicProperties();
props.Persistent = true;

只設定了 Durable Queue、卻忘了把訊息設成 Persistent，是最常見的漏洞：Queue 重啟後還在，但裡面是空的。

第三段：Consumer 端的 Ack

自動 Ack 的問題

BasicConsume 的 autoAck: true 代表訊息一送出給 Consumer，Broker 就把它刪掉。如果 Consumer 拿到訊息後、處理完成前當掉了，這則訊息就永遠消失了。

手動 Ack

改成 autoAck: false，處理成功之後才回覆 BasicAck。Consumer 如果在 Ack 之前斷線，Broker 會把這則訊息重新放回佇列，交給其他 Consumer：

var consumer = new EventingBasicConsumer(channel);
consumer.Received += (_, ea) =>
{
    try
    {
        Handle(ea.Body.ToArray());
        channel.BasicAck(ea.DeliveryTag, multiple: false);
    }
    catch (Exception ex)
    {
        // 失敗怎麼處理？見下一節
    }
};
channel.BasicConsume("franchise-sync", autoAck: false, consumer);

手動 Ack 帶來一個必然的結果：同一則訊息可能被處理不只一次（例如處理完成、Ack 送出前剛好斷線）。所以 Consumer 的處理邏輯要設計成冪等的——例如寫入 Elasticsearch 時用物件 ID 當文件 ID 做 Upsert，處理兩次的結果和一次相同。

Prefetch：一次拿幾則

Prefetch（BasicQos）限制「一個 Consumer 手上最多可以有幾則還沒 Ack 的訊息」：

channel.BasicQos(prefetchSize: 0, prefetchCount: 10, global: false);

• 不設定（無上限）：Broker 會把佇列裡的訊息盡可能全部推給第一個連上的 Consumer，後來加入的 Consumer 可能分不到工作；Consumer 的記憶體也可能被大量訊息塞爆
• 設太小（例如 1）：每處理完一則才會拿到下一則，網路往返的時間會拖慢吞吐量
• 一般做法：從 10～50 開始，依處理時間與 Consumer 數量調整。處理越慢的工作，Prefetch 越小，訊息才能平均分散到每個 Consumer

失敗了怎麼辦：不要無限 Requeue

處理失敗時，最直覺的做法是 BasicNack(requeue: true) 把訊息放回佇列。但如果失敗的原因是資料本身有問題（例如欄位格式錯誤），這則訊息會馬上又被拿出來、又失敗、又放回去，形成無限迴圈，CPU 飆高、日誌被洗版，後面的正常訊息也一直被它插隊。

失敗要分成兩種：

• 暫時性錯誤：Elasticsearch 短暫無回應、網路抖動、資料庫鎖定逾時——等一下再試就會成功
• 永久性錯誤：資料格式錯誤、找不到關聯資料、程式邏輯無法處理——試幾次都一樣

對應的做法是延遲重試 + 死信佇列。

延遲重試與死信佇列

RabbitMQ 沒有內建「延遲 30 秒再投遞」的功能（除非安裝 Delayed Message 外掛），但可以用 TTL + 死信交換器（Dead Letter Exchange, DLX） 組出來：

1. 宣告一個 retry-queue，設定訊息 TTL 為 30 秒，並把它的 DLX 指向主 Exchange。這個佇列沒有任何 Consumer
2. 處理失敗時，Consumer 先 Ack 原訊息，再把它連同 x-retry-count + 1 的標頭發到 retry-queue
3. 30 秒後訊息過期，RabbitMQ 依 DLX 設定把它送回主 Exchange，重新進入主佇列再處理一次
4. 重試次數超過上限（例如 3 次），或判斷是永久性錯誤，就直接送進 dead-letter-queue，不再重試

// 重試佇列：TTL 到期後死信回主 Exchange
channel.QueueDeclare("franchise-sync.retry", durable: true, exclusive: false, autoDelete: false,
    arguments: new Dictionary<string, object>
    {
        ["x-message-ttl"] = 30_000,
        ["x-dead-letter-exchange"] = "house.events"
    });

// 死信佇列：停止重試，等人處理
channel.QueueDeclare("franchise-sync.dlq", durable: true, exclusive: false, autoDelete: false);

回到主 Exchange 時，訊息會以原本的 Routing Key 重新路由。如果主 Exchange 上還有其他佇列綁定了相同的 Routing Key，它們也會再收到一次。比較乾淨的做法是另外建立一個只綁定主佇列的「重試回流 Exchange」，或在死信時用 x-dead-letter-routing-key 指定只有主佇列會收到的 Routing Key。

死信佇列不是垃圾桶。進到這裡的訊息代表「系統自己處理不了」，要搭配監控：佇列長度大於 0 就告警，由工程師檢查原因、修正資料或程式後，再把訊息重新發回主佇列。

使用情境：加盟平台的物件同步

加盟平台的同步服務從 franchise-sync 佇列取出物件事件，轉換格式後寫入 Elasticsearch 與 MSSQL：

• Elasticsearch 叢集重啟或短暫過載：寫入逾時屬於暫時性錯誤，進入延遲重試，30 秒後通常就恢復了。重試期間主佇列的其他訊息照常處理，不會被這一筆卡住
• 主力平台送來的物件缺少必要欄位：屬於永久性錯誤，直接送進死信佇列並告警，不會在系統裡無限重試
• 同步服務部署更新、Pod 重啟：使用手動 Ack，處理到一半的訊息會回到佇列，由新的 Pod 接手
• 同一則事件被處理兩次：以物件 ID 為文件 ID 做 Upsert，處理兩次結果一樣

一句話總結

訊息可靠性要三段一起顧：Producer 端用 Publisher Confirm 確認送達、Broker 端用 Durable 加 Persistent 撐過重啟、Consumer 端用手動 Ack 確保處理完才刪除；失敗時區分暫時性與永久性錯誤，前者用 TTL 加死信做延遲重試，後者送進死信佇列告警處理，而且因為訊息可能被處理不只一次，Consumer 一定要是冪等的。