老弟问我，RocketMQ中的ProcessQueue怎么理

　　今天来分享RocketMQ中一个非常重要又不太好理解的知识点ProcessQueue。
　　一句话概括，ProcessQueue就是MessageQueue的消费快照。看下面这张图：
　　1ProcessQueue构建
　　RocketMQ客户端启动时，会开启一个rebalance线程，代码如下：MQClientInstance。javapublicvoidstart（）throwsMQClientException｛synchronized（this）｛switch（this。serviceState）｛caseCREATEJUST：。。。Startrebalanceservicethis。rebalanceService。start（）；。。。｝｝｝
　　这个线程会不停的做重平衡操作，对ProcessQueue进行维护。在重平衡线程类RebalanceImpl定义了一个变量processQueueTable，数据结构如下：
　　可以看到，在processQueueTable这个数据结构上维护了MessageQueue和ProcessQueue的映射。
　　下面看一下维护processQueueTable的代码：privatebooleanupdateProcessQueueTableInRebalance（finalStringtopic，finalSetMessageQueuemqSet，finalbooleanisOrder）｛booleanchangedfalse；IteratorEntryMessageQueue，ProcessQueueitthis。processQueueTable。entrySet（）。iterator（）；while（it。hasNext（））｛EntryMessageQueue，ProcessQueuenextit。next（）；MessageQueuemqnext。getKey（）；ProcessQueuepqnext。getValue（）；if（mq。getTopic（）。equals（topic））｛if（！mqSet。contains（mq））｛从processQueueTable上移除｝elseif（pq。isPullExpired（））｛switch（this。consumeType（））｛caseCONSUMEACTIVELY：拉模式break；caseCONSUMEPASSIVELY：推模式从processQueueTable上移除break；default：break；｝｝｝｝创建ProcessQueue并放到processQueueTableListPullRequestpullRequestListnewArrayListPullRequest（）；for（MessageQueuemq：mqSet）｛if（！this。processQueueTable。containsKey（mq））｛。。。ProcessQueuepqnewProcessQueue（）；longnextOffset1L；try｛nextOffsetthis。computePullFromWhereWithException（mq）；｝catch（Exceptione）｛log。info（doRebalance，｛｝，computeoffsetfailed，｛｝，consumerGroup，mq）；continue；｝if（nextOffset0）｛ProcessQueueprethis。processQueueTable。putIfAbsent（mq，pq）；if（pre！null）｛log。info（doRebalance，｛｝，mqalreadyexists，｛｝，consumerGroup，mq）；｝else｛封装好processQueueTable后再创建一个PullRequest进行消息拉取log。info（doRebalance，｛｝，addanewmq，｛｝，consumerGroup，mq）；PullRequestpullRequestnewPullRequest（）；pullRequest。setConsumerGroup（consumerGroup）；pullRequest。setNextOffset（nextOffset）；pullRequest。setMessageQueue（mq）；pullRequest。setProcessQueue（pq）；pullRequestList。add（pullRequest）；changedtrue；｝｝else｛log。warn（doRebalance，｛｝，addnewmqfailed，｛｝，consumerGroup，mq）；｝｝｝this。dispatchPullRequest（pullRequestList）；returnchanged；｝2拉取消息
　　上一节中构建ProcessQueue后，会再创建一个PullRequest，这个PullRequest封装了MessageQueue和ProcessQueue，创建成功后被放到了PullMessageService中的pullRequestQueue变量：PullMessageService。javaprivatefinalLinkedBlockingQueuePullRequestpullRequestQueuenewLinkedBlockingQueuePullRequest（）；publicvoidexecutePullRequestImmediately（finalPullRequestpullRequest）｛try｛this。pullRequestQueue。put（pullRequest）；｝catch（InterruptedExceptione）｛log。error（executePullRequestImmediatelypullRequestQueue。put，e）；｝｝
　　这里以RocketMQ的推模式为例，Consumer拉取到消息后，会进行如下处理：对拉取到的消息根据TAG再次进行过滤；更新PullRequest下次拉取的偏移量nextOffset；
　　把拉取的消息封装到ProcessQueue的msgTreeMap（放到msgTreeMap之前首先要获取到写锁treeMapLock）；封装ConsumeRequest进行消息消费；封装消息拉取请求再次进行拉取。
　　代码如下：DefaultMQPushConsumerImpl。javapublicvoidonSuccess（PullResultpullResult）｛if（pullResult！null）｛1。对拉取到的消息根据TAG再次进行过滤pullResultDefaultMQPushConsumerImpl。this。pullAPIWrapper。processPullResult（pullRequest。getMessageQueue（），pullResult，subscriptionData）；switch（pullResult。getPullStatus（））｛caseFOUND：2。更新PullRequest下次拉取的偏移量nextOffsetpullRequest。setNextOffset（pullResult。getNextBeginOffset（））；if（pullResult。getMsgFoundList（）nullpullResult。getMsgFoundList（）。isEmpty（））｛DefaultMQPushConsumerImpl。this。executePullRequestImmediately（pullRequest）；｝else｛3。把拉取的消息封装到ProcessQueue的msgTreeMapbooleandispatchToConsumeprocessQueue。putMessage（pullResult。getMsgFoundList（））；4。封装ConsumeRequest进行消息消费DefaultMQPushConsumerImpl。this。consumeMessageService。submitConsumeRequest（pullResult。getMsgFoundList（），processQueue，pullRequest。getMessageQueue（），dispatchToConsume）；5。封装消息拉取请求if（DefaultMQPushConsumerImpl。this。defaultMQPushConsumer。getPullInterval（）0）｛DefaultMQPushConsumerImpl。this。executePullRequestLater（pullRequest，DefaultMQPushConsumerImpl。this。defaultMQPushConsumer。getPullInterval（））；｝else｛DefaultMQPushConsumerImpl。this。executePullRequestImmediately（pullRequest）；｝｝break；。。。｝｝｝3消费消息
　　在上一节提到过，拉取到消息后，会把消息封装成一个ConsumeRequest，这个线程类会调用消费者定义的MessageListener进行消费处理。看一下源代码：ConsumeMessageConcurrentlyService。ConsumeRequestpublicvoidrun（）｛if（this。processQueue。isDropped（））｛log。info（themessagequeuenotbeabletoconsume，becauseitsdropped。group｛｝｛｝，ConsumeMessageConcurrentlyService。this。consumerGroup，this。messageQueue）；return；｝MessageListenerConcurrentlylistenerConsumeMessageConcurrentlyService。this。messageListener；ConsumeConcurrentlyContextcontextnewConsumeConcurrentlyContext（messageQueue）；ConsumeConcurrentlyStatusstatusnull；try｛statuslistener。consumeMessage（Collections。unmodifiableList（msgs），context）；｝。。。if（！processQueue。isDropped（））｛ConsumeMessageConcurrentlyService。this。processConsumeResult（status，context，this）；｝｝
　　消息消费成功后，会调用processConsumeResult方法进行结果处理。对于广播模式，发送失败后不会做重试，相当于把消息丢弃，而对于集群模式，消费失败的消息会发送到Broker端等待消费者重新拉取进行重试。
　　消费结果处理完后，消费成功的消息会从ProcessQueue的msgTreeMap中移除（需要获取到写锁treeMapLock），同时从msgTreeMap中获取最小的Offset来更新对应MessageQueue的偏移量。这个逻辑可以参考下面代码：publicvoidprocessConsumeResult（finalConsumeConcurrentlyStatusstatus，finalConsumeConcurrentlyContextcontext，finalConsumeRequestconsumeRequest）｛intackIndexcontext。getAckIndex（）；switch（status）｛caseCONSUMESUCCESS：if（ackIndexconsumeRequest。getMsgs（）。size（））｛ackIndexconsumeRequest。getMsgs（）。size（）1；｝intokackIndex1；break；。。。｝switch（this。defaultMQPushConsumer。getMessageModel（））｛caseBROADCASTING：。。。break；caseCLUSTERING：ListMessageExtmsgBackFailednewArrayListMessageExt（consumeRequest。getMsgs（）。size（））；for（intiackIndex1；iconsumeRequest。getMsgs（）。size（）；i）｛MessageExtmsgconsumeRequest。getMsgs（）。get（i）；消费失败的，发送回Brokerbooleanresultthis。sendMessageBack（msg，context）；。。。｝break；default：break；｝从msgTreeMap中移除并返回msgTreeMap第一条消息的offsetlongoffsetconsumeRequest。getProcessQueue（）。removeMessage（consumeRequest。getMsgs（））；if（offset0！consumeRequest。getProcessQueue（）。isDropped（））｛this。defaultMQPushConsumerImpl。getOffsetStore（）。updateOffset（consumeRequest。getMessageQueue（），offset，true）；｝｝4消费者限流4。1缓存消息数量
　　如果消费者缓存的消息数量大于RocketMQ配置的阈值（默认1000），就会触发延迟拉取，而消费者缓存的消息数量就来自ProcessQueue，看下面代码：longcachedMessageCountprocessQueue。getMsgCount（）。get（）；if（cachedMessageCountthis。defaultMQPushConsumer。getPullThresholdForQueue（））｛this。executePullRequestLater（pullRequest，PULLTIMEDELAYMILLSWHENFLOWCONTROL）；return；｝4。2缓存的消息大小
　　如果消费者缓存的消息大小大于RocketMQ配置的阈值（默认100M），就会触发延迟拉取，而消费者缓存的消息大小就来自ProcessQueue，看下面代码：longcachedMessageSizeInMiBprocessQueue。getMsgSize（）。get（）（10241024）；if（cachedMessageSizeInMiBthis。defaultMQPushConsumer。getPullThresholdSizeForQueue（））｛this。executePullRequestLater（pullRequest，PULLTIMEDELAYMILLSWHENFLOWCONTROL）；return；｝4。3消息间隔
　　对于普通消息，如果消费偏移量间隔大于配置的阈值（默认2000），就会触发延迟拉取，而消息间隔就来自ProcessQueue，看下面代码：if（！this。consumeOrderly）｛if（processQueue。getMaxSpan（）this。defaultMQPushConsumer。getConsumeConcurrentlyMaxSpan（））｛this。executePullRequestLater（pullRequest，PULLTIMEDELAYMILLSWHENFLOWCONTROL）；return；｝｝4。4获取锁失败
　　对于顺序消息，如果获取锁失败，也会触发延迟拉取，而判断获取锁是否成功，也是在ProcessQueue，看下面代码：if（processQueue。isLocked（））｛。。。｝else｛this。executePullRequestLater（pullRequest，pullTimeDelayMillsWhenException）；｝5总结
　　ProcessQueue是MessageQueue的消费快照，可以协助消费者进行消息拉取、消息消费、更新偏移量、限流。最后，看一下ProcessQueue的数据结构：
　　来源：https：mp。weixin。qq。comszB7dM9xt26c6Z04PvYfOmQ