深入理解RocketMQ：Consumer消费消息原理
前言
RocketMQ版本：4.8
Consumer類型：DefaultMQPushConsumer
原理解析
consumer 啟動(dòng)時(shí)做了哪些事情？
定時(shí)從NameSrv獲取最新的Topic+Queue+Broker路由信息，獲取頻率默認(rèn)30秒，可以通過(guò)參數(shù) pollNameServerInterval 進(jìn)行設(shè)置
創(chuàng)建類型為L(zhǎng)inkedBlockingQueue的隊(duì)列pullRequestQueue，用于存放PullRequest請(qǐng)求對(duì)象。
創(chuàng)建線程RebalanceService，主要用于執(zhí)行負(fù)載均衡。 默認(rèn)根據(jù)平均分配原則，為當(dāng)前consumer分配對(duì)應(yīng)數(shù)量的queue。隨后遍歷queue分別創(chuàng)建PullRequest，放入隊(duì)列pullRequestQueue。
PS：這項(xiàng)工作默認(rèn)20秒執(zhí)行一次，可以通過(guò)System.setProperty(“rocketmq.client.rebalance.waitInterval”) 修改
Queue分配完畢后，如果線程是第一次運(yùn)行，會(huì)向Broker發(fā)送請(qǐng)求獲取當(dāng)前Consumer對(duì)應(yīng)每個(gè)Queue的消費(fèi)進(jìn)度。取值結(jié)果受Consumer消費(fèi)策略 consumeFromWhere 影響，默認(rèn)是獲取上次消費(fèi)的進(jìn)度。
創(chuàng)建線程PullMessageService，不斷地從pullRequestQueue拉取數(shù)據(jù)。取到PullRequest就通過(guò)Netty發(fā)送類型為RemotingCommand的消息，code=RequestCode.PULL_MESSAGE
總結(jié)
以上是生活随笔為你收集整理的深入理解RocketMQ：Consumer消费消息原理的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。
- 上一篇： RocketMQ源码解析：Produce
- 下一篇： Canal实时同步数据到RocketMQ