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

大家好，我是君哥。

为蓬莱等地区用户提供了全套网页设计制作服务，及蓬莱网站建设行业解决方案。主营业务为网站建设、成都做网站、蓬莱网站设计，以传统方式定制建设网站，并提供域名空间备案等一条龙服务，秉承以专业、用心的态度为用户提供真诚的服务。我们深信只要达到每一位用户的要求，就会得到认可，从而选择与我们长期合作。这样，我们也可以走得更远！

今天来分享 RocketMQ 中一个非常重要又不太好理解的知识点-ProcessQueue。

一句话概括，ProcessQueue 就是 MessageQueue 的消费快照。看下面这张图：

1 ProcessQueue 构建

RocketMQ 客户端启动时，会开启一个 rebalance 线程，代码如下:

//MQClientInstance.java
public void start() throws MQClientException {
 synchronized (this) {
  switch (this.serviceState) {
   case CREATE_JUST:
    //...
    // Start rebalance service
    this.rebalanceService.start();
   //...
  }
 }
}

这个线程会不停的做重平衡操作，对 ProcessQueue 进行维护。在重平衡线程类 RebalanceImpl 定义了一个变量 processQueueTable，数据结构如下：

可以看到，在 processQueueTable 这个数据结构上维护了 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;
    }
   }
  }
 }
    //创建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后再创建一个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 拉取消息

上一节中构建 ProcessQueue 后，会再创建一个 PullRequest，这个 PullRequest 封装了 MessageQueue 和 ProcessQueue，创建成功后被放到了 PullMessageService 中的 pullRequestQueue 变量：

//PullMessageService.java
private 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 拉取到消息后，会进行如下处理：

  1. 对拉取到的消息根据 TAG 再次
    进行过滤；
  2. 更新 PullRequest 下次拉取的偏移量 nextOffset；
  3. 把拉取的消息封装到 ProcessQueue 的 msgTreeMap（
    ​
    放到 msgTreeMap 之前首先要获取到写锁 treeMapLock
    ​
    ）；
  4. 封装 ConsumeRequest 进行消息消费;
  5. 封装消息拉取请求再次进行拉取。

代码如下：

//DefaultMQPushConsumerImpl.java
public void onSuccess(PullResult pullResult) {
 if (pullResult != null) {
     //1. 对拉取到的消息根据 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 消费消息

在上一节提到过，拉取到消息后，会把消息封装成一个 ConsumeRequest,这个线程类会调用消费者定义的 MessageListener 进行消费处理。看一下源代码：

//ConsumeMessageConcurrentlyService.ConsumeRequest
public 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);
 }
}

消息消费成功后，会调用 processConsumeResult 方法进行结果处理。对于广播模式，发送失败后不会做重试，相当于把消息丢弃，而对于集群模式，消费失败的消息会发送到 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);
    //消费失败的，发送回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 缓存消息数量

如果消费者缓存的消息数量大于 RocketMQ 配置的阈值（默认 1000），就会触发延迟拉取，而消费者缓存的消息数量就来自 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），就会触发延迟拉取，而消费者缓存的消息大小就来自 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），就会触发延迟拉取，而消息间隔就来自 ProcessQueue，看下面代码：

if (!this.consumeOrderly) {
 if (processQueue.getMaxSpan() > this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan()) {
  this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
  return;
 }
}

4.4 获取锁失败

对于顺序消息，如果获取锁失败，也会触发延迟拉取，而判断获取锁是否成功，也是在 ProcessQueue，看下面代码：

if (processQueue.isLocked()) {
 //...
} else {
 this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
}

5 总结

ProcessQueue 是 MessageQueue 的消费快照，可以协助消费者进行消息拉取、消息消费、更新偏移量、限流。最后，看一下 ProcessQueue 的数据结构：

当前题目：老弟问我，RocketMQ中的ProcessQueue怎么理解？
本文链接：http://www.mswzjz.cn/qtweb/news36/45786.html

攀枝花网站建设、攀枝花网站运维推广公司-贝锐智能，是专注品牌与效果的网络营销公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 贝锐智能

贝锐智能技术为您推荐以下文章

营销型网站建设知识

同城分类信息

成都商城网站建设　　 　上市企业网站建设方案　　 　名片印刷厂　　 　成都广告招牌制作　　 　成都网站建设　　 　成都劢可为　　 　重庆网站制作　　 　成都h5建站开发　　 　营山月子护理　　 　成都网站设计　　 　软文推广　　 　温江服务器租用　　 　成都网站运维　　 　时嘉做网站　　 　盐亭网站建设　　 　东电技服　　 　温江电信机房　　 　雕琢时光茶叶　　 　香港云主机　　 　大英网站建设