消息服務中如何確保消息至少被消費一次

對消息服務須要了解的朋友，能夠移步：

1. 聊聊mq的使用場景
2. 聊聊業務系統中投遞消息到mq的幾種方式
3. 談談mq消息消費的幾種方式

本章討論主題

1. 如何確保消息至少消費一次，確保消費者最大程度消費成功

消費者消費消息有2中方式：

1. push方式

消息服務接收到消息以後，主動將消息推送給消費者消費

2. pull方式

消費者定時從消息服務中拉取消息進行消費

下面咱們將討論2中方式中如何確保消息至少被消費一次。

push模式

消費的過程：

1. 消息服務查詢待消費的消息列表
2. 輪詢待消息列表
3. 調用消費者
4. 消費者收到消費請求，執行業務處理，將處理結果返回給消息服務
5. 消息服務接收到消費成功的信息，將消息狀態置爲消費成功狀態
6. 繼續消費下一條消息

探討一下上面須要考慮的問題：

若消息一直消費失敗如何處理？

先說一下影響：

1. 消息被阻塞

   消息若是一直消費失敗，消息服務會不斷調用消費者進行消費，會阻塞其餘消息的消費，直接影響到業務的正常進行.

消費失敗的緣由：

1. 代碼問題

   這種狀況無論嘗試多少次，消息都會消費失敗，須要人工介入修復bug，這個能夠依靠監控系統發現bug，同時開發進行修復。

2. 系統運行異常

   如調用超時、網絡問題等一些不可控的因素。產生這種錯誤，繼續重試，最終會處理成功。

此處我們只用討論消息服務中重試機制如何設計？

系統異常狀況下，可能過一段時間，系統恢復了，此時去重試，消費也就成功了。

因此咱們對於消費失敗的消息採用延遲處理的方式，能夠這麼實現：

消息中增長几個字段用於重試：next_dispose_time【下次處理時間】、max_failure【最大容許失敗次數】、failure【當前失敗次數】，消息入庫時：next_dispose_time=需消費的時間，max_failure = 運行最大失敗次數, failure=0;

當消費失敗時，處理過程：

1. 計算下次處理時間（next_dispose_time），能夠在當前時間上面作指數遞增，好比根據失敗次數依次在當前時間上遞增2的failure次方秒，如：

   第1次失敗：當前時間 + 2秒

   第2次失敗：當前時間 + 4秒

   第3次失敗：當前時間 + 8秒

   第4次失敗：當前時間 + 16秒

   .......

   第n次失敗：當前時間 + 2的n次方秒

2. failure++

消息服務查詢待消費的消息也須要作調整：

select * from 消息表 where next_dispose_time<=當前時間 and failure<max_failure and status = 待處理;

此時可以最大程度保證消息最少消費成功一次。

pull方式

這種會複雜一些，爲什麼會複雜一些，我們先看一下常規的流程：

1. 消費者從消息服務中拉取消息
2. 本地進行處理
3. 從消息服務中刪除此消息
4. 繼續拉取下一條進行處理

若是本地一直處理失敗，那麼後面拉取到的都是同一條消息，這條消息直接阻塞後續消息的消費，這種狀況如何解？

我們先分析一下出現這種問題的後果及緣由：

1. 後果：消息被阻塞，業務沒法正常運行
2. 緣由：代碼問題或其餘異常
3. 確保代碼沒問題，能夠解決上面問題，及時性不夠高，線上要考慮系統的容錯能力。

遇到這種問題仍是挺嚴重了，業務方都是沒法接受的，一條消息消費失敗，會影響到其餘全部消息的消費，這個咱們仍是得想辦法解決，能夠這樣：

1. 消費者拉取消息
2. 落地到本地
3. 從消息服務中刪除此消息
4. 異步去消費本地落地的消息

消息先落地，而後異步處理，本地須要有個補償的job，去處理本地消費失敗的消息，這個能夠參考push方式消費的過程。