快捷導(dǎo)航

RocketMQ消息丟失場(chǎng)景以及解決方法

更新時(shí)間：2020年09月16日 09:06:59 作者：霽云HYY

這篇文章主要給大家介紹了關(guān)于RocketMQ消息丟失場(chǎng)景以及解決方法，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

既然使用在項(xiàng)目中使用了MQ，那么就不可避免的需要考慮消息丟失問題。在一些涉及到了金錢交易的場(chǎng)景下，消息丟失還是很致命的。那么在RocketMQ中存在哪幾種消息丟失的場(chǎng)景呢？

先來一張最簡(jiǎn)單的消費(fèi)流程圖：

消費(fèi)流程

上圖中大致包含了這么幾種場(chǎng)景：

生產(chǎn)者產(chǎn)生消息發(fā)送給RocketMQRocketMQ接收到了消息之后，必然需要存到磁盤中，否則斷電或宕機(jī)之后會(huì)造成數(shù)據(jù)的丟失消費(fèi)者從RocketMQ中獲取消息消費(fèi)，消費(fèi)成功之后，整個(gè)流程結(jié)束

這三種場(chǎng)景都可能會(huì)產(chǎn)生消息的丟失，如下圖所示：

消息丟失

場(chǎng)景1中生產(chǎn)者將消息發(fā)送給Rocket MQ的時(shí)候，如果出現(xiàn)了網(wǎng)絡(luò)抖動(dòng)或者通信異常等問題，消息就有可能會(huì)丟失場(chǎng)景2中消息需要持久化到磁盤中，這時(shí)會(huì)有兩種情況導(dǎo)致消息丟失

①RocketMQ為了減少磁盤的IO，會(huì)先將消息寫入到os cache中，而不是直接寫入到磁盤中，消費(fèi)者從os cache中獲取消息類似于直接從內(nèi)存中獲取消息，速度更快，過一段時(shí)間會(huì)由os線程異步的將消息刷入磁盤中，此時(shí)才算真正完成了消息的持久化。在這個(gè)過程中，如果消息還沒有完成異步刷盤，RocketMQ中的Broker宕機(jī)的話，就會(huì)導(dǎo)致消息丟失

②如果消息已經(jīng)被刷入了磁盤中，但是數(shù)據(jù)沒有做任何備份，一旦磁盤損壞，那么消息也會(huì)丟失消費(fèi)者成功從RocketMQ中獲取到了消息，還沒有將消息完全消費(fèi)完的時(shí)候，就通知RocketMQ我已經(jīng)將消息消費(fèi)了，然后消費(fèi)者宕機(jī)，但是RocketMQ認(rèn)為消費(fèi)者已經(jīng)成功消費(fèi)了數(shù)據(jù)，所以數(shù)據(jù)依舊丟失了

那么如何保證消息的零丟失呢？

保證消息零丟失

1、場(chǎng)景1中保證消息不丟失的方案是使用RocketMQ自帶的事務(wù)機(jī)制來發(fā)送消息，大致流程為

①首先生產(chǎn)者發(fā)送half消息到RocketMQ中，此時(shí)消費(fèi)者是無法消費(fèi)half消息的，若half消息就發(fā)送失敗了，則執(zhí)行相應(yīng)的回滾邏輯

②half消息發(fā)送成功之后，且RocketMQ返回成功響應(yīng)，則執(zhí)行生產(chǎn)者的核心鏈路

③如果生產(chǎn)者自己的核心鏈路執(zhí)行失敗，則回滾，并通知RocketMQ刪除half消息

④如果生產(chǎn)者的核心鏈路執(zhí)行成功，則通知RocketMQ commit half消息，讓消費(fèi)者可以消費(fèi)這條數(shù)據(jù)

其中還有一些RocketMQ長(zhǎng)時(shí)間沒有收到生產(chǎn)者是要commit/rollback操作的響應(yīng)，回調(diào)生產(chǎn)者接口的細(xì)節(jié)，感興趣的可以參考文末的 RocketMQ分布式事務(wù)原理

在使用了RocketMQ事務(wù)將生產(chǎn)者的消息成功發(fā)送給RocketMQ，就可以保證在這個(gè)階段消息不會(huì)丟失在場(chǎng)景2中要保證消息不丟失，首先需要將os cache的異步刷盤策略改為同步刷盤，這一步需要修改Broker的配置文件，將flushDiskType改為SYNC_FLUSH同步刷盤策略，默認(rèn)的是ASYNC_FLUSH異步刷盤。一旦同步刷盤返回成功，那么就一定保證消息已經(jīng)持久化到磁盤中了；為了保證磁盤損壞不會(huì)丟失數(shù)據(jù)，我們需要對(duì)RocketMQ采用主從機(jī)構(gòu)，集群部署，Leader中的數(shù)據(jù)在多個(gè)Follower中都存有備份，防止單點(diǎn)故障。在場(chǎng)景3中，消息到達(dá)了消費(fèi)者，RocketMQ在代碼中就能保證消息不會(huì)丟失

//注冊(cè)消息監(jiān)聽器處理消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
 @Override
 public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context){ 		 						      
  //對(duì)消息進(jìn)行處理
  return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
 }
});

