既然使用在專案中使用了MQ，那麼就不可避免的需要考慮訊息丟失問題。在一些涉及到了金錢交易的場景下，訊息丟失還是很致命的。那麼在RocketMQ中存在哪幾種訊息丟失的場景呢？

先來一張最簡單的消費流程圖：

上圖中大致包含了這麼幾種場景：

生產者產生訊息傳送給RocketMQRocketMQ接收到了訊息之後，必然需要存到磁碟中，否則斷電或宕機之後會造成資料的丟失消費者從RocketMQ中獲取訊息消費，消費成功之後，整個流程結束

這三種場景都可能會產生訊息的丟失，如下圖所示：

場景1中生產者將訊息傳送給Rocket MQ的時候，如果出現了網路抖動或者通訊異常等問題，訊息就有可能會丟失場景2中訊息需要持久化到磁碟中，這時會有兩種情況導致訊息丟失

①RocketMQ為了減少磁碟的IO，會先將訊息寫入到os cache中，而不是直接寫入到磁碟中，消費者從os cache中獲取訊息類似於直接從記憶體中獲取訊息，速度更快，過一段時間會由os執行緒非同步的將訊息刷入磁碟中，此時才算真正完成了訊息的持久化。在這個過程中，如果訊息還沒有完成非同步刷盤，RocketMQ中的Broker宕機的話，就會導致訊息丟失

②如果訊息已經被刷入了磁碟中，但是資料沒有做任何備份，一旦磁碟損壞，那麼訊息也會丟失消費者成功從RocketMQ中獲取到了訊息，還沒有將訊息完全消費完的時候，就通知RocketMQ我已經將訊息消費了，然後消費者宕機，但是RocketMQ認為消費者已經成功消費了資料，所以資料依舊丟失了

那麼如何保證訊息的零丟失呢？

1、場景1中保證訊息不丟失的方案是使用RocketMQ自帶的事務機制來發送訊息，大致流程為

①首先生產者傳送half訊息到RocketMQ中，此時消費者是無法消費half訊息的，若half訊息就傳送失敗了，則執行相應的回滾邏輯

②half訊息傳送成功之後，且RocketMQ返回成功響應，則執行生產者的核心鏈路

③如果生產者自己的核心鏈路執行失敗，則回滾，並通知RocketMQ刪除half訊息

④如果生產者的核心鏈路執行成功，則通知RocketMQ commit half訊息，讓消費者可以消費這條資料

其中還有一些RocketMQ長時間沒有收到生產者是要commit/rollback操作的響應，回撥生產者介面的細節，感興趣的可以參考文末的 RocketMQ分散式事務原理

在使用了RocketMQ事務將生產者的訊息成功傳送給RocketMQ，就可以保證在這個階段訊息不會丟失在場景2中要保證訊息不丟失，首先需要將os cache的非同步刷盤策略改為同步刷盤，這一步需要修改Broker的配置檔案，將flushDiskType改為SYNC_FLUSH同步刷盤策略，預設的是ASYNC_FLUSH非同步刷盤。一旦同步刷盤返回成功，那麼就一定保證訊息已經持久化到磁碟中了；為了保證磁碟損壞不會丟失資料，我們需要對RocketMQ採用主從機構，叢集部署，Leader中的資料在多個Follower中都存有備份，防止單點故障。在場景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消費完成

訊息零丟失是一把雙刃劍，要想用好，還是要視具體的業務場景而定，選擇合適的方案才是最好的

RocketMQ分散式事務原理

分散式事務常見的方案有TCC（Try-Confirm-Cancel），XA兩階段提交方案，可靠訊息最終一致性方案，最大努力通知方案等等。其中可靠訊息最終一致性方案主要就可以依靠RocketMQ來做，因為RocketMQ支援訊息事務。先上一張圖：

RocketMQ 事務訊息的實現步驟如下：

1. Producer傳送half message給RocketMQ
2. RocketMQ返回half message success（half message傳送成功之後RocketMQ的消費者並不能消費這條訊息，因為訊息儲存在Topic為 RMQ_SYS_TRANS_HALF_TOPIC 的訊息消費佇列中，而不是原先的Topic）
3. 執行核心交易鏈路
4. 返回執行交易鏈路的結果，如果失敗則回滾
5. 如果執行成功，則Producer返回一個COMMIT狀態給RocketMQ
6. 如果RocketMQ遲遲收不到Producer的返回結果，即這條訊息的狀態為UNKNOWN，則會回撥服務介面，查詢這條訊息到底是commit還是rollback
7. RocketMQ確認訊息為commit，則Consumer可以消費到這條訊息
8. Consumer操作資料庫，執行自己的事務
9. Consumer成功消費之後返回一個ACK訊息給RocketMQ，如果成功消費則顯示消費成功，否則RocketMQ會重發訊息給Consumer繼續消費
   

RocketMQ 事務訊息的實現原理基於兩階段提交和定時事務狀態回查來決定訊息最終是提交還是回滾，RocketMQ 先執行第一部分的事務，如果失敗則回滾，如果成功則定時任務會去回查到事務執行成功，這個時候通知消費者執行第二階段的事務，如果失敗則不斷重發訊息給消費者消費，如果成功則整個流程走完，保證了事務的原子性。

總結

到此這篇關於RocketMQ訊息丟失場景以及解決方法的文章就介紹到這了,更多相關RocketMQ訊息丟失場景內容請搜尋我們以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援我們！