rocketmq如何保證消息的不丟失


 

不知道大家對於這個問題遇到過沒有,或者大家聽到這個問題的第一反應是什么,應該如何做,如何避免消息丟失,一起來看看

首先我們知道rocketmq的一個消息從生產到最終的消費過程需要經歷總共三個階段,或者說會經過三個地方,分別是producer的發送端、broker的持久化機制、以及consumer的消費端

 

 

從生產者producer的角度:消息生產之后傳遞到broker,如果消息未能正確的存儲到broker中,算作消息丟失

從broker的角度:消息默認保存到broker的內存中,異步保存到磁盤上,如果發生宕機、磁盤崩潰會造成消息丟失

從消費者consumer的角度:消息完成了持久化之后,consumer拉取之后未能成功消費且未反饋給broker,這樣算作消息丟失,可能消費過程異常或者網絡抖動造成消息丟失

生產者角度

從生產者的角度,生產了消息就是要通過網絡發送到broker,其實只需要保證一點,就是確認這個消息已經成功發送到broker上了

生產者只需要接收發送消息返回的確認響應即可,就可以代表消息發送成功

代碼示例:

DefaultMQProducer mqProducer=new DefaultMQProducer("test");
// 設置 nameSpace地址
mqProducer.setNamesrvAddr("namesrvAddr");
mqProducer.start();
Message msg = new Message("topic" /* Topic */,
        "Captain".getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
);
// 發送消息到Broker
try {
    SendResult sendResult = mqProducer.send(msg);
} catch (Exception e) {
    e.printStackTrace();
}

當然,發送消息也分為同步和異步兩種,消息發送成功之后會返回下面這四種不同的響應狀態

SEND_OK

消息發送成功,但是也並不意味這完全代表不會丟失消息,這還要取決於broker的刷盤方式

這個下面在broker方面會說,需要啟動SYNC_MASTER或SYNC_FLUSH。(也就是同步)

FLUSH_DISK_TIMEOUT

如果Broker設置MessageStoreConfig的FlushDiskType = SYNC_FLUSH(默認為ASYNC_FLUSH),並且Broker沒有在MessageStoreConfig的syncFlushTimeout(默認為5秒)內完成刷新磁盤,得到這個狀態

也就是此時刷盤超時,未能在規定時間內落到硬盤上,檢查設置是否合理、硬盤大小等情況

FLUSH_SLAVE_TIMEOUT

如果Broker的角色是SYNC_MASTER(默認為ASYNC_MASTER),並且從屬Broker未在MessageStoreConfig的syncFlushTimeout(默認為5秒)內完成與主服務器的同步,返回這個結果

主從同步時間默認也是5秒,需要完成主從同步,這個下面在說broker的時候也會說到,你想啊,要是master掛了或者磁盤崩潰了,這是不是也不能百分百保證消息不丟失

SLAVE_NOT_AVAILABLE

如果Broker的角色是SYNC_MASTER(默認為ASYNC_MASTER),但沒有配置slave Broker,返回這個狀態

這個在保證消息絕對不丟失的情況下是不允許存在的,這個狀態也是屬於需要處理的,沒有可靠的slave,也就意味着沒有可靠的備份數據,所以這種情況也需要考慮

另外呢,上面還說了一種異步的消息發送方式,這種一般是用於鏈路較長,對於時間比較敏感的業務,這種情況下需要特別注意的就是我們需要設置消息發送完成的回調,這樣才能更好的保證消息不丟失 。

采取事務消息的投遞方式,並不能保證消息100%投遞成功到了Broker,但是如果消息發送Ack失敗的話,此消息會存儲在CommitLog當中,但是對ConsumerQueue是不可見的

可以在日志中查看到這條異常的消息,嚴格意義上來講,也並沒有完全丟失

從broker的角度

順序消費這個場景其實不是特別的常見,但是也是必不可少的,因為在某些業務場景下順序是很關鍵的,保證消息的消費順序也是很關鍵

消息到了broker之后,默認是優先保存到broker的內存中,然后立刻返回響應給生產者producer,然后broker自己定期將消息批量的異步的保存到硬盤上

有的小伙伴一小子就發現了問題不是那么簡單,消息來了之后還沒保存到硬盤,就直接返回了,broker直接宕機崩潰了,那這消息豈不無跡可尋了

這樣的優點是提高交互的效率,同時減少IO的次數,問題就是會造成消息丟失

如果我們想要保證消息不丟失,那就需要保證消息成功保存到broker之后才可以返回,只需要將消息的保存機制修改為同步刷盤的方式,也就是只有消息保存到broker的磁盤成功之后,才會返回響應 

## 默認情況為 ASYNC_FLUSH 
flushDiskType = SYNC_FLUSH

如果broker未能在規定的同步時間(默認5秒)完成刷盤,將返回FLUSH_DISK_TIMEOUT給生產者

上面也介紹了這個了FLUSH_DISK_TIMEOUT了

一般在系統中為了保證可用性,broker通常采用的都是一主master多從slave的部署方式,屬於集群部署

為了保證消息不丟失,消息需要復制到slave節點,其實默認的情況下,消息寫入到broker之后就會返回成功 

但是!如果master突然宕機或者磁盤崩潰了,那么這個消息就徹底丟失了,沒有備份,所以呢,這里還需要把master和slave的異步復制改成同步復制

## master 節點配置
flushDiskType = SYNC_FLUSH
brokerRole=SYNC_MASTER


## slave 節點配置
brokerRole=slave
flushDiskType = SYNC_FLUSH

也就是只有slave也刷盤到磁盤成功之后,才會給producer返回成功 

當然你要這里說,master和slave也可能同時宕機,同時磁盤崩潰,那最終還是無法滿足百分百保證消息的不丟失

這種問題啊,其實就像是TCP的三次交互一樣,三次交互之后一定保證客戶端和服務端通信成功了嗎,答案是不一定

我們只能在有限的資源下盡量的去滿足系統的穩定性

為了保證可用性和性能,通常的做法是:同步雙寫+異步刷盤的方式部署集群

從消費者consumer的角度

消費者從broker拉取消息,然后進行相應的業務的消費,消費成功會返回一個消費成功的狀態給broker,broker如果沒收到確認信息,消費者下次拉取重新拉取該消息

consumer自身可以維護一個持久化的offset,對應MessageQueue里面的min offset,標記已經成功消費或者已經成功發回到broker的消息下標

如果consumer消費失敗,會把這個消息發回給broker,發回成功后,更新自己的offset

如果發回給broker時,broker掛掉了,那么consumer也會定時重試這個操作

即使consumer和broker一起掛掉了,消息也不會丟失,因為consumer里面的offset會定時持久化,重啟之后,繼續拉取offset之前的消息到本地,重新消費

 

轉載:https://blog.csdn.net/qq_24692021/article/details/121007540 

 

  

 


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM