[[430419]]

rocketmq是阿里開(kāi)源的一個(gè)性能很強(qiáng)大的消息隊(duì)列，很多公司都在用，而且經(jīng)歷了多次雙十一的洗禮，支持多種特性

對(duì)于這個(gè)技術(shù)點(diǎn)不知道大家掌握的如何了，消息隊(duì)列現(xiàn)在應(yīng)該是公司必備的技能之一了，無(wú)論是RabbitMQ還是rocketmq，或者支持大數(shù)量的kafka

今天我們要說(shuō)的一個(gè)問(wèn)題，是rocketmq如何保證消息的不丟失??

不知道大家對(duì)于這個(gè)問(wèn)題遇到過(guò)沒(méi)有，或者大家聽(tīng)到這個(gè)問(wèn)題的第一反應(yīng)是什么，應(yīng)該如何做，如何避免消息丟失，一起來(lái)看看

首先我們知道rocketmq的一個(gè)消息從生產(chǎn)到最終的消費(fèi)過(guò)程需要經(jīng)歷總共三個(gè)階段，或者說(shuō)會(huì)經(jīng)過(guò)三個(gè)地方，分別是producer的發(fā)送端、broker的持久化機(jī)制、以及consumer的消費(fèi)端

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

從broker的角度：消息默認(rèn)保存到broker的內(nèi)存中，異步保存到磁盤上，如果發(fā)生宕機(jī)、磁盤崩潰會(huì)造成消息丟失

從消費(fèi)者consumer的角度：消息完成了持久化之后，consumer拉取之后未能成功消費(fèi)且未反饋給broker，這樣算作消息丟失，可能消費(fèi)過(guò)程異?；蛘呔W(wǎng)絡(luò)抖動(dòng)造成消息丟失

生產(chǎn)者角度：消費(fèi)生產(chǎn)之后傳遞到broker，如果消息未能正確的保存到broker中，算作消息丟失

從生產(chǎn)者的角度，生產(chǎn)了消息就是要通過(guò)網(wǎng)絡(luò)發(fā)送到broker，其實(shí)只需要保證一點(diǎn)，就是確認(rèn)這個(gè)消息已經(jīng)成功發(fā)送到broker上了

生產(chǎn)者只需要接收發(fā)送消息返回的確認(rèn)響應(yīng)即可，就可以代表消息發(fā)送成功

代碼示例：

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

當(dāng)然，發(fā)送消息也分為同步和異步兩種，消息發(fā)送成功之后會(huì)返回下面這四種不同的響應(yīng)狀態(tài)

SEND_OK

消息發(fā)送成功，但是也并不意味這完全代表不會(huì)丟失消息，這還要取決于broker的刷盤方式

這個(gè)下面在broker方面會(huì)說(shuō)，需要啟動(dòng)SYNC_MASTER或SYNC_FLUSH。(也就是同步)

FLUSH_DISK_TIMEOUT

如果Broker設(shè)置MessageStoreConfig的FlushDiskType = SYNC_FLUSH(默認(rèn)為ASYNC_FLUSH)，并且Broker沒(méi)有在MessageStoreConfig的syncFlushTimeout(默認(rèn)為5秒)內(nèi)完成刷新磁盤，得到這個(gè)狀態(tài)

也就是此時(shí)刷盤超時(shí)，未能在規(guī)定時(shí)間內(nèi)落到硬盤上，檢查設(shè)置是否合理、硬盤大小等情況

FLUSH_SLAVE_TIMEOUT

如果Broker的角色是SYNC_MASTER(默認(rèn)為ASYNC_MASTER)，并且從屬Broker未在MessageStoreConfig的syncFlushTimeout(默認(rèn)為5秒)內(nèi)完成與主服務(wù)器的同步，返回這個(gè)結(jié)果

主從同步時(shí)間默認(rèn)也是5秒，需要完成主從同步，這個(gè)下面在說(shuō)broker的時(shí)候也會(huì)說(shuō)到，你想啊，要是master掛了或者磁盤崩潰了，這是不是也不能百分百保證消息不丟失

SLAVE_NOT_AVAILABLE

如果Broker的角色是SYNC_MASTER(默認(rèn)為ASYNC_MASTER)，但沒(méi)有配置slave Broker，返回這個(gè)狀態(tài)

這個(gè)在保證消息絕對(duì)不丟失的情況下是不允許存在的，這個(gè)狀態(tài)也是屬于需要處理的，沒(méi)有可靠的slave，也就意味著沒(méi)有可靠的備份數(shù)據(jù)，所以這種情況也需要考慮

另外呢，上面還說(shuō)了一種異步的消息發(fā)送方式，這種一般是用于鏈路較長(zhǎng)，對(duì)于時(shí)間比較敏感的業(yè)務(wù)

這種情況下需要特別注意的就是我們需要設(shè)置消息發(fā)送完成的回調(diào)，這樣才能更好的保證消息不丟失

