消息中間件使用廣泛，常用來削峰填谷、系統解耦、異步處理。異步處理可能是使用的最多的場景了，比如現在的技術博客網站，都采用積分制，用戶發表一篇文章后，可以獲取想要的積分，為了提升系統的性能，給用戶加積分的操作可以異步處理，并不需要放在同步流程中。

[[318312]]

我們可以把用戶ID，需要增加的積分封裝成一條消息投遞到消息系統中，異步處理加積分操作，由于這是發生在不同服務器之間，消息有可能投遞失敗、處理失敗等問題，從而導致用戶加積分失敗，還有一種可能是消息重復投遞，那么用戶就有可能重復加積分，不管出現那種情況，都是不正常的情況。

要避免上面的兩種情況，就需要我們盡量保證消息不丟失和消息只被消費一次，這篇文章拋開具體的消息中間件，從消息系統的通用層面上，談談如何避免這兩種情況。

1、保證消息不丟失

一條消息從生產到消費這條鏈路中，有三個地方可能會造成消息丟失，分別如下：

• 消息從生產者寫入到消息隊列的過程投遞失敗。
• 消息在消息隊列中，持久化失敗。
• 消息被消費者消費的過程出現異常。

在消息生產的過程中投遞失敗

消息生產者和消息系統一般都是獨立部署在不同的服務器上，兩臺服務器之間要通信就要通過網絡來完成，網絡是不穩定，可能會發生抖動，那么數據就有可能丟失。網絡發生抖動會有以下兩種情況。

在消息生產的過程中丟失消息

情景一：消息在傳送給消息系統的過程中發生網絡抖動，數據直接丟失。

情景二：消息已經到達消息系統，但是在消息系統給生產者服務器返回信息時，網絡發生抖動，此時的數據不一定真正的丟失，很可能只是生產者認為數據丟失。

針對消息在消息生產時丟失，可以采取重投機制，當程序檢測到網絡異常時，將消息再次投遞到消息系統。但是重新投遞在情景二情況下，可能造成數據重復，如何解決這個問題，在后面會提到。

在消息隊列中持久化失敗

消息系統是可以對消息進行持久化，一般都是將消息存儲到本地磁盤中，當然也有少數消息中間件支持將數據持久化到數據庫中，那么消息系統的性能可能就會下降。

如果你對 Redis 的持久化有一定的了解話，你會發現 Redis 在持久化數據時并不是每新增一條就立即存入到本地磁盤，而是會將數據先寫入到操作系統的 Page Cache 中，當滿足一定條件時，再將 Page Cache 中的數據刷入磁盤，因為這樣可以減少對磁盤的隨機 I/O 操作，我們知道隨機 I/O 是非常耗時的，這樣也提高了系統性能，消息中間件也不例外，在持久化時也是采用這種方式。

在某些極端情況下，可能會造成 Page Cache 中的數據丟失，比如突然停電或者機器異常重啟操作。要解決 Page Cache 中數據丟失問題，可以采用集群部署的方式，來盡量保證數據不丟失。

在消費的過程中存在消息丟失

消息在消費過程中也是會發生丟失的，而且在消費過程中丟失的概率要比前兩種情況大很多。一條消息消費過程大概分成三步：消費者拉取消息，消費者處理消息，消息系統更新消費進度。

圖片描述

第一步在拉取消息的時候可能發生網絡抖動異常，第二步在處理消息的時候可能發生一些業務異常，而導致流程并沒有走完，如果在第一步、第二步發生異常的情況下，通知消息系統更新消費進度，那么這條失敗的消息就永遠不會在被處理了，自然就丟失了，其實我們的業務并沒有跑完。

要避免消息在消費時丟失的情況，可以在消息接收和處理完成之后才更新消費進度，但是在極端的情況下，會出現消息重復消費的問題，比如某一條消息在處理完成之后，消費者宕機了，這時還沒有更新消費進度，消費者重啟后，這條消息還是會被消費到。

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 語句并不會執行成功，從而保證了消息的冪等性。

要保證消息僅被消費一次，我們需要把重點放在消費者這一段，利用等冪性來保證消息被消費一次。

今天站在消息中間件的通用層面上，聊了聊如何保證數據不丟失和僅被消費一次，希望今天的文章對您的學習或者工作有所幫助，如果您認為文章有價值，歡迎點個贊，謝謝。