上面這段代碼中，RocketMQ在消費(fèi)者中注冊(cè)了一個(gè)監(jiān)聽器，當(dāng)消費(fèi)者獲取到了消息，就會(huì)去回調(diào)這個(gè)監(jiān)聽器函數(shù)，去處理里面的消息

當(dāng)你的消息處理完畢之后，才會(huì)返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS
只有返回了CONSUME_SUCCESS，消費(fèi)者才會(huì)告訴RocketMQ我已經(jīng)消費(fèi)完了，此時(shí)如果消費(fèi)者宕機(jī)，消息已經(jīng)處理完了，也就不會(huì)丟失消息了

如果消費(fèi)者還沒有返回CONSUME_SUCCESS時(shí)就宕機(jī)了，那么RocketMQ就會(huì)認(rèn)為你這個(gè)消費(fèi)者節(jié)點(diǎn)掛掉了，會(huì)自動(dòng)故障轉(zhuǎn)移，將消息交給消費(fèi)者組的其他消費(fèi)者去消費(fèi)這個(gè)消息，保證消息不會(huì)丟失

為了保證消息不會(huì)丟失，在consumeMessage方法中就直接寫消息消費(fèi)的業(yè)務(wù)邏輯就可以了，如果非要搞一些騷操作，比如下面的代碼

//注冊(cè)消息監(jiān)聽器處理消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
 @Override
 public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context){ 
 	//開啟子線程異步處理消息
 	new Thread() {
			public void run() {
				//對(duì)消息進(jìn)行處理
			}
		}.start();		 						      
  return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
 }
});

如果新開子線程異步處理消息的話，就有可能出現(xiàn)消息還沒有被消費(fèi)完，消費(fèi)者告訴RocketMQ消息已經(jīng)被消費(fèi)了，結(jié)果宕機(jī)丟失消息的情況。

使用上面一整套的方案就可以在使用RocketMQ時(shí)保證消息零丟失，但是性能和吞吐量也將大幅下降

使用事務(wù)機(jī)制傳輸消息，會(huì)比普通的消息傳輸多出很多步驟，耗費(fèi)性能同步刷盤相比異步刷盤，一個(gè)是存儲(chǔ)在磁盤中，一個(gè)存儲(chǔ)在內(nèi)存中，速度完全不是一個(gè)數(shù)量級(jí)主從架構(gòu)的話，需要Leader將數(shù)據(jù)同步給Follower消費(fèi)時(shí)無法異步消費(fèi)，只能等待消費(fèi)完成再通知RocketMQ消費(fèi)完成

消息零丟失是一把雙刃劍，要想用好，還是要視具體的業(yè)務(wù)場(chǎng)景而定，選擇合適的方案才是最好的

RocketMQ分布式事務(wù)原理

分布式事務(wù)常見的方案有TCC（Try-Confirm-Cancel），XA兩階段提交方案，可靠消息最終一致性方案，最大努力通知方案等等。其中可靠消息最終一致性方案主要就可以依靠RocketMQ來做，因?yàn)镽ocketMQ支持消息事務(wù)。先上一張圖：

RocketMQ 事務(wù)消息的實(shí)現(xiàn)步驟如下：

Producer發(fā)送half message給RocketMQ
RocketMQ返回half message success（half message發(fā)送成功之后RocketMQ的消費(fèi)者并不能消費(fèi)這條消息，因?yàn)橄⒋鎯?chǔ)在Topic為 RMQ_SYS_TRANS_HALF_TOPIC 的消息消費(fèi)隊(duì)列中，而不是原先的Topic）
執(zhí)行核心交易鏈路
返回執(zhí)行交易鏈路的結(jié)果，如果失敗則回滾
如果執(zhí)行成功，則Producer返回一個(gè)COMMIT狀態(tài)給RocketMQ
如果RocketMQ遲遲收不到Producer的返回結(jié)果，即這條消息的狀態(tài)為UNKNOWN，則會(huì)回調(diào)服務(wù)接口，查詢這條消息到底是commit還是rollback
RocketMQ確認(rèn)消息為commit，則Consumer可以消費(fèi)到這條消息
Consumer操作數(shù)據(jù)庫(kù)，執(zhí)行自己的事務(wù)
Consumer成功消費(fèi)之后返回一個(gè)ACK消息給RocketMQ，如果成功消費(fèi)則顯示消費(fèi)成功，否則RocketMQ會(huì)重發(fā)消息給Consumer繼續(xù)消費(fèi)

RocketMQ 事務(wù)消息的實(shí)現(xiàn)原理基于兩階段提交和定時(shí)事務(wù)狀態(tài)回查來決定消息最終是提交還是回滾，RocketMQ 先執(zhí)行第一部分的事務(wù)，如果失敗則回滾，如果成功則定時(shí)任務(wù)會(huì)去回查到事務(wù)執(zhí)行成功，這個(gè)時(shí)候通知消費(fèi)者執(zhí)行第二階段的事務(wù)，如果失敗則不斷重發(fā)消息給消費(fèi)者消費(fèi)，如果成功則整個(gè)流程走完，保證了事務(wù)的原子性。

總結(jié)

到此這篇關(guān)于RocketMQ消息丟失場(chǎng)景以及解決方法的文章就介紹到這了,更多相關(guān)RocketMQ消息丟失場(chǎng)景內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: