RocketMQ客戶端PUSH消費(fèi)中并發(fā)消費(fèi)與順序消費(fèi)的示例分析

這篇文章主要介紹了RocketMQ客戶端PUSH消費(fèi)中并發(fā)消費(fèi)與順序消費(fèi)的示例分析，具有一定借鑒價(jià)值，感興趣的朋友可以參考下，希望大家閱讀完這篇文章之后大有收獲，下面讓小編帶著大家一起了解一下。

一、消息拉取與處理

1.消息拉取

小結(jié)：PullMessageService處理拉取消息請求。通過組織RequetHeader需要包含從哪里開始拉取（ConsumerGroup、Topic，Queue，queueOffset）等信息，向Broker發(fā)起請求，取回消息后對消息進(jìn)行處理。當(dāng)該Queue的消息數(shù)量超過1000，或者最小與最大偏移量之間的差距超過默認(rèn)2000也會(huì)觸發(fā)限流，即：延遲50毫秒放入請求隊(duì)列。也可以通過掛起消費(fèi)線程來延遲（1秒）消息拉取，從而達(dá)到消費(fèi)限流作用。

2.消息處理

小結(jié)：PullMessageService處理消息部分流程：將消息提交給了processQueue紅黑樹緩存；同時(shí)將消息提交給consumeMessageService來處理具體的消息內(nèi)容。

二、并發(fā)消費(fèi)流程

1.ConsumeMessageConcurrentlyService職責(zé)

小結(jié)：ConsumeMessageService并發(fā)消費(fèi)（ConsumeMessageConcurrentlyService）主要工作交給Listener（客戶端傳入）進(jìn)行處理，并對處理結(jié)果進(jìn)行統(tǒng)計(jì)和處理；對于失敗消息，廣播消費(fèi)會(huì)丟棄，集群消費(fèi)會(huì)發(fā)回Broker重新消費(fèi)；清理ProcessQueue并更新緩存（offsetTable）消費(fèi)進(jìn)度。

三、順序消費(fèi)流程
1.ConsumeMessageOrderlyService職責(zé)

小結(jié)：順序消費(fèi)流程跟并發(fā)消費(fèi)最大的區(qū)別在于，對要處理的隊(duì)列加鎖，確保同一隊(duì)列，同一時(shí)間，只允許一個(gè)消費(fèi)線程處理。

1.為什么順序消費(fèi)時(shí)需要對Broker發(fā)請求對要處理的隊(duì)列加鎖？
2.對Broker端隊(duì)列加鎖流程是怎么樣的？
3.既然加鎖了需要解鎖嗎？
4.會(huì)存在Broker加鎖過期了客戶端還在處理該隊(duì)列的情況嗎？

2.Broker端隊(duì)列加鎖流程

小結(jié)：順序消費(fèi)時(shí)對Broker端隊(duì)列加鎖防止該隊(duì)列在特定時(shí)間內(nèi)（一次默認(rèn)60秒）被分配給其他clientId處理；Broker端加鎖了，一次加鎖失效時(shí)長為60秒；不存在Broker加鎖過期了客戶端還在處理該隊(duì)列的情況，Broker加鎖時(shí)長為60秒，而客戶端加鎖時(shí)長為30秒，當(dāng)客戶端加鎖時(shí)長失效時(shí)會(huì)重新請求Broker加鎖并更新時(shí)間戳，從而可以持續(xù)延長加鎖時(shí)間。

四、交互示意圖

感謝你能夠認(rèn)真閱讀完這篇文章，希望小編分享的“RocketMQ客戶端PUSH消費(fèi)中并發(fā)消費(fèi)與順序消費(fèi)的示例分析”這篇文章對大家有幫助，同時(shí)也希望大家多多支持億速云，關(guān)注億速云行業(yè)資訊頻道，更多相關(guān)知識(shí)等著你來學(xué)習(xí)!