该文所涉及的 RocketMQ 源码版本为 4.9.3。

RocketMQ broker 处理拉取消息请求流程

org.apache.rocketmq.broker.processor.PullMessageProcessor#processRequest(io.netty.channel.ChannelHandlerContext, org.apache.rocketmq.remoting.protocol.RemotingCommand)

第 1 步、校验broker是否可读

  1. if (!PermName.isReadable(this.brokerController.getBrokerConfig().getBrokerPermission())) {
  2.     response.setCode(ResponseCode.NO_PERMISSION);
  3.     response.setRemark(String.format("the broker[%s] pulling message is forbidden", this.brokerController.getBrokerConfig().getBrokerIP1()));
  4.     return response;
  5. }

第 2 步、根据消费组获取订阅信息

  1. SubscriptionGroupConfig subscriptionGroupConfig =
  2.     this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(requestHeader.getConsumerGroup());

第 3 步、校验是否允许消费

  1. if (!subscriptionGroupConfig.isConsumeEnable()) {
  2.     response.setCode(ResponseCode.NO_PERMISSION);
  3.     response.setRemark("subscription group no permission, " + requestHeader.getConsumerGroup());
  4.     return response;
  5. }

第 4 步、根据主题获取对应的配置信息

  1. TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
  2. if (null == topicConfig) {
  3.     log.error("the topic {} not exist, consumer: {}", requestHeader.getTopic(), RemotingHelper.parseChannelRemoteAddr(channel));
  4.     response.setCode(ResponseCode.TOPIC_NOT_EXIST);
  5.     response.setRemark(String.format("topic[%s] not exist, apply first please! %s", requestHeader.getTopic(), FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL)));
  6.     return response;
  7. }

第 5 步、校验主题对应的队列

  1. if (requestHeader.getQueueId() < 0 || requestHeader.getQueueId() >= topicConfig.getReadQueueNums()) {
  2.     String errorInfo = String.format("queueId[%d] is illegal, topic:[%s] topicConfig.readQueueNums:[%d] consumer:[%s]",
  3.         requestHeader.getQueueId(), requestHeader.getTopic(), topicConfig.getReadQueueNums(), channel.remoteAddress());
  4.     log.warn(errorInfo);
  5.     response.setCode(ResponseCode.SYSTEM_ERROR);
  6.     response.setRemark(errorInfo);
  7.     return response;
  8. }

第 6 步、如果配置了消息过滤表达式，根据表达式进行构建consumerFilterData，如果没有，则根据主题构建

  1. consumerFilterData = ConsumerFilterManager.build(
  2.                         requestHeader.getTopic(), requestHeader.getConsumerGroup(), requestHeader.getSubscription(),
  3.                         requestHeader.getExpressionType(), requestHeader.getSubVersion()
  5. consumerFilterData = this.brokerController.getConsumerFilterManager().get(requestHeader.getTopic(),
  6.                     requestHeader.getConsumerGroup());

第 7 步、校验如果不是Tag过滤，是否开启了自定义属性过滤，如果没有开启，不允许操作 只有使用push推送模式的消费者才能用使用SQL92标准的sql语句，pull拉取模式的消费者是不支持这个功能的。

  1. if (!ExpressionType.isTagType(subscriptionData.getExpressionType())
  2.     && !this.brokerController.getBrokerConfig().isEnablePropertyFilter()) {
  3.     response.setCode(ResponseCode.SYSTEM_ERROR);
  4.     response.setRemark("The broker does not support consumer to filter message by " + subscriptionData.getExpressionType());
  5.     return response;
  6. }

第 8 步、根据是否支持重试过滤创建不同的MessageFilter

  1. if (this.brokerController.getBrokerConfig().isFilterSupportRetry()) {
  2.     messageFilter = new ExpressionForRetryMessageFilter(subscriptionData, consumerFilterData,
  3.         this.brokerController.getConsumerFilterManager());
  4. } else {
  5.     messageFilter = new ExpressionMessageFilter(subscriptionData, consumerFilterData,
  6.         this.brokerController.getConsumerFilterManager());
  7. }

第 9 步、根据消费组、主题、队列、偏移量、最大拉取消息数量、消息过滤器查找信息

  1. final GetMessageResult getMessageResult =
  2.     this.brokerController.getMessageStore().getMessage(requestHeader.getConsumerGroup(), requestHeader.getTopic(),
  3.         requestHeader.getQueueId(), requestHeader.getQueueOffset(), requestHeader.getMaxMsgNums(), messageFilter);

第 10 步、消息为空 设置code为系统错误 返回response

  1. response.setCode(ResponseCode.SYSTEM_ERROR);
  2. response.setRemark("store getMessage return null");

第 11 步、提交偏移量

  1. if (storeOffsetEnable) {
  2.     this.brokerController.getConsumerOffsetManager().commitOffset(RemotingHelper.parseChannelRemoteAddr(channel),
  3.         requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getCommitOffset());
  4. }