大家好，我是君哥。
今天來分享 RocketMQ 中一個非常重要又不太好理解的知識點-ProcessQueue。
(資料圖片)
一句話概括，ProcessQueue 就是 MessageQueue 的消費快照?？聪旅孢@張圖：
1 ProcessQueue 構建
RocketMQ 客戶端啟動時，會開啟一個 rebalance 線程，代碼如下:
//MQClientInstance.javapublic void start() throws MQClientException { synchronized (this) { switch (this.serviceState) { case CREATE_JUST: //... // Start rebalance service this.rebalanceService.start(); //... } }}這個線程會不停的做重平衡操作，對 ProcessQueue 進行維護。在重平衡線程類 RebalanceImpl 定義了一個變量 processQueueTable，數(shù)據(jù)結構如下：
可以看到，在 processQueueTable 這個數(shù)據(jù)結構上維護了 MessageQueue 和 ProcessQueue 的映射。
下面看一下維護 processQueueTable 的代碼：
private boolean updateProcessQueueTableInRebalance(final String topic, final Set mqSet, final boolean isOrder) { boolean changed = false; Iterator> it = this.processQueueTable.entrySet().iterator(); while (it.hasNext()) { Entry next = it.next(); MessageQueue mq = next.getKey(); ProcessQueue pq = next.getValue(); if (mq.getTopic().equals(topic)) { if (!mqSet.contains(mq)) { //從processQueueTable上移除 } else if (pq.isPullExpired()) { switch (this.consumeType()) { case CONSUME_ACTIVELY://拉模式 break; case CONSUME_PASSIVELY://推模式 //從processQueueTable上移除 break; default: break; } } } } //創(chuàng)建ProcessQueue并放到processQueueTable List pullRequestList = new ArrayList(); for (MessageQueue mq : mqSet) { if (!this.processQueueTable.containsKey(mq)) { //... ProcessQueue pq = new ProcessQueue(); long nextOffset = -1L; try { nextOffset = this.computePullFromWhereWithException(mq); } catch (Exception e) { log.info("doRebalance, {}, compute offset failed, {}", consumerGroup, mq); continue; } if (nextOffset >= 0) { ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq); if (pre != null) { log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq); } else { //封裝好processQueueTable后再創(chuàng)建一個PullRequest進行消息拉取 log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq); PullRequest pullRequest = new PullRequest(); pullRequest.setConsumerGroup(consumerGroup); pullRequest.setNextOffset(nextOffset); pullRequest.setMessageQueue(mq); pullRequest.setProcessQueue(pq); pullRequestList.add(pullRequest); changed = true; } } else { log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq); } } } this.dispatchPullRequest(pullRequestList); return changed;}
2 拉取消息
上一節(jié)中構建 ProcessQueue 后，會再創(chuàng)建一個 PullRequest，這個 PullRequest 封裝了 MessageQueue 和 ProcessQueue，創(chuàng)建成功后被放到了 PullMessageService 中的 pullRequestQueue 變量：
//PullMessageService.javaprivate final LinkedBlockingQueue pullRequestQueue = new LinkedBlockingQueue();public void executePullRequestImmediately(final PullRequest pullRequest) { try { this.pullRequestQueue.put(pullRequest); } catch (InterruptedException e) { log.error("executePullRequestImmediately pullRequestQueue.put", e); }}
這里以 RocketMQ 的推模式為例，Consumer 拉取到消息后，會進行如下處理：
對拉取到的消息根據(jù) TAG 再次進行過濾；更新 PullRequest 下次拉取的偏移量 nextOffset；把拉取的消息封裝到 ProcessQueue 的 msgTreeMap（
?放到 msgTreeMap 之前首先要獲取到寫鎖 treeMapLock?）；封裝 ConsumeRequest 進行消息消費;封裝消息拉取請求再次進行拉取。
代碼如下：
//DefaultMQPushConsumerImpl.javapublic void onSuccess(PullResult pullResult) { if (pullResult != null) { //1. 對拉取到的消息根據(jù) TAG 再次進行過濾 pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult, subscriptionData); switch (pullResult.getPullStatus()) { case FOUND: //2. 更新 PullRequest 下次拉取的偏移量 nextOffset pullRequest.setNextOffset(pullResult.getNextBeginOffset()); if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) { DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest); } else { //3. 把拉取的消息封裝到 ProcessQueue 的 msgTreeMap boolean dispatchToConsume = processQueue.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 消費消息
在上一節(jié)提到過，拉取到消息后，會把消息封裝成一個 ConsumeRequest,這個線程類會調(diào)用消費者定義的 MessageListener 進行消費處理?？匆幌略创a：
//ConsumeMessageConcurrentlyService.ConsumeRequestpublic void run() { if (this.processQueue.isDropped()) { log.info("the message queue not be able to consume, because it"s dropped. group={} {}", ConsumeMessageConcurrentlyService.this.consumerGroup, this.messageQueue); return; } MessageListenerConcurrently listener = ConsumeMessageConcurrentlyService.this.messageListener; ConsumeConcurrentlyContext context = new ConsumeConcurrentlyContext(messageQueue); ConsumeConcurrentlyStatus status = null; try { status = listener.consumeMessage(Collections.unmodifiableList(msgs), context); }//... if (!processQueue.isDropped()) { ConsumeMessageConcurrentlyService.this.processConsumeResult(status, context, this); }}消息消費成功后，會調(diào)用 processConsumeResult 方法進行結果處理。對于廣播模式，發(fā)送失敗后不會做重試，相當于把消息丟棄，而對于集群模式，消費失敗的消息會發(fā)送到 Broker 端等待消費者重新拉取進行重試。
消費結果處理完后，消費成功的消息會從 ProcessQueue 的 msgTreeMap 中移除（需要獲取到寫鎖 treeMapLock），同時從 msgTreeMap 中獲取最小的 Offset 來更新對應 MessageQueue 的偏移量。這個邏輯可以參考下面代碼：
public void processConsumeResult( final ConsumeConcurrentlyStatus status, final ConsumeConcurrentlyContext context, final ConsumeRequest consumeRequest) { int ackIndex = context.getAckIndex(); switch (status) { case CONSUME_SUCCESS: if (ackIndex >= consumeRequest.getMsgs().size()) { ackIndex = consumeRequest.getMsgs().size() - 1; } int ok = ackIndex + 1; break; //... } switch (this.defaultMQPushConsumer.getMessageModel()) { case BROADCASTING: //... break; case CLUSTERING: List msgBackFailed = new ArrayList(consumeRequest.getMsgs().size()); for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) { MessageExt msg = consumeRequest.getMsgs().get(i); //消費失敗的，發(fā)送回Broker boolean result = this.sendMessageBack(msg, context); //... } break; default: break; } //從msgTreeMap中移除并返回msgTreeMap第一條消息的offset long offset = consumeRequest.getProcessQueue().removeMessage(consumeRequest.getMsgs()); if (offset >= 0 && !consumeRequest.getProcessQueue().isDropped()) { this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), offset, true); }}4 消費者限流4.1 緩存消息數(shù)量
如果消費者緩存的消息數(shù)量大于 RocketMQ 配置的閾值（默認 1000），就會觸發(fā)延遲拉取，而消費者緩存的消息數(shù)量就來自 ProcessQueue，看下面代碼：
long cachedMessageCount = processQueue.getMsgCount().get();if (cachedMessageCount > this.defaultMQPushConsumer.getPullThresholdForQueue()) { this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL); return;}4.2 緩存的消息大小
如果消費者緩存的消息大小大于 RocketMQ 配置的閾值（默認 100M），就會觸發(fā)延遲拉取，而消費者緩存的消息大小就來自 ProcessQueue，看下面代碼：
long cachedMessageSizeInMiB = processQueue.getMsgSize().get() / (1024 * 1024);if (cachedMessageSizeInMiB > this.defaultMQPushConsumer.getPullThresholdSizeForQueue()) { this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL); return;}4.3 消息間隔
對于普通消息，如果消費偏移量間隔大于配置的閾值（默認 2000），就會觸發(fā)延遲拉取，而消息間隔就來自 ProcessQueue，看下面代碼：
if (!this.consumeOrderly) { if (processQueue.getMaxSpan() > this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan()) { this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL); return; }}4.4 獲取鎖失敗
對于順序消息，如果獲取鎖失敗，也會觸發(fā)延遲拉取，而判斷獲取鎖是否成功，也是在 ProcessQueue，看下面代碼：
if (processQueue.isLocked()) { //...} else { this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);}5 總結
ProcessQueue 是 MessageQueue 的消費快照，可以協(xié)助消費者進行消息拉取、消息消費、更新偏移量、限流。最后，看一下 ProcessQueue 的數(shù)據(jù)結構：
關鍵詞：