采取事務(wù)消息的投遞方式，并不能保證消息100%投遞成功到了Broker，但是如果消息發(fā)送Ack失敗的話，此消息會(huì)存儲(chǔ)在CommitLog當(dāng)中，但是對(duì)ConsumerQueue是不可見(jiàn)的

可以在日志中查看到這條異常的消息，嚴(yán)格意義上來(lái)講，也并沒(méi)有完全丟失

• broker：消息默認(rèn)保存到broker的內(nèi)存中，異步保存到磁盤上，如果發(fā)生宕機(jī)、磁盤崩潰會(huì)造成消息丟失

順序消費(fèi)這個(gè)場(chǎng)景其實(shí)不是特別的常見(jiàn)，但是也是必不可少的，因?yàn)樵谀承I(yè)務(wù)場(chǎng)景下順序是很關(guān)鍵的，保證消息的消費(fèi)順序也是很關(guān)鍵

消息到了broker之后，默認(rèn)是優(yōu)先保存到broker的內(nèi)存中，然后立刻返回響應(yīng)給生產(chǎn)者producer，然后broker自己定期將消息批量的異步的保存到硬盤上

有的小伙伴一小子就發(fā)現(xiàn)了問(wèn)題不是那么簡(jiǎn)單，消息來(lái)了之后還沒(méi)保存到硬盤，就直接返回了，broker直接宕機(jī)崩潰了，那這消息豈不無(wú)跡可尋了

這樣的優(yōu)點(diǎn)是提高交互的效率，同時(shí)減少IO的次數(shù)，問(wèn)題就是會(huì)造成消息丟失

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

## 默認(rèn)情況為 ASYNC_FLUSH  
flushDiskType = SYNC_FLUSH 

如果broker未能在規(guī)定的同步時(shí)間(默認(rèn)5秒)完成刷盤，將返回FLUSH_DISK_TIMEOUT給生產(chǎn)者

上面也介紹了這個(gè)了FLUSH_DISK_TIMEOUT了

一般在系統(tǒng)中為了保證可用性，broker通常采用的都是一主master多從slave的部署方式，屬于集群部署

為了保證消息不丟失，消息需要復(fù)制到slave節(jié)點(diǎn)，其實(shí)默認(rèn)的情況下，消息寫(xiě)入到broker之后就會(huì)返回成功

但是!如果master突然宕機(jī)或者磁盤崩潰了，那么這個(gè)消息就徹底丟失了，沒(méi)有備份，所以呢，這里還需要把master和slave的異步復(fù)制改成同步復(fù)制

## master 節(jié)點(diǎn)配置 
flushDiskType = SYNC_FLUSH 
brokerRole=SYNC_MASTER 
 
## slave 節(jié)點(diǎn)配置 
brokerRole=slave 
flushDiskType = SYNC_FLUSH 

也就是只有slave也刷盤到磁盤成功之后，才會(huì)給producer返回成功

當(dāng)然你要這里說(shuō)，master和slave也可能同時(shí)宕機(jī)，同時(shí)磁盤崩潰，那最終還是無(wú)法滿足百分百保證消息的不丟失

這種問(wèn)題啊，其實(shí)就像是TCP的三次交互一樣，三次交互之后一定保證客戶端和服務(wù)端通信成功了嗎，答案是不一定

我們只能在有限的資源下盡量的去滿足系統(tǒng)的穩(wěn)定性

• consumer：消息完成了持久化之后，consumer拉取之后未能成功消費(fèi)且未反饋給broker，這樣算作消息丟失，可能消費(fèi)過(guò)程異?；蛘呔W(wǎng)絡(luò)抖動(dòng)造成消息丟失

消費(fèi)者從broker拉取消息，然后進(jìn)行相應(yīng)的業(yè)務(wù)的消費(fèi)，消費(fèi)成功會(huì)返回一個(gè)消費(fèi)成功的狀態(tài)給broker，broker如果沒(méi)收到確認(rèn)信息，消費(fèi)者下次拉取重新拉取該消息

// 注冊(cè)回調(diào)實(shí)現(xiàn)類來(lái)處理從broker拉取回來(lái)的消息 
consumer.registerMessageListener(new MessageListenerConcurrently() { 
    @Override 
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { 
        // 標(biāo)記該消息已經(jīng)被成功消費(fèi) 
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; 
    } 
}); 

consumer自身可以維護(hù)一個(gè)持久化的offset，對(duì)應(yīng)MessageQueue里面的min offset，標(biāo)記已經(jīng)成功消費(fèi)或者已經(jīng)成功發(fā)回到broker的消息下標(biāo)

如果consumer消費(fèi)失敗，會(huì)把這個(gè)消息發(fā)回給broker，發(fā)回成功后，更新自己的offset

如果發(fā)回給broker時(shí)，broker掛掉了，那么consumer也會(huì)定時(shí)重試這個(gè)操作

即使consumer和broker一起掛掉了，消息也不會(huì)丟失，因?yàn)閏onsumer里面的offset會(huì)定時(shí)持久化，重啟之后，繼續(xù)拉取offset之前的消息到本地，重新消費(fèi)

本文轉(zhuǎn)載自微信公眾號(hào)「Java賊船」