四vip課程分布式框架專題32api源碼分析rocketmq原理解析

前 第一章： 第二章 三：長(zhǎng)輪 六：pull消息消 七 第三章： 三 四：HA&master HA異步 ：Broker 第四章： 一：Namesrv功 二：Namesrv啟動(dòng)流程 三 一： Server 公用抽象 invokeSyncImpl同步調(diào)用實(shí) invokeAsyncImpl異步調(diào)用實(shí) invokeOnewayImpl單向請(qǐng) mand接收請(qǐng)求處 mand接收響應(yīng)處 二：NettyRemotingServerRemoting服務(wù)端實(shí) 前此文檔是從學(xué)習(xí)mq源碼過(guò)程中的筆記中整理出來(lái)的，由于時(shí)間及能力原因，理 mq有所幫助。mq是阿里基于開(kāi)源思一款產(chǎn)品，代碼托管于上，要想學(xué)好用mq請(qǐng)從 MQ獲取最的文檔、問(wèn)題解答、原理介第一章一：Producer啟動(dòng)流Producerproducer信息（procduergroup）定時(shí)發(fā)送到，brokerAddrTable集合中列出的broker上去Producer發(fā)送消息只發(fā)送到master的broker機(jī)器，在通過(guò)broker的主從機(jī)制拷貝二 如何發(fā)送消Broker1的隊(duì)列為queue0,queue1Broker3當(dāng)然一般情況下的brokerbroker1_queue0,broker1_queue1,> queue都會(huì)通過(guò)sendWhichQueuequeue的輪詢lastBrokerNamequeue發(fā)送失敗，這次選擇應(yīng)該避開(kāi)同一個(gè)queue發(fā)送失敗后，重試幾次retryTimesWhenSendFailed=-- -- //代表發(fā)送消息到達(dá)的 向指定broker的指定topic的指定queue發(fā)送消息發(fā)送失敗(1)重試次數(shù)不到兩次(2)3000(毫秒),換 發(fā)送普通消順序消息發(fā)mq能夠保證消息嚴(yán)格順序，但是mqproducer保證順序消息按順序發(fā)送到同一個(gè)queue中，比如流程(1)下單(2)支付(3)支付成功，這三個(gè)消息需要根據(jù)特定規(guī)則將這個(gè)三個(gè)消息按順序發(fā)送到一個(gè)queue一般消息是通過(guò)輪詢所有隊(duì)列發(fā)送的，順序消息可以根據(jù)業(yè)務(wù)比如說(shuō)訂單號(hào),messageQueueList分布式事物消一階段，向broker發(fā)送一條prepared的消息，返回消息的offsetLocalTransactionExecuter二階段，處理完本地事物務(wù)得到事物狀態(tài)，根據(jù)offset查找到commitLog中的prepared消息，設(shè)置消息狀態(tài)commitType或者rollbackType，讓后將信息添加到commitLog中，其實(shí)二階段生成了兩條消息三：Broker落地消普通消息落Broker根據(jù)producer請(qǐng)求的 ode.SEND_MESSAGE選擇對(duì)應(yīng)的處理分布式事物消息落commitLog20位開(kāi)始的八位記錄是的消息在邏輯隊(duì)列中的queueoffset，但是針對(duì)事物消息為preparedType和rollbackType的的是事物狀態(tài)ConsumeQueuepreparedType和rollbackType的消息不會(huì)將請(qǐng)求分發(fā)到ConsumeQueue中去，即不處理，所以不會(huì)被消息transactionstabletableprepared消息記，通過(guò)TransactionStateService服務(wù)將消息加到事務(wù)狀態(tài)的表格tranStateTable的文件中；如果是commitType和rollbackType消息，修改事物狀態(tài)表格tranStateTable中的消息狀態(tài)。記錄TransactionRedoLog日志：記錄了commitLogOffset,msgSize,preapredTransactionOffset,storeTimestamp。事物狀態(tài)表是有MapedFileQueue將多個(gè)文件組成續(xù)的隊(duì)列，它的單元是定長(zhǎng)為24個(gè)字節(jié)的數(shù)據(jù)，tranStateTableOffset可以認(rèn)為是事物狀態(tài)消息的個(gè)數(shù)，索引偏移量，它的值是定時(shí)回查線程會(huì)定時(shí)掃描（默認(rèn)每分鐘）每個(gè)事務(wù)狀態(tài)的表格文件，遍歷事prepared狀態(tài)的如果消息小于事務(wù)回查至少間隔時(shí)間（默認(rèn)是一分鐘）跳出終根據(jù)group隨機(jī)選擇一臺(tái)producer建一個(gè)Runnable任務(wù)異步執(zhí)行producer的回調(diào)接口，處理回調(diào)，在調(diào)endTransactionOneway向broker發(fā)送請(qǐng)求更新事物消息的最終狀態(tài)事物消息的()tranRedoLog文件的消息DispatchRequest,更新TransactionState記錄TransactionRedo刪除事物狀態(tài)表重建 將上面過(guò)濾出的prepared消息，添加到事物狀態(tài)表文件redologcommitlog來(lái)做的第二拉模式拉取消息，consumer做負(fù)載均衡并通過(guò)長(zhǎng)輪詢向broker拉消息。 mq的一 啟動(dòng)流消費(fèi)端Start向mq Factory本消費(fèi)者啟動(dòng)端通信定時(shí)清理下線的borker定時(shí)持久化Consumer消費(fèi)進(jìn)度（廣播到本地，集群到Broker）啟動(dòng)拉消息服務(wù)namesrv更新topic路由信息喚醒Rebalance服務(wù)線程二：消費(fèi)端負(fù)載均消費(fèi)端會(huì)通過(guò)RebalanceService線程，10topic下的所有隊(duì)列負(fù)載消費(fèi)端遍歷自己的所有topic，依次調(diào)rebalanceByTopic選擇一臺(tái)broker獲取基于group的所有消費(fèi)端（有心跳向所有broker客戶端信息）選擇隊(duì)列分配策略實(shí)例AllocateMessageQueueStrategy執(zhí)行分配算法獲取隊(duì)列集合Set<MessageQueue>mqSet將所有queue排好序類似于記錄3）按照機(jī)房來(lái)配置隊(duì)列獲取指定機(jī)房的queue,置ProcessQueue的droped屬性為true構(gòu)建這個(gè)隊(duì)列的ProcessQueue注：ProcessQueue正在被消費(fèi)的隊(duì)列，長(zhǎng)輪詢拉取到消息都會(huì)先到ProcessQueue的TreeMap<Long,的processqueue才能被執(zhí)行消費(fèi)rollback:將消費(fèi)在msgTreeMapTemp中的消息，放回msgTreeMap重新消費(fèi)mq的消息是由consumer端主動(dòng)到broker拉取的consumerbroker發(fā)送拉消息PullMessageServiceLinkedBlockingQueue<PullRequest>中的PullRequest到broker拉取消息2000,從內(nèi)存中獲取 //TODO這個(gè)值跟pullRequest.getNextOffset區(qū)構(gòu)建 pull接口用到的消息中放入隊(duì)列最大最小offset,方便應(yīng)用來(lái)感知消息堆積度pullInterval0（拉消息間隔，如果為了降低拉取速度，0的值pullInterval0立刻在執(zhí)行拉四 消息—并發(fā)消費(fèi)消通過(guò)長(zhǎng)輪詢拉取到消息后會(huì)提交到消息服務(wù)ConsumeMessageConcurrentlyService，ConsumeMessageConcurrentlyServic的submitConsumeRequest方法構(gòu)建ConsumeRequest消費(fèi)端consumer構(gòu)建一個(gè)消費(fèi)消息任務(wù)ConsumeRequest消費(fèi)一批消息的個(gè)數(shù)是可配置的consumeMessageBat axSize=1，默認(rèn)批量個(gè)數(shù)為一個(gè)ConsumeRequest任務(wù)run給消息設(shè)置消費(fèi)失敗時(shí)候的retrytopic，當(dāng)消息發(fā)送失敗的時(shí)候發(fā)送到

