訊息推送平臺有沒有保證資料不丟？

我們在使用mq的時候，就會很自然思考一個問題：怎么保證資料不丟失？

現在austin接入層是把訊息發到mq，下發邏輯層從mq消費資料，隨后呼叫對應渠道介面來下發訊息，

訊息推送平臺??推送下發【郵件】【短信】【微信服務號】【微信小程式】【企業微信】【釘釘】等訊息型別，

• https://gitee.com/zhongfucheng/austin/
• https://github.com/ZhongFuCheng3y/austin

訊息丟棄一般我們考慮的是消費端，于是重點看的是下發邏輯層，

（因為對于mq使用方來說：生產端只要配置mq相關的引數，在呼叫下發時有回呼重試機制，那就足夠了，生產端能做的東西確實不多）

目前為止，下發邏輯層(消費端)使用的是自動提交offset策略，只要消費端存在系統重啟或者行程被kill掉，那就會有丟訊息的情況，

spring.kafka.consumer.enable-auto-commit=true

當前下發邏輯層(消費端)有可能放大了這個丟棄訊息的問題，因為現在是消費到mq資料后，會把訊息給到執行緒池去處理，執行緒池會指定一個阻塞佇列，那佇列數量越大，可能由重啟所丟棄的訊息就越多，

這里我的策略是：當應用重啟的時候，系統里的執行緒池是優雅關閉的(盡可能等待一段時間，等阻塞佇列里沒有訊息了，再關閉執行緒池)，

但回到問題的本質上，只要消費端是自動提交offset策略，就一定會有丟訊息的問題，所以要做到消費端的訊息不丟，我們就要設定為手動提交offset，這個是必要條件，

有沒有必要保證不丟

在探討具體的技術實作方案之前，我們來看看在業務上有沒有必要保證訊息不丟，我剛接觸到訊息推送平臺的時候，當時那個交接的哥們告訴我和我學長：訊息少發比多發要好，

1、重要的訊息用戶很可能會手動重試觸發，

austin是一個發送各類渠道訊息的平臺，從我的經驗來說，這里面最重要的是短信渠道，經過austin下發很可能是登陸驗證碼，銀行卡提現驗證碼，這類訊息從全域上看是最重要的，

而其他渠道，例如push通知欄的通知訊息，微信渠道的營銷訊息，這種訊息即便用戶沒收到，也不會對用戶帶來很大的使用體驗問題，這種訊息或許對絕大數用戶都是無感知的（少發幾條，用戶可能更樂意），

我們先假設用戶的某一次銀行卡提現的驗證碼恰好因為我們重啟系統而丟棄，這時候，絕大數用戶可能懷疑自己的信號問題，會繼續操作，重新發送一次，

(因為客服經常找我排查這種問題，每次都能看到有好幾條下發記錄，當然了，能到技術的，99%的問題都不是由系統重啟丟失訊息導致的，更多可能是用戶的客戶端本身確實就存在問題)

2、訊息是有時效性的，比如驗證碼這種短信一般就5min的時效性，由于系統的問題，你超過這個時間給用戶發送，對用戶的體驗是非常差的，

3、訊息推送平臺是有全鏈路追蹤的，是可以知道下發的訊息有沒有到達到用戶手上，至少都可以知道在我們的系統內部執行程序中有沒有丟，如果這條訊息真的那么重要，那可以單獨為丟棄的訊息單獨做重發處理，這些功能在訊息推送平臺都是支持的，

這個問題我以前的同事也跟我探討過，就是把上面的內容給我隔壁的老哥聽的，他說：你就盡扯淡吧，到面試的時候人家可不認你，丟了就是丟了，其他都是借口，

我說：沒事，要是不認的話，就把我們處理訂單那一套給他講講嘛，反正處理的思路都是一樣的，

不過啊，廣告訂單邏輯處理又相對沒那么復雜，廣告訂單最后是以入資料庫作為標準的，又可以接受一定的延遲，只要能保證處理完就行了，

要想client端消費資料不能丟，肯定是不能使用autoCommit的，所以必須是手動提交的，

候選者：我們這邊是這樣實作的：

