<kbd id="afajh"><form id="afajh"></form></kbd>
<strong id="afajh"><dl id="afajh"></dl></strong>
    <del id="afajh"><form id="afajh"></form></del>
        1. <th id="afajh"><progress id="afajh"></progress></th>
          <b id="afajh"><abbr id="afajh"></abbr></b>
          <th id="afajh"><progress id="afajh"></progress></th>

          回答面試官:如何保證消息不丟失

          共 4049字,需瀏覽 9分鐘

           ·

          2021-11-03 16:36

          48bae621436b3af2fbdb806cee92c217.webp



          先撈一撈之前的文章



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


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


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

          5e8b3c56e3ccf46456741fb69c79ef34.webp


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


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


          4a6ecbb2a617a7a8cbaf802d5ffa7bf1.webp


          從生產(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)過程異常或者網(wǎng)絡(luò)抖動(dòng)造成消息丟失


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


          從生產(chǎn)者的角度,生產(chǎn)了消息就是要通過網(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ā)送消息到Brokertry {    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ì)說,需要啟動(dòng)SYNC_MASTERSYNC_FLUSH。(也就是同步)


          FLUSH_DISK_TIMEOUT


          如果Broker設(shè)置MessageStoreConfigFlushDiskType = SYNC_FLUSH(默認(rèn)為ASYNC_FLUSH),并且Broker沒有在MessageStoreConfigsyncFlushTimeout(默認(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未在MessageStoreConfigsyncFlushTimeout(默認(rèn)為5秒)內(nèi)完成與主服務(wù)器的同步,返回這個(gè)結(jié)果


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

          ?

          SLAVE_NOT_AVAILABLE


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


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


          7e67d55d133e5d940ea95c597d00a892.webp


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


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


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


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



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


          順序消費(fèi)這個(gè)場(chǎng)景其實(shí)不是特別的常見,但是也是必不可少的,因?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)了問題不是那么簡(jiǎn)單,消息來了之后還沒保存到硬盤,就直接返回了,broker直接宕機(jī)崩潰了,那這消息豈不無跡可尋了


          這樣的優(yōu)點(diǎn)是提高交互的效率,同時(shí)減少IO的次數(shù),問題就是會(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)的情況下,消息寫入到broker之后就會(huì)返回成功


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


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


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


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


          04218b56d967475d90c0dcaf8e29635f.webp


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


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


          consumer:消息完成了持久化之后,consumer拉取之后未能成功消費(fèi)且未反饋給broker,這樣算作消息丟失,可能消費(fèi)過程異常或者網(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如果沒收到確認(rèn)信息,消費(fèi)者下次拉取重新拉取該消息


          // 注冊(cè)回調(diào)實(shí)現(xiàn)類來處理從broker拉取回來的消息consumer.registerMessageListener(new MessageListenerConcurrently() {    @Override????public?ConsumeConcurrentlyStatus?consumeMessage(List?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)


          佛系求關(guān)注




          Captain希望有一天能夠靠寫作養(yǎng)活自己,現(xiàn)在還在磨練,這個(gè)時(shí)間可能會(huì)持續(xù)很久,但是,請(qǐng)看我漂亮的堅(jiān)持


          感謝大家能夠做我最初的讀者和傳播者,請(qǐng)大家相信,只要你給我一份愛,我終究會(huì)還你們一頁情的。


          Captain會(huì)持續(xù)更新技術(shù)文章,和生活中的暴躁文章,歡迎大家關(guān)注【Java賊船】,成為船長(zhǎng)的學(xué)習(xí)小伙伴,和船長(zhǎng)一起乘千里風(fēng)、破萬里浪


          哦對(duì)了,后續(xù)所有的遠(yuǎn)程文章都會(huì)更新到這里


          https://github.com/DayuMM2021/Java


          a0632f936c20872ea30496cb508be713.webp



          瀏覽 57
          點(diǎn)贊
          評(píng)論
          收藏
          分享

          手機(jī)掃一掃分享

          分享
          舉報(bào)
          評(píng)論
          圖片
          表情
          推薦
          點(diǎn)贊
          評(píng)論
          收藏
          分享

          手機(jī)掃一掃分享

          分享
          舉報(bào)
          <kbd id="afajh"><form id="afajh"></form></kbd>
          <strong id="afajh"><dl id="afajh"></dl></strong>
            <del id="afajh"><form id="afajh"></form></del>
                1. <th id="afajh"><progress id="afajh"></progress></th>
                  <b id="afajh"><abbr id="afajh"></abbr></b>
                  <th id="afajh"><progress id="afajh"></progress></th>
                  高清无码成人电影 | 大香蕉久久伊 | 午夜免费无码视频 | 精品蜜桃视频 | 久久网欧美大香蕉 |