ackIndex來(lái)標(biāo)記成ackIndex設(shè)置為-秒鐘以后重試重新消費(fèi)消息，在走一次上面的消費(fèi)流程。offsetTable的更新，后面有定時(shí)任務(wù)定時(shí)更新到broker上去五：push消費(fèi)-順序消費(fèi)消構(gòu)建一個(gè)線程池來(lái)接收消費(fèi)請(qǐng)求ConsumeRequest構(gòu)建一個(gè)單線程的本地線程,ConsumeRequest,用來(lái)執(zhí)行,Map<brokernameSet<MessageQueue>>brokernamebrokermaster機(jī)器地址brokerNameSet<MessageQueue>發(fā)送broker請(qǐng)求鎖定這些隊(duì)列。brokerbrokerqueuequeue被某個(gè)BrokerMessageQueue,設(shè)置對(duì)應(yīng)的正在處理隊(duì)列ProccessQueuelocked屬性為true沒(méi)有鎖定設(shè)置為false通過(guò)長(zhǎng)輪詢拉取到消息后會(huì)提交到消息服務(wù)ConsumeMessageOrderlyService，交到線程池。ConsumeRequest是由ProcessQueue和Messagequeue組成。messagequeueconsumer內(nèi)，同一個(gè)隊(duì)列串行truelock放到本地線程稍后鎖定在消費(fèi)。如果locktrue且沒(méi)有過(guò)期，開(kāi)始消費(fèi)消息計(jì)算任務(wù)執(zhí)行的時(shí)間如果大于一分鐘且線程數(shù)小于隊(duì)列數(shù)情況下，將processqueue，proccessqueue獲取批次消息，processqueue.takeMessags(batchSize從msgTreeMap mit方新到broker上去六：pull消息消列七 利用DefaultMQPushConsumerImpl 獲取ProcessQueueTable<MessageQueue,ProcessQueue>的keyset的messagequeue去獲取offsetTable中的messagequeue的值，在update的時(shí)候如果沒(méi)有對(duì)應(yīng)的Messagequeue會(huì)構(gòu)建，但是也會(huì)rebalance的時(shí)候?qū)](méi)有分配到的messagequeue刪除offsetbrokerUnregiser客戶端 第三章一：brker的啟BrokernamesrvTopic在broker文件上的json格式 }Namesrv接收Broker的topic信息，namesrv只存內(nèi)存，但是broker有任務(wù)定時(shí)推 Mq通過(guò)ConfigMananger來(lái)管理配置加載以及持久{"%RETRY%group_name":{ //100讀權(quán)限,10寫(xiě)權(quán) 6是110讀寫(xiě)權(quán) }}}加載消費(fèi)進(jìn)度偏移量{"%RETRYgroup_name@group_name":{0:0重試隊(duì)列消費(fèi)group_name":{//隊(duì)列 //隊(duì)列 }}}{" "groupName":"group_name",}}}二：消mq的消息的是由consumequeue和commitLog配合完成consumequeue消息的邏輯隊(duì)列，相當(dāng)于字典的 文件commitLog上的位置,文件地址：${user.home}\store\consumequeue\${ consumequeue中單元是一個(gè)20字節(jié)定長(zhǎng)的數(shù)據(jù)，是順序?qū)戫樞蜃xcommitLogOffset是指這條消息在commitLog成一個(gè)ConsumeQueue,TopicAQueue=1組成一個(gè)另一個(gè)ConsumeQueue. //消息體BODY 這個(gè)標(biāo)志 //QUEUEOFFSETconsumequeue的偏移量，可以代表這個(gè)隊(duì)列中消息的個(gè)數(shù)，要通過(guò)這個(gè)值查找到consumequeue中數(shù)據(jù)，QUEUEOFFSET*20才是偏移地址+8+4+8+8 +8+8+8//RECONSUMETIMES消息被某個(gè)訂閱組重新消費(fèi)了幾次（立計(jì)數(shù)）,topic名字為%retry%groupNamequeueId=0的隊(duì)列 //4個(gè)字節(jié)存放消息體大小值，后bodylength大小 //topictopic + //2個(gè)字節(jié)（short）存放屬性值大小，后存放MapedFilePageCache文件封裝，操作物理文件在內(nèi)存中的映射以及將內(nèi)存數(shù)據(jù)持久化到物理文件中，代碼中寫(xiě)死了要求os系統(tǒng)的頁(yè)大小為4k,消息刷盤(pán)根據(jù)參數(shù)FileChannelfileChannel=newRandomAccessFile(file,fileSize:filename:filename是消息在此文件的中初始偏移量，排好序后組成了續(xù)的消息隊(duì)getLastMapedFile獲取，此函數(shù)如果集合中一個(gè)也沒(méi)有創(chuàng)建一個(gè)，如果最后一個(gè)寫(xiě)滿了也創(chuàng)一個(gè)MapedFile文件地址，通過(guò)預(yù)分配服務(wù)AllocateMapedFileService異步預(yù)創(chuàng)建下一個(gè)MapedFile1G的文mitWhereDefaultMessageStore消息層實(shí)<1>從mapedFileQueue獲取的映射文 <5>三LoadLoadcommit遍歷出${user.homestore\${commitlog}commitLog文件，按文件名（文件名就是文件的初始偏移量MapedFile對(duì)象,在MapedFileQueue中用集合list把這些MapedFile文件組成一個(gè)邏輯上連續(xù)的隊(duì)列Loadconsume遍歷${user.homestore\consumequeue下的所有文件夾（topic就是一個(gè)文件}\queueId，文件來(lái)構(gòu)建ConsueQueue對(duì)象DefaultMessageStore中結(jié)構(gòu)Map<topic,Map<queueId,ConsumequeueMapedFileQueuemapedFile組成一個(gè)邏輯上連續(xù)的隊(duì)加載${user.home}\store\checkpoint這個(gè)文件了3個(gè)long類型的值來(lái)記錄模型最終一致的時(shí)間點(diǎn)，這個(gè)3個(gè)long的值為physicMsgTimestamp為commitLog最后刷盤(pán)的時(shí)間indexMsgTimestamp為索引最終刷盤(pán)時(shí)間abort,linuxviabort文件存在，系統(tǒng)認(rèn)為是異?；譃槭裁凑f(shuō)先正?；謴?fù)異常恢復(fù)在哪呢？broker是異常啟動(dòng)時(shí)候異常恢復(fù)commitLog時(shí)會(huì)重新構(gòu)建請(qǐng)到DispatessageService服務(wù)，來(lái)重新生成ConsumeQueue數(shù)據(jù)，索引以及事物消息的redolog,確，現(xiàn)在文件寫(xiě)到哪了(wrotePositionFlush到了什么位置(committedPosition)?ConsumeQueuemapedFiles集合中，從倒數(shù)第三個(gè)文件開(kāi)始恢復(fù)(為什元是20字節(jié)的定長(zhǎng)數(shù)據(jù)，所以是依次分別取了Offsetlong類型了commitLog的數(shù)據(jù)偏移量 int類型了在commitLog上消息大小tagcodetag的哈希值目前 mq判斷的consumequeue數(shù)據(jù)是否有效的方式為判斷offset>=0&&size>0設(shè)置 所在文件 的 commitedPosition消息大小值msgSize大于 文件錯(cuò)誤恢復(fù)結(jié)束等于0 當(dāng)${user.home}\store\abort文件存在，代表異?；謴?fù)${user.homestore\checkpoint獲取最終一致的時(shí)間點(diǎn)說(shuō)明checkpoint在此mapedfile文件中從checkpoint所在mapedFile開(kāi)始恢復(fù)數(shù)據(jù)，它的整體過(guò)程跟正常恢復(fù)（1）消息后派送到分發(fā)消息服務(wù)Dispat 建ConsumeQueue以及索引 四：HA&masterbroker啟動(dòng)的時(shí)候BrokerControllerslavemaster地址更新，沒(méi)有配置所有broker會(huì)想namesrv，從namesrv獲取haServerAddr，然后更新到HA當(dāng)HA 的Mas ddress不為空的時(shí)（因?yàn)閎rokermaster和slave都構(gòu)建了HA 會(huì)主動(dòng)連接master獲取SocketChannelMasterSlave請(qǐng)求的端口，默認(rèn)為服務(wù)端口+1接收slave上傳的offsetlong類型intposthis.byteBufferRead.position()(this.byteBufferRead.position(8)//沒(méi)有理解意圖longreadOffset=this.byteBufferRead.getLong(pos-8); 同步topic的配置信息 Master通過(guò)AcceptSocketServiceslave的連接，每個(gè)masterslave連接都會(huì)構(gòu)建一個(gè)HAConnectionmasterslave部署結(jié)構(gòu)的會(huì)有多個(gè)HAConnection實(shí)例，MasterHAConnectionslaveWriteSocketService對(duì)象和Slave反饋服務(wù)線程對(duì)象ReadSocketServiceslaveRequestOffsetslave同步完數(shù)據(jù)都會(huì)向masterack表示下次同步的數(shù)據(jù)的offset。slaveslaveRequestOffset=0mastercommitLog文件開(kāi)始同步（如果要把master上的所有commitLog文件同步到slave的話，把masterOffset值賦為minOffset）socket<PhyOffset><BodySize>Bodyslave通過(guò)HA 通知前端線程，如果是同步的話通知是否成功Slave通過(guò) masterslavecommitlog的數(shù)據(jù)，masterslave寫(xiě)入數(shù)據(jù)的格式是Slave初始化DefaultMessageStore時(shí)候會(huì)構(gòu)建ReputMessageService服務(wù)線程并在啟動(dòng)服務(wù)的start方法中被啟動(dòng)HA同步，當(dāng)msg寫(xiě)入master的commitlog文件后，判斷maser的角色如果是同步雙SYNC_MASTER，等待master同步到slave在返回結(jié)果HA異步五：刷盤(pán)策,六：索引服索引結(jié)索引文件由索引文件頭IndexHeader,Slot: 8位long類型，索引文件構(gòu)建第一個(gè)索引的消息落在broker的時(shí)間 8位long類型，索引文件構(gòu)建最后一個(gè)索引消息落broker時(shí)間 8位long類型，索引文件構(gòu)建第一個(gè)索引的消息commitLog偏移量 8位long類型，索引文件構(gòu)建最后一個(gè)索引消息commitLog偏移量 4位int類型，構(gòu)建索引占用的槽位數(shù)(這個(gè)值貌似沒(méi)有具體作用) 4位int類型，索引文件中構(gòu)建的索引個(gè)數(shù)計(jì)算消息的對(duì)應(yīng)的slotPos=Math.abs(keyHash)%hashSlotNum消息在IndexFile中的偏移量absSlotPos=IndexHeader.INDEX_HEADER_SIZE+slotPos*4位int值，的是key的hash8位long值的是消息在commitlog的物理偏移量4位int 4位int ：索引服IndexService文件indexFile的路徑queryOffsettopickey時(shí)間跨度來(lái)查詢消息：構(gòu)建索引服分發(fā)消息索引服務(wù)將消息位置分發(fā)到ConsumeQueue中后，加入IndexService構(gòu)建索引key(topic+"#"+key)計(jì)算槽位在indexfile的具體偏移量位置：Broker （comsumer，producer）之間的心跳 ChannelInfo的時(shí)間戳，來(lái)表 , HouseKeeService線程定時(shí)清除不活動(dòng)的連接 ：Broker與namesrv之間的（1）broker的topic信構(gòu)建或者更新第四章接收broker的請(qǐng)求broker路由信息（包括master和slave） 的請(qǐng)求根據(jù)某個(gè)topic獲取所有到broker的路由信息二 啟動(dòng)流程三它的多個(gè)slave id表示是master還是slaveid=0為 大于0為, BrokerLiveInfo代表一個(gè)活的broker由最后更新時(shí)間，一個(gè)channel，數(shù)據(jù)版本和Ha地址組成四：Namesrv與broker間的心也是topic信息到namesrvNamesrvControllerRouteInfoMangerscanNotActiveBroker方法來(lái)定時(shí)清理不活動(dòng)的broker（默認(rèn)兩分鐘沒(méi)有向namesrv發(fā)送心跳更新BrokerLiveInfo時(shí)間戳的BrokerLiveInfo的時(shí)間戳，如果過(guò)期關(guān)閉channel連接第五Remoting通信層一： Server 公用抽象過(guò)opaque從緩存查找對(duì)應(yīng)的ResponseFuture對(duì)象invokeSyncImpl同步調(diào)用應(yīng)過(guò)來(lái)，就用不到緩存中的ResponseFuturel）invokeAsyncImpl異步調(diào)用限默認(rèn)2048個(gè)發(fā)送成功 dRequestOK(false),信號(hào)量通過(guò)once釋放， 單向請(qǐng)mand接收請(qǐng)求根據(jù)請(qǐng)求code查找對(duì)應(yīng)的處理器線程池pair，沒(méi)有用默認(rèn)的 若不是o