候選者：一、從Kafka拉取訊息（一次批量拉取500條，這里主要看配置）時

候選者：二、為每條拉取的訊息分配一個msgId（遞增）

候選者：三、將msgId存入記憶體佇列（sortSet）中

候選者：四、使用Map存盤msgId與msg(有offset相關的資訊）的映射關系，通過msgId用來獲取相關元資訊

候選者：五、當業務處理完訊息后，ack時，獲取當前處理的訊息msgId，然后從sortSet洗掉該msgId（此時代表已經處理過了）

候選者：六、接著與sortSet佇列（本地記憶體佇列）的首部第一個Id比較（其實就是最小的msgId），如果當前msgId<=sort Set第一個ID，則提交當前offset

候選者：七、系統即便掛了，在下次重啟時就會從sortSet隊首的訊息開始拉取，實作至少處理一次語意

候選者：八、會有少量的訊息重復，但只要下游做好冪等就OK了，

面試官：嗯，你也提到了冪等，你們這業務怎么實作冪等性的呢？

候選者：嗯，還是以處理訂單訊息為例好了，

候選者：冪等Key我們由訂單編號+訂單狀態所組成（一筆訂單的狀態只會處理一次）

候選者：在處理之前，我們首先會去查Redis是否存在該Key，如果存在，則說明我們已經處理過了，直接丟掉

候選者：如果Redis沒處理過，則繼續往下處理，最終的邏輯是將處理過的資料插入到業務DB上，再到最后把冪等Key插入到Redis上

候選者：顯然，單純通過Redis是無法保證冪等的（：

候選者：所以，Redis其實只是一個「前置」處理，最終的冪等性是依賴資料庫的唯一Key來保證的（唯一Key實際上也是訂單編號+狀態）

候選者：總的來說，就是通過Redis做前置處理，DB唯一索引做最終保證來實作冪等性的

保證austin資料不丟需要做什么？

保證資料不丟簡單來說，就是我們要在消費端手動ack offset，不能再用自動提交策略了，這樣當我們系統重啟時，kafka會自動從未ack的offset中拉取，

如果要實作訊息推送平臺不丟訊息的話，有幾個問題是需要考慮的：

1、訊息少發比多發要好，那么要實作訊息不丟，就必須要在系統內實作冪等，因為現在的訊息不丟，一般都是基于【至少一次]消費語意去做的，

2、那實作冪等的邏輯是在呼叫渠道下發介面前，還是渠道下發介面后？

如果做在下發介面前，那是不是會有可能第一次下發記錄寫入了，但實際呼叫下發介面卻失敗了，后面的重試都被冪等處理掉了，

如果做在下發介面后，那是不是會有可能呼叫呼叫下發介面成功了，但寫入冪等處理的訊息失敗了，后面的重試就會導致訊息多發

3、訊息是有時效性的，那如果重試的處理時間過長，那是不是要考慮把這條訊息給丟棄掉，不再重試了，

4、重試的訊息不應該影響到正常訊息的下發，他得作為一種補償的機制，而非主流程，

稍微細想下技術實作，應該不太好搞，還有很多細節的地方得關注到，比如業務上的：應該是不需要所有的渠道的所有型別訊息都得實作訊息不丟吧？現在的設計是追求高性能的，能在短時間內下發批量的訊息，而如果做到所有訊息不丟，肯定會影響到下發的速率，

什么時候動手？

1、對于這個功能吧，有用肯定是有用，但這功能又沒那么急，

2、我估摸對現有代碼改動還是蠻大的，現在我還沒想好該怎么實作比較好，也一直沒下手，

3、最近作業的事挺多的，沒那么有空，

結論：先看看想要這個功能的人多不多，不多就鴿一會，

都看到這了，如果按上面的理由，我不實作這個功能，你認不認可？

訊息推送平臺??推送下發【郵件】【短信】【微信服務號】【微信小程式】【企業微信】【釘釘】等訊息型別，

• https://gitee.com/zhongfucheng/austin/
• https://github.com/ZhongFuCheng3y/austin

標籤：Java

上一篇：【深入淺出 Yarn 架構與實作】6-4 Container 生命周期原始碼分析

下一篇：返回列表