<form id="jrvnn"></form>

        <address id="jrvnn"><listing id="jrvnn"></listing></address>
        <form id="jrvnn"><nobr id="jrvnn"><meter id="jrvnn"></meter></nobr></form>

          <sub id="jrvnn"><listing id="jrvnn"></listing></sub>

                RocketMQ 消息發送system busy、broker busy原因分析與解決方案

                小編:管理員 34閱讀 2022.08.01

                現象

                最近收到很多RocketMQ使用者反饋在消息發送過程中偶爾會出現如下4個錯誤信息之一:

                • [REJECTREQUEST]system busy, start flow control for a while
                • too many requests and system thread pool busy, RejectedExecutionException
                • [PC_SYNCHRONIZED]broker busy, start flow control for a while
                • [PCBUSY_CLEAN_QUEUE]broker busy, start flow control for a while, period in queue: %sms, size of queue: %d
                原理解讀

                在進行消息中間件的選型時,如果待選中間件在功能上、性能上都能滿足業務的情況下,建議把中間件的實現語言這個因素也考慮進去,畢竟選擇一門用自己擅長的語言實現的中間件會更具掌控性。在出現異常的情況下,我們可以根據自己的經驗提取錯誤信息關鍵字system busy,在RocketMQ源碼中直接搜索,得到拋出上述錯誤信息的代碼如下:

                其代碼入口為:org.apache.rocketmq.remoting.netty.NettyRemotingAbstract#processRequestCommand。從圖中可以看出,拋出上述錯誤的關鍵原因是:pair.getObject1().rejectRequest()和拋出RejectedExecutionException異常。

                備注:本文偏實戰,源碼只是作為分析的重點證據,故本文只會點出關鍵源碼,并不會詳細跟蹤其整個實現流程,如果想詳細了解其實現,可以查閱筆者編著的《RocketMQ技術內幕》。

                2.1 RocketMQ 網絡處理機制概述

                RocketMQ的網絡設計非常值得我們學習與借鑒,首先在客戶端端將不同的請求定義不同的請求命令CODE,服務端會將客戶端請求進行分類,每個命令或每類請求命令定義一個處理器(NettyRequestProcessor),然后每一個NettyRequestProcessor綁定到一個單獨的線程池,進行命令處理,不同類型的請求將使用不同的線程池進行處理,實現線程隔離。

                為了方便下文的描述,我們先簡單的認識一下NettyRequestProcessor、Pair、RequestCode。其核心關鍵點如下:

                1. NettyRequestProcessor RocketMQ 服務端請求處理器,例如SendMessageProcessor是消息發送處理器、PullMessageProcessor是消息拉取命令處理器。
                2. RequestCode 請求CODE,用來區分請求的類型,例如SEND_MESSAGE:表示該請求為消息發送,PULL_MESSAGE:消息拉取請求。
                3. Pair 用來封裝NettyRequestProcessor與ExecuteService的綁定關系。在RocketMQ的網絡處理模型中,會為每一個NettyRequestProcessor與特定的線程池綁定,所有該NettyRequestProcessor的處理邏輯都在該線程池中運行。
                2.2 pair.getObject1().rejectRequest()

                由于讀者朋友提出的問題,都是發生在消息發送過程中,故本文重點關注SendMessageProcessor#rejectRequest方法。

                SendMessageProcessor#rejectRequest

                1public boolean rejectRequest() {
                2    return this.brokerController.getMessageStore().isOSPageCacheBusy() ||               // @1
                3        this.brokerController.getMessageStore().isTransientStorePoolDeficient();        // @2
                4}
                復制

                拒絕請求的條件有兩個,只要其中任意一個滿足,則返回true。

                代碼@1:Os PageCache busy,判斷操作系統PageCache是否繁忙,如果忙,則返回true。想必看到這里大家肯定與我一樣好奇,RocketMQ是如何判斷pageCache是否繁忙呢?下面會重點分析。

                代碼@2:transientStorePool是否不足。

                2.2.1 isOSPageCacheBusy()

                DefaultMessageStore#isOSPageCacheBusy()

                1public boolean isOSPageCacheBusy() {
                2    long begin = this.getCommitLog().getBeginTimeInLock();  // @1 start
                3    long diff = this.systemClock.now() - begin;                         // @1  end
                4
                5    return diff < 10000000
                6                && diff > this.messageStoreConfig.getOsPageCacheBusyTimeOutMills();     // @2
                7}
                復制

                代碼@1:先重點解釋begin、diff兩個局部變量的含義:

                • begin 通俗的一點講,就是將消息寫入Commitlog文件所持有鎖的時間,精確說是將消息體追加到內存映射文件(DirectByteBuffer)或pageCache(FileChannel#map)該過程中開始持有鎖的時間戳,具體的代碼請參考:CommitLog#putMessage。
                • diff 一次消息追加過程中持有鎖的總時長,即往內存映射文件或pageCache追加一條消息所耗時間。

                代碼@2:如果一次消息追加過程的時間超過了Broker配置文件osPageCacheBusyTimeOutMills,則認為pageCache繁忙,osPageCacheBusyTimeOutMills默認值為1000,表示1s。

                2.2.2 isTransientStorePoolDeficient()

                DefaultMessageStore#isTransientStorePoolDeficient

                1public boolean isTransientStorePoolDeficient() {
                2    return remainTransientStoreBufferNumbs() == 0;
                3}
                4public int remainTransientStoreBufferNumbs() {
                5    return this.transientStorePool.remainBufferNumbs();
                6}
                復制

                最終調用TransientStorePool#remainBufferNumbs方法。

                1public int remainBufferNumbs() {
                2        if (storeConfig.isTransientStorePoolEnable()) {
                3            return availableBuffers.size();
                4        }
                5        return Integer.MAX_VALUE;
                6}
                復制

                如果啟用transientStorePoolEnable機制,返回當前可用的ByteBuffer個數,即整個isTransientStorePoolDeficient方法的用意是是否還存在可用的ByteBuffer,如果不存在,即表示pageCache繁忙。那什么是transientStorePoolEnable機制呢?

                2.3 漫談transientStorePoolEnable機制

                Java NIO的內存映射機制,提供了將文件系統中的文件映射到內存機制,實現對文件的操作轉換對內存地址的操作,極大的提高了IO特性,但這部分內存并不是常駐內存,可以被置換到交換內存(虛擬內存),RocketMQ為了提高消息發送的性能,引入了內存鎖定機制,即將最近需要操作的commitlog文件映射到內存,并提供內存鎖定功能,確保這些文件始終存在內存中,該機制的控制參數就是transientStorePoolEnable。

                2.3.1 MappedFile

                重點關注MappedFile的ByteBuffer writeBuffer、MappedByteBuffer mappedByteBuffer這兩個屬性的初始化,因為這兩個方法是寫消息與查消息操作的直接數據結構。

                兩個關鍵點如下:

                • ByteBuffer writeBuffer 如果開啟了transientStorePoolEnable,則使用ByteBuffer.allocateDirect(fileSize),創建(java.nio的內存映射機制)。如果未開啟,則為空。
                • MappedByteBuffer mappedByteBuffer 使用FileChannel#map方法創建,即真正意義上的PageCache。

                消息寫入時:

                MappedFile#appendMessagesInner

                從中可見,在消息寫入時,如果writerBuffer不為空,說明開啟了transientStorePoolEnable機制,則消息首先寫入writerBuffer中,如果其為空,則寫入mappedByteBuffer中。

                消息拉取(讀消息):

                MappedFile#selectMappedBuffer

                消息讀取時,是從mappedByteBuffer中讀(pageCache)。

                大家是不是發現了一個有趣的點,如果開啟transientStorePoolEnable機制,是不是有了讀寫分離的效果,先寫入writerBuffer中,讀卻是從mappedByteBuffer中讀取。

                為了對transientStorePoolEnable引入意圖闡述的更加明白,這里我引入Rocketmq社區貢獻者胡宗棠關于此問題的見解。

                通常有如下兩種方式進行讀寫:

                1. 第一種,Mmap+PageCache的方式,讀寫消息都走的是pageCache,這樣子讀寫都在pagecache里面不可避免會有鎖的問題,在并發的讀寫操作情況下,會出現缺頁中斷降低,內存加鎖,污染頁的回寫。
                2. 第二種,DirectByteBuffer(堆外內存)+PageCache的兩層架構方式,這樣子可以實現讀寫消息分離,寫入消息時候寫到的是DirectByteBuffer——堆外內存中,讀消息走的是PageCache(對于,DirectByteBuffer是兩步刷盤,一步是刷到PageCache,還有一步是刷到磁盤文件中),帶來的好處就是,避免了內存操作的很多容易堵的地方,降低了時延,比如說缺頁中斷降低,內存加鎖,污染頁的回寫。

                溫馨提示:如果想與胡宗棠大神進一步溝通交流,可以關注他的github賬號:https://github.com/zongtanghu

                不知道大家會不會有另外一個擔憂,如果開啟了transientStorePoolEnable,內存鎖定機制,那是不是隨著commitlog文件的不斷增加,最終導致內存溢出?

                2.3.2 TransientStorePool初始化

                從這里可以看出,TransientStorePool默認會初始化5個DirectByteBuffer(對外內存),并提供內存鎖定功能,即這部分內存不會被置換,可以通過transientStorePoolSize參數控制。

                在消息寫入消息時,首先從池子中獲取一個DirectByteBuffer進行消息的追加。當5個DirectByteBuffer全部寫滿消息后,該如何處理呢?從RocketMQ的設計中來看,同一時間,只會對一個commitlog文件進行順序寫,寫完一個后,繼續創建一個新的commitlog文件。故TransientStorePool的設計思想是循環利用這5個DirectByteBuffer,只需要寫入到DirectByteBuffer的內容被提交到PageCache后,即可重復利用。對應的代碼如下: TransientStorePool#returnBuffer

                1public void returnBuffer(ByteBuffer byteBuffer) {
                2    byteBuffer.position(0);
                3    byteBuffer.limit(fileSize);
                4    this.availableBuffers.offerFirst(byteBuffer);
                5}
                復制

                其調用棧如下:

                從上面的分析看來,并不會隨著消息的不斷寫入而導致內存溢出。

                現象解答3.1 [REJECTREQUEST]system busy

                其拋出的源碼入口點:NettyRemotingAbstract#processRequestCommand,上面的原理分析部分已經詳細介紹其實現原理,總結如下。

                在不開啟transientStorePoolEnable機制時,如果Broker PageCache繁忙時則拋出上述錯誤,判斷PageCache繁忙的依據就是向PageCache追加消息時,如果持有鎖的時間超過1s,則會拋出該錯誤;在開啟transientStorePoolEnable機制時,其判斷依據是如果TransientStorePool中不存在可用的堆外內存時拋出該錯誤。

                3.2 too many requests

                其拋出的源碼入口點:NettyRemotingAbstract#processRequestCommand,其調用地方緊跟3.1,是在向線程池執行任務時,被線程池拒絕執行時拋出的,我們可以順便看看Broker消息處理發送的線程信息: BrokerController#registerProcessor

                該線程池的隊列長度默認為10000,我們可以通過sendThreadPoolQueueCapacity來改變默認值。

                3.3 [PC_SYNCHRONIZED]broker busy

                其拋出的源碼入口點:DefaultMessageStore#putMessage,在進行消息追加時,再一次判斷PageCache是否繁忙,如果繁忙,則拋出上述錯誤。

                3.4 broker busy, period in queue: %sms, size of queue: %d

                其拋出源碼的入口點:BrokerFastFailure#cleanExpiredRequest。該方法的調用頻率為每隔10s中執行一次,不過有一個執行前提條件就是Broker端要開啟快速失敗,默認為開啟,可以通過參數brokerFastFailureEnable來設置。該方法的實現要點是每隔10s,檢測一次,如果檢測到PageCache繁忙,并且發送隊列中還有排隊的任務,則直接不再等待,直接拋出系統繁忙錯誤,使正在排隊的線程快速失敗,結束等待。

                實踐建議

                經過上面的原理講解與現象分析,消息發送時拋出system busy、broker busy的原因都是PageCache繁忙,那是不是可以通過調整上述提到的某些參數來避免拋出錯誤呢?.例如如下參數:

                • osPageCacheBusyTimeOutMills 設置PageCache系統超時的時間,默認為1000,表示1s,那是不是可以把增加這個值,例如設置為2000或3000。作者觀點:非常不可取。
                • sendThreadPoolQueueCapacity Broker服務器處理的排隊隊列,默認為10000,如果隊列中積壓了10000個請求,則會拋出RejectExecutionException。作者觀點:不可取。
                • brokerFastFailureEnable 是否啟用快速失敗,默認為true,表示當如果發現Broker服務器的PageCache繁忙,如果發現sendThreadPoolQueue隊列中不為空,表示還有排隊的發送請求在排隊等待執行,則直接結束等待,返回broker busy。那如果不開啟快速失敗,則同樣可以避免拋出這個錯誤。作者觀點:非常不可取。

                修改上述參數,都不可取,原因是出現system busy、broker busy這個錯誤,其本質是系統的PageCache繁忙,通俗一點講就是向PageCache追加消息時,單個消息發送占用的時間超過1s了,如果繼續往該Broker服務器發送消息并等待,其TPS根本無法滿足,哪還是高性能的消息中間了呀。故才會采用快速失敗機制,直接給消息發送者返回錯誤,消息發送者默認情況會重試2次,將消息發往其他Broker,保證其高可用。

                下面根據個人的見解,提出如下解決辦法:

                4.1 開啟transientStorePoolEnable

                在broker配置文件中將transientStorePoolEnable設置為true。

                • 方案依據: 啟用“讀寫”分離,消息發送時消息先追加到DirectByteBuffer(堆外內存)中,然后在異步刷盤機制下,會將DirectByteBuffer中的內容提交到PageCache,然后刷寫到磁盤。消息拉取時,直接從PageCache中拉取,實現了讀寫分離,減輕了PageCaceh的壓力,能從根本上解決該問題。
                • 方案缺點: 會增加數據丟失的可能性,如果Broker JVM進程異常退出,提交到PageCache中的消息是不會丟失的,但存在堆外內存(DirectByteBuffer)中但還未提交到PageCache中的這部分消息,將會丟失。但通常情況下,RocketMQ進程退出的可能性不大。
                4.2 擴容Broker服務器

                方案依據:

                當Broker服務器自身比較忙的時候,快速失敗,并且在接下來的一段時間內會規避該Broker,這樣該Broker恢復提供了時間保證,Broker本身的架構是支持分布式水平擴容的,增加Topic的隊列數,降低單臺Broker服務器的負載,從而避免出現PageCache。

                溫馨提示:在Broker擴容時候,可以復制集群中任意一臺Broker服務下${ROCKETMQ_HOME}/store/config/topics.json到新Broker服務器指定目錄,避免在新Broker服務器上為Broker創建隊列,然后消息發送者、消息消費者都能動態獲取Topic的路由信息。

                與之擴容對應的,也可以通過對原有Broker進行升配,例如增加內存、把機械盤換成SSD,但這種情況,通常需要重啟Broekr服務器,沒有擴容來的方便。

                關聯標簽:
                皇宫里的共享小公主

                  <form id="jrvnn"></form>

                      <address id="jrvnn"><listing id="jrvnn"></listing></address>
                      <form id="jrvnn"><nobr id="jrvnn"><meter id="jrvnn"></meter></nobr></form>

                        <sub id="jrvnn"><listing id="jrvnn"></listing></sub>