RocketMQ訊息丟失場景及解決辦法
既然使用在專案中使用了MQ,那麼就不可避免的需要考慮訊息丟失問題。在一些涉及到了金錢交易的場景下,訊息丟失還是很致命的。那麼在RocketMQ中存在哪幾種訊息丟失的場景呢?
先來一張最簡單的消費流程圖:
上圖中大致包含了這麼幾種場景:
- 生產者產生訊息傳送給RocketMQ
- RocketMQ接收到了訊息之後,必然需要存到磁碟中,否則斷電或宕機之後會造成資料的丟失
- 消費者從RocketMQ中獲取訊息消費,消費成功之後,整個流程結束
這三種場景都可能會產生訊息的丟失,如下圖所示:
1、場景1中生產者將訊息傳送給Rocket MQ的時候,如果出現了網路抖動或者通訊異常等問題,訊息就有可能會丟失
2、場景2中訊息需要持久化到磁碟中,這時會有兩種情況導致訊息丟失
- RocketMQ為了減少磁碟的IO,會先將訊息寫入到os cache中,而不是直接寫入到磁碟中,消費者從os cache中獲取訊息類似於直接從記憶體中獲取訊息,速度更快,過一段時間會由os執行緒非同步的將訊息刷入磁碟中,此時才算真正完成了訊息的持久化。在這個過程中,如果訊息還沒有完成非同步刷盤,RocketMQ中的Broker宕機的話,就會導致訊息丟失
- 如果訊息已經被刷入了磁碟中,但是資料沒有做任何備份,一旦磁碟損壞,那麼訊息也會丟失
3、消費者成功從RocketMQ中獲取到了訊息,還沒有將訊息完全消費完的時候,就通知RocketMQ我已經將訊息消費了,然後消費者宕機,但是RocketMQ認為消費者已經成功消費了資料,所以資料依舊丟失了。
那麼如何保證訊息的零丟失呢?
1、場景1中保證訊息不丟失的方案是使用RocketMQ自帶的事務機制來發送訊息,大致流程為
- 首先生產者傳送half訊息到RocketMQ中,此時消費者是無法消費half訊息的,若half訊息就傳送失敗了,則執行相應的回滾邏輯
- half訊息傳送成功之後,且RocketMQ返回成功響應,則執行生產者的核心鏈路
- 如果生產者自己的核心鏈路執行失敗,則回滾,並通知RocketMQ刪除half訊息
- 如果生產者的核心鏈路執行成功,則通知RocketMQ commit half訊息,讓消費者可以消費這條資料
其中還有一些RocketMQ長時間沒有收到生產者是要commit/rollback操作的響應,回撥生產者介面的細節,感興趣的可以參考:
在使用了RocketMQ事務將生產者的訊息成功傳送給RocketMQ,就可以保證在這個階段訊息不會丟失
2、在場景2中要保證訊息不丟失,首先需要將os cache的非同步刷盤策略改為同步刷盤,這一步需要修改Broker的配置檔案,將flushDiskType改為SYNC_FLUSH同步刷盤策略,預設的是ASYNC_FLUSH非同步刷盤。
一旦同步刷盤返回成功,那麼就一定保證訊息已經持久化到磁碟中了;為了保證磁碟損壞不會丟失資料,我們需要對RocketMQ採用主從機構,叢集部署,Leader中的資料在多個Follower中都存有備份,防止單點故障。
搜尋Java知音公眾號,回覆“後端面試”,送你一份Java面試題寶典
3、在場景3中,訊息到達了消費者,RocketMQ在程式碼中就能保證訊息不會丟失
//註冊訊息監聽器處理訊息 consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context){ //對訊息進行處理 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } });
上面這段程式碼中,RocketMQ在消費者中註冊了一個監聽器,當消費者獲取到了訊息,就會去回撥這個監聽器函式,去處理裡面的訊息
當你的訊息處理完畢之後,才會返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS 只有返回了CONSUME_SUCCESS,消費者才會告訴RocketMQ我已經消費完了,此時如果消費者宕機,訊息已經處理完了,也就不會丟失訊息了
如果消費者還沒有返回CONSUME_SUCCESS時就宕機了,那麼RocketMQ就會認為你這個消費者節點掛掉了,會自動故障轉移,將訊息交給消費者組的其他消費者去消費這個訊息,保證訊息不會丟失
為了保證訊息不會丟失,在consumeMessage方法中就直接寫訊息消費的業務邏輯就可以了,如果非要搞一些騷操作,比如下面的程式碼
//註冊訊息監聽器處理訊息 consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context){ //開啟子執行緒非同步處理訊息 new Thread() { public void run() { //對訊息進行處理 } }.start(); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } });
如果新開子執行緒非同步處理訊息的話,就有可能出現訊息還沒有被消費完,消費者告訴RocketMQ訊息已經被消費了,結果宕機丟失訊息的情況。
使用上面一整套的方案就可以在使用RocketMQ時保證訊息零丟失,但是效能和吞吐量也將大幅下降
- 使用事務機制傳輸訊息,會比普通的訊息傳輸多出很多步驟,耗費效能
- 同步刷盤相比非同步刷盤,一個是儲存在磁碟中,一個儲存在記憶體中,速度完全不是一個數量級
- 主從機構的話,需要Leader將資料同步給Follower
- 消費時無法非同步消費,只能等待消費完成再通知RocketMQ消費完成
訊息零丟失是一把雙刃劍,要想用好,還是要視具體的業務場景而定,選擇合適的方案才是最好的。