rocketmq是阿里開源的一個性能很強大的消息隊列，很多公司都在用，而且經歷了多次雙十一的洗禮，支持多種特性

對于這個技術點不知道大家掌握的如何了，消息隊列現在應該是公司必備的技能之一了，無論是RabbitMQ還是rocketmq，或者支持大數量的kafka

今天我們要說的一個問題，是rocketmq如何保證消息的不丟失？？

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

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

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

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

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

生產者角度：消費生產之后傳遞到broker，如果消息未能正確的保存到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之后，默認是優先保存到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：消息完成了持久化之后，consumer拉取之后未能成功消費且未反饋給broker，這樣算作消息丟失，可能消費過程異常或者網絡抖動造成消息丟失

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

// 注冊回調實現類來處理從broker拉取回來的消息 consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {// 標記該消息已經被成功消費return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;} });

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

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

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

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

有道無術，術可成；有術無道，止于術

歡迎大家關注Java之道公眾號

好文章，我在看??

總結

以上是生活随笔為你收集整理的回答面试官：如何保证消息不丢失的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。