使用消息中間件時，如何保證消息僅僅被消費一次？
- 2020 年 3 月 12 日
- 筆記
消息中間件使用廣泛，常用來削峰填谷、系統解耦、非同步處理。非同步處理可能是使用的最多的場景了，比如現在的技術部落格網站，都採用積分制，用戶發表一篇文章後，可以獲取想要的積分，為了提升系統的性能，給用戶加積分的操作可以非同步處理，並不需要放在同步流程中。
我們可以把用戶ID，需要增加的積分封裝成一條消息投遞到消息系統中，非同步處理加積分操作，由於這是發生在不同伺服器之間，消息有可能投遞失敗、處理失敗等問題，從而導致用戶加積分失敗，還有一種可能是消息重複投遞，那麼用戶就有可能重複加積分，不管出現那種情況，都是不正常的情況。
要避免上面的兩種情況，就需要我們盡量保證消息不丟失和消息只被消費一次，這篇文章拋開具體的消息中間件，從消息系統的通用層面上，談談如何避免這兩種情況。
1、保證消息不丟失
一條消息從生產到消費這條鏈路中，有三個地方可能會造成消息丟失，分別如下：
- 消息從生產者寫入到消息隊列的過程投遞失敗。
- 消息在消息隊列中，持久化失敗。
- 消息被消費者消費的過程出現異常。
1.1 在消息生產的過程中投遞失敗
消息生產者和消息系統一般都是獨立部署在不同的伺服器上，兩台伺服器之間要通訊就要通過網路來完成，網路是不穩定，可能會發生抖動，那麼數據就有可能丟失。網路發生抖動會有以下兩種情況。
情景一：消息在傳送給消息系統的過程中發生網路抖動，數據直接丟失。
情景二：消息已經到達消息系統，但是在消息系統給生產者伺服器返回資訊時，網路發生抖動，此時的數據不一定真正的丟失，很可能只是生產者認為數據丟失。
針對消息在消息生產時丟失，可以採取重投機制，當程式檢測到網路異常時，將消息再次投遞到消息系統。但是重新投遞在情景二情況下，可能造成數據重複，如何解決這個問題，在後面會提到。
1.2 在消息隊列中持久化失敗
消息系統是可以對消息進行持久化，一般都是將消息存儲到本地磁碟中，當然也有少數消息中間件支援將數據持久化到資料庫中，那麼消息系統的性能可能就會下降。
如果你對 Redis 的持久化有一定的了解話，你會發現 Redis 在持久化數據時並不是每新增一條就立即存入到本地磁碟，而是會將數據先寫入到作業系統的 Page Cache 中，當滿足一定條件時，再將 Page Cache 中的數據刷入磁碟，因為這樣可以減少對磁碟的隨機 I/O 操作，我們知道隨機 I/O 是非常耗時的，這樣也提高了系統性能，消息中間件也不例外，在持久化時也是採用這種方式。
在某些極端情況下，可能會造成 Page Cache 中的數據丟失，比如突然停電或者機器異常重啟操作。要解決 Page Cache 中數據丟失問題，可以採用集群部署的方式，來盡量保證數據不丟失。
1.3 在消費的過程中存在消息丟失
消息在消費過程中也是會發生丟失的，而且在消費過程中丟失的概率要比前兩種情況大很多。一條消息消費過程大概分成三步：消費者拉取消息，消費者處理消息，消息系統更新消費進度。
第一步在拉取消息的時候可能發生網路抖動異常，第二步在處理消息的時候可能發生一些業務異常，而導致流程並沒有走完，如果在第一步、第二步發生異常的情況下，通知消息系統更新消費進度，那麼這條失敗的消息就永遠不會在被處理了，自然就丟失了，其實我們的業務並沒有跑完。
要避免消息在消費時丟失的情況，可以在消息接收和處理完成之後才更新消費進度，但是在極端的情況下，會出現消息重複消費的問題，比如某一條消息在處理完成之後，消費者宕機了，這時還沒有更新消費進度，消費者重啟後，這條消息還是會被消費到。
2、如何保證消息只被消費一次
消息系統本身不能保證消息僅被消費一次，因為消費本身可能重複、下游系統啟動拉取重複、失敗重試帶來的重複、補償邏輯導致的重複都有可能造重複消息，要保證消息僅被消費一次可以利用等冪性來實現。
等冪是數學上的一個概念，就是多次執行同一個操作和執行一次操作，最終得到的結果是相同的。
從等冪的概念上就可以看出來，就算消息執行多次也不會對系統造成影響，那麼在使用消息系統時如何保證等冪性呢？因為生產者和消費者都有可能產生重複消息，所以要在生產者和消費者兩端都保證等冪性。
保證生產者等冪性，在生產消息的時候，利用雪花演算法給消息生成一個全局 ID，在消息系統中維護消息已 ID 映射關係，如果在映射表中已經存在相同 ID，這丟棄這條消息，雖然消息被投遞了兩次，但是實際上就保存了一條，避免了消息重複問題。
生產者等冪性跟所選者的消息中間件有關係，因為絕大數情況下消息系統不需要我們自己實現，所以等冪性是不太好控制的，消費者等冪性才是我們開發人員控制的重點方向。
在消費者端可以從通用層和業務層兩個方面來做等冪操作，取決於我們的業務要求。
在通用層面中，利用好消息生成是產生的全局唯一ID，消息被處理成功後，把這個全局 ID 存入到數據中，在處理下一條消息之前，先從資料庫中查詢這個全局 ID 是否存在，如果已經存在，則直接放棄該消息。
利用這個全局唯一ID就實現了消息等冪性，偽程式碼如下：
boolean isIDExisted = selectByID(ID); // 判斷ID是否存在 if(isIDExisted) { return; //存在則直接返回 } else { process(message); //不存在，則處理消息 saveID(ID); //存儲ID }但是在極端情況下，這種方式還是會出問題，如果消息在處理之後，還沒來得及保存到資料庫，消費者就宕機重啟了，重啟之後還會再次獲取該消息，執行時查詢該消息並未被消費過，還是會執行兩次消費。可以引入資料庫事務來解決這個問題，但是會降低系統性能。如果對消息重複消費沒有特別嚴格要求的話，直接使用這種沒有引入事務的通用方案就好了，畢竟這也是極小概率的事情。
在業務層面上，我們可選擇性就變多了，比如樂觀鎖、悲觀鎖、記憶體去重（https://github.com/RoaringBitmap/RoaringBitmap）等方法。
我們拿樂觀鎖來舉例，比如我們要給一個用戶加積分，因為加積分操作並不需要放在主業務中，所以就可以使用消息系統來非同步通知，要使用樂觀鎖，就需要給積分表添加一個版本號欄位。並且在生產消息的時候先查詢這個帳號的版本號並且連同消息一起發送到消息系統中。
消費者拿到消息和版本號後，在執行更新積分操作的 SQL 時帶上版本號，類似於：
update score set score = score + 20, version=version+1 where userId=1 and version=1;這條消息消費成功後，version 就變成了 2，那麼如果有重複的 version=1 的消息再次被消費者拉取到，SQL 語句並不會執行成功，從而保證了消息的冪等性。
要保證消息僅被消費一次，我們需要把重點放在消費者這一段，利用等冪性來保證消息被消費一次。
今天站在消息中間件的通用層面上，聊了聊如何保證數據不丟失和僅被消費一次，希望今天的文章對您的學習或者工作有所幫助，如果您認為文章有價值，歡迎點個贊，謝謝。
最後
目前互聯網上很多大佬都有消息中間件相關文章，如有雷同，請多多包涵了。原創不易，碼字不易，還希望大家多多支援。若文中有所錯誤之處，還望提出，謝謝。
