RocketMQ消息消费（一）

迈不过友情╰ 2023-01-17 14:54 438阅读 0赞

**消息消费以组的模式开展，一个消费组内可以包含多个消费者，每一个消费组可订阅多个主题，消费组之间有集群模式与广播模式两种消费模式。集群模式，主题下的同一条消息只允许被其中一个消费者消费。广播模式，主题下的同一条消息将被集群内的所有消费者消费一次。消息服务器与消费者之间的消息传送也有两种方式：推模式、拉模式。所谓的拉模式，是消费端主动发起拉消息请求，而推模式是消息到达消息服务器后，推送给消息消费者。RocketMQ消息推模式的实现基于拉模式，在拉模式上包装一层，一个拉取任务完成后开始下一个拉取任务。**

**消息队列负载机制遵循一个通用的思想：一个消息队列同一时间只允许被一个消费者消费，一个消费者可以消费多个消息队列。**

**RocketMQ支持局部顺序消息消费，也就是保证同一个消息队列上的消息顺序消费。不支持消息全局顺序消费，如果要实现某一主题的全局顺序消息消费，可以将该主题的队列数设置为1，牺牲高可用性。**

**RocketMQ支持两种消息过滤模式：表达式（TAG、SQL92）与类过滤模式。**

**消息拉模式，主要是由客户端手动调用消息拉取API，而消息推模式是消息服务器主动将消息推送到消息消费端。**

# 消费者启动流程 #

**DefaultMQPushConsumerImpl的start方法**

public synchronized void start() throws MQClientException {
            switch (this.serviceState) {
                case CREATE_JUST:
                    log.info("the consumer [{}] start beginning. messageModel={}, isUnitMode={}", this.defaultMQPushConsumer.getConsumerGroup(),
                        this.defaultMQPushConsumer.getMessageModel(), this.defaultMQPushConsumer.isUnitMode());
                    this.serviceState = ServiceState.START_FAILED;
    
                    this.checkConfig();
    
                    this.copySubscription();
    
                    if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {
                        this.defaultMQPushConsumer.changeInstanceNameToPID();
                    }
    
                    this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
    
                    this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
                    this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
                    this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
                    this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);
    
                    this.pullAPIWrapper = new PullAPIWrapper(
                        mQClientFactory,
                        this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
                    this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);
    
                    if (this.defaultMQPushConsumer.getOffsetStore() != null) {
                        this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
                    } else {
                        switch (this.defaultMQPushConsumer.getMessageModel()) {
                            case BROADCASTING:
                                this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                                break;
                            case CLUSTERING:
                                this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                                break;
                            default:
                                break;
                        }
                        this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
                    }
                    this.offsetStore.load();
    
                    if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
                        this.consumeOrderly = true;
                        this.consumeMessageService =
                            new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
                    } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
                        this.consumeOrderly = false;
                        this.consumeMessageService =
                            new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
                    }
    
                    this.consumeMessageService.start();
    
                    boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
                    if (!registerOK) {
                        this.serviceState = ServiceState.CREATE_JUST;
                        this.consumeMessageService.shutdown(defaultMQPushConsumer.getAwaitTerminationMillisWhenShutdown());
                        throw new MQClientException("The consumer group[" + this.defaultMQPushConsumer.getConsumerGroup()
                            + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
                            null);
                    }
    
                    mQClientFactory.start();
                    log.info("the consumer [{}] start OK.", this.defaultMQPushConsumer.getConsumerGroup());
                    this.serviceState = ServiceState.RUNNING;
                    break;
                case RUNNING:
                case START_FAILED:
                case SHUTDOWN_ALREADY:
                    throw new MQClientException("The PushConsumer service state not OK, maybe started once, "
                        + this.serviceState
                        + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
                        null);
                default:
                    break;
            }
    
            this.updateTopicSubscribeInfoWhenSubscriptionChanged();
            this.mQClientFactory.checkClientInBroker();
            this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
            this.mQClientFactory.rebalanceImmediately();
        }

## 1.步骤一 ##

**构建主题订阅信息SubscriptionData并加入到RebalanceImpl的订阅消息中。**

**订阅关系来源主要有两个。**

*  **通过调用DefaultMQPushConsumerImpl\#subscribe（String topic, String sub Expression）方法。**
 *  **订阅重试主题消息。从这里可以看出，RocketMQ消息重试是以消费组为单位，而不是主题，消息重试主题名为%RETRY%+消费组名。消费者在启动的时候会自动订阅该主题，参与该主题的消息队列负载。**

private void copySubscription() throws MQClientException {
            try {
                Map<String, String> sub = this.defaultMQPushConsumer.getSubscription();
                if (sub != null) {
                    for (final Map.Entry<String, String> entry : sub.entrySet()) {
                        final String topic = entry.getKey();
                        final String subString = entry.getValue();
                        SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(),
                            topic, subString);
                        this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
                    }
                }
    
                if (null == this.messageListenerInner) {
                    this.messageListenerInner = this.defaultMQPushConsumer.getMessageListener();
                }
    
                switch (this.defaultMQPushConsumer.getMessageModel()) {
                    case BROADCASTING:
                        break;
                    case CLUSTERING:
                        final String retryTopic = MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup());
                        SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(),
                            retryTopic, SubscriptionData.SUB_ALL);
                        this.rebalanceImpl.getSubscriptionInner().put(retryTopic, subscriptionData);
                        break;
                    default:
                        break;
                }
            } catch (Exception e) {
                throw new MQClientException("subscription exception", e);
            }
        }

## **2.**步骤二 ##

**初始化MQClientInstance、RebalanceImple（消息重新负载实现类）等。**

if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {
                        this.defaultMQPushConsumer.changeInstanceNameToPID();
                    }
    
                    this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
    
                    this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
                    this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
                    this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
                    this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);
    
                    this.pullAPIWrapper = new PullAPIWrapper(
                        mQClientFactory,
                        this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
                    this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);

## 3.步骤三 ##

**初始化消息进度。如果消息消费是集群模式，那么消息进度保存在Broker上；如果是广播模式，那么消息消费进度存储在消费端。**

if (this.defaultMQPushConsumer.getOffsetStore() != null) {
                        this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
                    } else {
                        switch (this.defaultMQPushConsumer.getMessageModel()) {
                            case BROADCASTING:
                                this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                                break;
                            case CLUSTERING:
                                this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                                break;
                            default:
                                break;
                        }
                        this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
                    }
                    this.offsetStore.load();

## 4.步骤四 ##

**根据是否是顺序消费，创建消费端消费线程服务。ConsumeMessageService主要负责消息消费，内部维护一个线程池。**

if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
                        this.consumeOrderly = true;
                        this.consumeMessageService =
                            new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
                    } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
                        this.consumeOrderly = false;
                        this.consumeMessageService =
                            new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
                    }
    
                    this.consumeMessageService.start();

## 5.步骤五 ##

**向MQClientInstance注册消费者，并启动MQClientInstance，在一个JVM中的所有消费者、生产者持有同一个MQClientInstance, MQClientInstance只会启动一次。**

boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
                    if (!registerOK) {
                        this.serviceState = ServiceState.CREATE_JUST;
                        this.consumeMessageService.shutdown(defaultMQPushConsumer.getAwaitTerminationMillisWhenShutdown());
                        throw new MQClientException("The consumer group[" + this.defaultMQPushConsumer.getConsumerGroup()
                            + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
                            null);
                    }
    
                    mQClientFactory.start();

# 消息拉取 #

**消息消费有两种模式：广播模式与集群模式，广播模式比较简单，每一个消费者需要去拉取订阅主题下所有消费队列的消息。在集群模式下，同一个消费组内有多个消息消费者，同一个主题存在多个消费队列。**

## **PullMessageService实现机制** ##

**PullMessageService继承的是ServiceThread，从名称来看，它是服务线程，通过run方法启动。**

@Override
        public void run() {
            log.info(this.getServiceName() + " service started");
    
            while (!this.isStopped()) {
                try {
                    PullRequest pullRequest = this.pullRequestQueue.take();
                    this.pullMessage(pullRequest);
                } catch (InterruptedException ignored) {
                } catch (Exception e) {
                    log.error("Pull Message Service Run Method exception", e);
                }
            }
    
            log.info(this.getServiceName() + " service end");
        }

1.  **while（! this.isStopped（））这是一种通用的设计技巧，stopped声明为volatile，每执行一次业务逻辑检测一下其运行状态，可以通过其他线程将stopped设置为true从而停止该线程。**
2.  **从pullRequestQueue中获取一个PullRequest消息拉取任务，如果pullRequest Queue为空，则线程将阻塞，直到有拉取任务被放入。**
3.  **调用pullMessage方法进行消息拉取。**

**PullMessageService提供延迟添加与立即添加2种方式将PullRequest放入到pullRequestQueue中。**

**主要有两个地方会调用，一个是在RocketMQ根据PullRequest拉取任务执行完一次消息拉取任务后，又将PullRequest对象放入到pullRequestQueue，第二个是在RebalancceImpl中创建。**

## **ProcessQueue实现机制** ##

**ProcessQueue是MessageQueue在消费端的重现、快照。PullMessageService从消息服务器默认每次拉取32条消息，按消息的队列偏移量顺序存放在ProcessQueue中，PullMessageService然后将消息提交到消费者消费线程池，消息成功消费后从ProcessQueue中移除。**

## **消息拉取基本流程** ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70][]

**消息拉取分为3个主要步骤。**

1.  **消息拉取客户端消息拉取请求封装。**
2.  **消息服务器查找并返回消息。**
3.  **消息拉取客户端处理返回的消息。**

### 客户端封装消息拉取请求 ###

**消息拉取入口：DefaultMQPushConsumerImpl\#pullMessage。**

### **1.步骤一** ###

**从PullRequest中获取ProcessQueue，如果处理队列当前状态未被丢弃，则更新ProcessQueue的lastPullTimestamp为当前时间戳；如果当前消费者被挂起，则将拉取任务延迟1s再次放入到PullMessageService的拉取任务队列中，结束本次消息拉取。**

final ProcessQueue processQueue = pullRequest.getProcessQueue();
            if (processQueue.isDropped()) {
                log.info("the pull request[{}] is dropped.", pullRequest.toString());
                return;
            }
    
            pullRequest.getProcessQueue().setLastPullTimestamp(System.currentTimeMillis());
    
            try {
                this.makeSureStateOK();
            } catch (MQClientException e) {
                log.warn("pullMessage exception, consumer state not ok", e);
                this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
                return;
            }
    
            if (this.isPause()) {
                log.warn("consumer was paused, execute pull request later. instanceName={}, group={}", this.defaultMQPushConsumer.getInstanceName(), this.defaultMQPushConsumer.getConsumerGroup());
                this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_SUSPEND);
                return;
            }

### 2**.步骤二** ###

**进行消息拉取流控。从消息消费数量与消费间隔两个维度进行控制。**

1.  **消息处理总数，如果ProcessQueue当前处理的消息条数超过了pullThresholdFor-Queue=1000将触发流控，放弃本次拉取任务，并且该队列的下一次拉取任务将在50毫秒后才加入到拉取任务队列中，每触发1000次流控后输出提示语：theconsumer message buffer is full, so do flow control, minOffset=\{队列最小偏移量\}, maxOffset=\{队列最大偏移量\}, size=\{消息总条数\}, pullRequest=\{拉取任务\}, flowControlTimes=\{流控触发次数\}。**
2.  **ProcessQueue中队列最大偏移量与最小偏离量的间距，不能超过consumeConcurrently-MaxSpan，否则触发流控，每触发1000次输出提示语：thequeue's messages, span too long, so do flow control, minOffset=\{队列最小偏移量\}, maxOffset=\{队列最大偏移量\}, maxSpan=\{间隔\}, pullRequest=\{拉取任务信息\}, flowControlTimes=\{流控触发次数\}。这里主要的考量是担心一条消息堵塞，消息进度无法向前推进，可能造成大量消息重复消费。**

long cachedMessageCount = processQueue.getMsgCount().get();
            long cachedMessageSizeInMiB = processQueue.getMsgSize().get() / (1024 * 1024);
    
            if (cachedMessageCount > this.defaultMQPushConsumer.getPullThresholdForQueue()) {
                this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
                if ((queueFlowControlTimes++ % 1000) == 0) {
                    log.warn(
                        "the cached message count exceeds the threshold {}, so do flow control, minOffset={}, maxOffset={}, count={}, size={} MiB, pullRequest={}, flowControlTimes={}",
                        this.defaultMQPushConsumer.getPullThresholdForQueue(), processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), cachedMessageCount, cachedMessageSizeInMiB, pullRequest, queueFlowControlTimes);
                }
                return;
            }
    
            if (cachedMessageSizeInMiB > this.defaultMQPushConsumer.getPullThresholdSizeForQueue()) {
                this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
                if ((queueFlowControlTimes++ % 1000) == 0) {
                    log.warn(
                        "the cached message size exceeds the threshold {} MiB, so do flow control, minOffset={}, maxOffset={}, count={}, size={} MiB, pullRequest={}, flowControlTimes={}",
                        this.defaultMQPushConsumer.getPullThresholdSizeForQueue(), processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), cachedMessageCount, cachedMessageSizeInMiB, pullRequest, queueFlowControlTimes);
                }
                return;
            }

### 3.步骤三 ###

**拉取该主题订阅信息，如果为空，结束本次消息拉取，关于该队列的下一次拉取任务延迟3s。**

final SubscriptionData subscriptionData = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());
            if (null == subscriptionData) {
                this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
                log.warn("find the consumer's subscription failed, {}", pullRequest);
                return;
            }

### 4.步骤四 ###

**构建消息拉取系统标记。**

String subExpression = null;
            boolean classFilter = false;
            SubscriptionData sd = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());
            if (sd != null) {
                if (this.defaultMQPushConsumer.isPostSubscriptionWhenPull() && !sd.isClassFilterMode()) {
                    subExpression = sd.getSubString();
                }
    
                classFilter = sd.isClassFilterMode();
            }
    
            int sysFlag = PullSysFlag.buildSysFlag(
                commitOffsetEnable, // commitOffset
                true, // suspend
                subExpression != null, // subscription
                classFilter // class filter
            );

### 5.步骤五 ###

**调用PullAPIWrapper.pullKernelImpl方法后与服务端交互。**

this.pullAPIWrapper.pullKernelImpl(
                    pullRequest.getMessageQueue(),
                    subExpression,
                    subscriptionData.getExpressionType(),
                    subscriptionData.getSubVersion(),
                    pullRequest.getNextOffset(),
                    this.defaultMQPushConsumer.getPullBatchSize(),
                    sysFlag,
                    commitOffsetValue,
                    BROKER_SUSPEND_MAX_TIME_MILLIS,
                    CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND,
                    CommunicationMode.ASYNC,
                    pullCallback
                );

### **6.步骤六** ###

**根据brokerName、BrokerId从MQClientInstance中获取Broker地址，在整个RocketMQ Broker的部署结构中，相同名称的Broker构成主从结构，其BrokerId会不一样，在每次拉取消息后，会给出一个建议，下次拉取从主节点还是从节点拉取。**

FindBrokerResult findBrokerResult =
                this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(),
                    this.recalculatePullFromWhichNode(mq), false);
            if (null == findBrokerResult) {
                this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
                findBrokerResult =
                    this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(),
                        this.recalculatePullFromWhichNode(mq), false);
            }

### **7.步骤七** ###

**如果消息过滤模式为类过滤，则需要根据主题名称、broker地址找到注册在Broker上的FilterServer地址，从FilterServer上拉取消息，否则从Broker上拉取消息。**

String brokerAddr = findBrokerResult.getBrokerAddr();
                if (PullSysFlag.hasClassFilterFlag(sysFlagInner)) {
                    brokerAddr = computePullFromWhichFilterServer(mq.getTopic(), brokerAddr);
                }

## 消息服务端Broker组装消息  ##

**根据消息拉取命令Code:RequestCode.PULL\_MESSAGE，很容易找到Brokder端处理消息拉取的入口：org.apache.rocketmq.broker.processor.PullMessageProcessor\#processRequest。**

### **1.步骤一** ###

**根据订阅信息，构建消息过滤器。**

RemotingCommand response = RemotingCommand.createResponseCommand(PullMessageResponseHeader.class);
            final PullMessageResponseHeader responseHeader = (PullMessageResponseHeader) response.readCustomHeader();
            final PullMessageRequestHeader requestHeader =
                (PullMessageRequestHeader) request.decodeCommandCustomHeader(PullMessageRequestHeader.class);
    
            response.setOpaque(request.getOpaque());
    
            log.debug("receive PullMessage request command, {}", request);
    
            if (!PermName.isReadable(this.brokerController.getBrokerConfig().getBrokerPermission())) {
                response.setCode(ResponseCode.NO_PERMISSION);
                response.setRemark(String.format("the broker[%s] pulling message is forbidden", this.brokerController.getBrokerConfig().getBrokerIP1()));
                return response;
            }
    
            SubscriptionGroupConfig subscriptionGroupConfig =
                this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(requestHeader.getConsumerGroup());
            if (null == subscriptionGroupConfig) {
                response.setCode(ResponseCode.SUBSCRIPTION_GROUP_NOT_EXIST);
                response.setRemark(String.format("subscription group [%s] does not exist, %s", requestHeader.getConsumerGroup(), FAQUrl.suggestTodo(FAQUrl.SUBSCRIPTION_GROUP_NOT_EXIST)));
                return response;
            }
    
            if (!subscriptionGroupConfig.isConsumeEnable()) {
                response.setCode(ResponseCode.NO_PERMISSION);
                response.setRemark("subscription group no permission, " + requestHeader.getConsumerGroup());
                return response;
            }
    
            final boolean hasSuspendFlag = PullSysFlag.hasSuspendFlag(requestHeader.getSysFlag());
            final boolean hasCommitOffsetFlag = PullSysFlag.hasCommitOffsetFlag(requestHeader.getSysFlag());
            final boolean hasSubscriptionFlag = PullSysFlag.hasSubscriptionFlag(requestHeader.getSysFlag());
    
            final long suspendTimeoutMillisLong = hasSuspendFlag ? requestHeader.getSuspendTimeoutMillis() : 0;
    
            TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
            if (null == topicConfig) {
                log.error("the topic {} not exist, consumer: {}", requestHeader.getTopic(), RemotingHelper.parseChannelRemoteAddr(channel));
                response.setCode(ResponseCode.TOPIC_NOT_EXIST);
                response.setRemark(String.format("topic[%s] not exist, apply first please! %s", requestHeader.getTopic(), FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL)));
                return response;
            }
    
            if (!PermName.isReadable(topicConfig.getPerm())) {
                response.setCode(ResponseCode.NO_PERMISSION);
                response.setRemark("the topic[" + requestHeader.getTopic() + "] pulling message is forbidden");
                return response;
            }
    
            if (requestHeader.getQueueId() < 0 || requestHeader.getQueueId() >= topicConfig.getReadQueueNums()) {
                String errorInfo = String.format("queueId[%d] is illegal, topic:[%s] topicConfig.readQueueNums:[%d] consumer:[%s]",
                    requestHeader.getQueueId(), requestHeader.getTopic(), topicConfig.getReadQueueNums(), channel.remoteAddress());
                log.warn(errorInfo);
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark(errorInfo);
                return response;
            }
    
            SubscriptionData subscriptionData = null;
            ConsumerFilterData consumerFilterData = null;
            if (hasSubscriptionFlag) {
                try {
                    subscriptionData = FilterAPI.build(
                        requestHeader.getTopic(), requestHeader.getSubscription(), requestHeader.getExpressionType()
                    );
                    if (!ExpressionType.isTagType(subscriptionData.getExpressionType())) {
                        consumerFilterData = ConsumerFilterManager.build(
                            requestHeader.getTopic(), requestHeader.getConsumerGroup(), requestHeader.getSubscription(),
                            requestHeader.getExpressionType(), requestHeader.getSubVersion()
                        );
                        assert consumerFilterData != null;
                    }
                } catch (Exception e) {
                    log.warn("Parse the consumer's subscription[{}] failed, group: {}", requestHeader.getSubscription(),
                        requestHeader.getConsumerGroup());
                    response.setCode(ResponseCode.SUBSCRIPTION_PARSE_FAILED);
                    response.setRemark("parse the consumer's subscription failed");
                    return response;
                }
            } else {
                ConsumerGroupInfo consumerGroupInfo =
                    this.brokerController.getConsumerManager().getConsumerGroupInfo(requestHeader.getConsumerGroup());
                if (null == consumerGroupInfo) {
                    log.warn("the consumer's group info not exist, group: {}", requestHeader.getConsumerGroup());
                    response.setCode(ResponseCode.SUBSCRIPTION_NOT_EXIST);
                    response.setRemark("the consumer's group info not exist" + FAQUrl.suggestTodo(FAQUrl.SAME_GROUP_DIFFERENT_TOPIC));
                    return response;
                }
    
                if (!subscriptionGroupConfig.isConsumeBroadcastEnable()
                    && consumerGroupInfo.getMessageModel() == MessageModel.BROADCASTING) {
                    response.setCode(ResponseCode.NO_PERMISSION);
                    response.setRemark("the consumer group[" + requestHeader.getConsumerGroup() + "] can not consume by broadcast way");
                    return response;
                }
    
                subscriptionData = consumerGroupInfo.findSubscriptionData(requestHeader.getTopic());
                if (null == subscriptionData) {
                    log.warn("the consumer's subscription not exist, group: {}, topic:{}", requestHeader.getConsumerGroup(), requestHeader.getTopic());
                    response.setCode(ResponseCode.SUBSCRIPTION_NOT_EXIST);
                    response.setRemark("the consumer's subscription not exist" + FAQUrl.suggestTodo(FAQUrl.SAME_GROUP_DIFFERENT_TOPIC));
                    return response;
                }
    
                if (subscriptionData.getSubVersion() < requestHeader.getSubVersion()) {
                    log.warn("The broker's subscription is not latest, group: {} {}", requestHeader.getConsumerGroup(),
                        subscriptionData.getSubString());
                    response.setCode(ResponseCode.SUBSCRIPTION_NOT_LATEST);
                    response.setRemark("the consumer's subscription not latest");
                    return response;
                }
                if (!ExpressionType.isTagType(subscriptionData.getExpressionType())) {
                    consumerFilterData = this.brokerController.getConsumerFilterManager().get(requestHeader.getTopic(),
                        requestHeader.getConsumerGroup());
                    if (consumerFilterData == null) {
                        response.setCode(ResponseCode.FILTER_DATA_NOT_EXIST);
                        response.setRemark("The broker's consumer filter data is not exist!Your expression may be wrong!");
                        return response;
                    }
                    if (consumerFilterData.getClientVersion() < requestHeader.getSubVersion()) {
                        log.warn("The broker's consumer filter data is not latest, group: {}, topic: {}, serverV: {}, clientV: {}",
                            requestHeader.getConsumerGroup(), requestHeader.getTopic(), consumerFilterData.getClientVersion(), requestHeader.getSubVersion());
                        response.setCode(ResponseCode.FILTER_DATA_NOT_LATEST);
                        response.setRemark("the consumer's consumer filter data not latest");
                        return response;
                    }
                }
            }
    
            if (!ExpressionType.isTagType(subscriptionData.getExpressionType())
                && !this.brokerController.getBrokerConfig().isEnablePropertyFilter()) {
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("The broker does not support consumer to filter message by " + subscriptionData.getExpressionType());
                return response;
            }
    
            MessageFilter messageFilter;
            if (this.brokerController.getBrokerConfig().isFilterSupportRetry()) {
                messageFilter = new ExpressionForRetryMessageFilter(subscriptionData, consumerFilterData,
                    this.brokerController.getConsumerFilterManager());
            } else {
                messageFilter = new ExpressionMessageFilter(subscriptionData, consumerFilterData,
                    this.brokerController.getConsumerFilterManager());
            }

### 2.步骤二 ###

**调用MessageStore.getMessage查找消息。**

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

### 3.步骤三 ###

**根据主题名称与队列编号获取消息消费队列。**

GetMessageStatus status = GetMessageStatus.NO_MESSAGE_IN_QUEUE;
            long nextBeginOffset = offset;
            long minOffset = 0;
            long maxOffset = 0;
    
            GetMessageResult getResult = new GetMessageResult();
    
            final long maxOffsetPy = this.commitLog.getMaxOffset();

### 4.步骤四 ###

**消息偏移量异常情况校对下一次拉取偏移量。**

1.  **maxOffset = 0，表示当前消费队列中没有消息，拉取结果：NO\_MESSAGE\_IN\_QUEUE。如果当前Broker为主节点或offsetCheckInSlave为false，下次拉取偏移量依然为offset。如果当前Broker为从节点，offsetCheckInSlave为true，设置下次拉取偏移量为0。**
2.  **offset < minOffset，表示待拉取消息偏移量小于队列的起始偏移量，拉取结果为：OFFSET\_TOO\_SMALL。如果当前Broker为主节点或offsetCheckInSlave为false，下次拉取偏移量依然为offset。如果当前Broker为从节点并且offsetCheckInSlave为true，下次拉取偏移量设置为minOffset。**
3.  **offset == maxOffset，如果待拉取偏移量等于队列最大偏移量，拉取结果：OFFSET\_OVERFLOW\_ONE。下次拉取偏移量依然为offset。**
4.  **offset > maxOffset，表示偏移量越界，拉取结果：OFFSET\_OVERFLOW\_BADLY。根据是否是主节点、从节点，同样校对下次拉取偏移量。**

ConsumeQueue consumeQueue = findConsumeQueue(topic, queueId);
            if (consumeQueue != null) {
                minOffset = consumeQueue.getMinOffsetInQueue();
                maxOffset = consumeQueue.getMaxOffsetInQueue();
    
                if (maxOffset == 0) {
                    status = GetMessageStatus.NO_MESSAGE_IN_QUEUE;
                    nextBeginOffset = nextOffsetCorrection(offset, 0);
                } else if (offset < minOffset) {
                    status = GetMessageStatus.OFFSET_TOO_SMALL;
                    nextBeginOffset = nextOffsetCorrection(offset, minOffset);
                } else if (offset == maxOffset) {
                    status = GetMessageStatus.OFFSET_OVERFLOW_ONE;
                    nextBeginOffset = nextOffsetCorrection(offset, offset);
                } else if (offset > maxOffset) {
                    status = GetMessageStatus.OFFSET_OVERFLOW_BADLY;
                    if (0 == minOffset) {
                        nextBeginOffset = nextOffsetCorrection(offset, minOffset);
                    } else {
                        nextBeginOffset = nextOffsetCorrection(offset, maxOffset);
                    }

### 5.步骤五 ###

**如果待拉取偏移量大于minOffset并且小于maxOffset，从当前offset处尝试拉取32条消息，根据消息队列偏移量（ConsumeQueue）从commitlog文件中查找消息。**

### **6.步骤六** ###

**根据PullResult填充responseHeader的nextBegionOffset、minOffset、maxOffset。**

response.setRemark(getMessageResult.getStatus().name());
                responseHeader.setNextBeginOffset(getMessageResult.getNextBeginOffset());
                responseHeader.setMinOffset(getMessageResult.getMinOffset());
                responseHeader.setMaxOffset(getMessageResult.getMaxOffset());

### 7.步骤七 ###

**根据主从同步延迟，如果从节点数据包含下一次拉取的偏移量，设置下一次拉取任务的brokerId。**

if (getMessageResult.isSuggestPullingFromSlave()) {
                    responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getWhichBrokerWhenConsumeSlowly());
                } else {
                    responseHeader.setSuggestWhichBrokerId(MixAll.MASTER_ID);
                }

### 8.步骤八 ###

**根据GetMessageResult编码转换成关系。**

switch (response.getCode()) {
                        case ResponseCode.SUCCESS:
                            int commercialBaseCount = brokerController.getBrokerConfig().getCommercialBaseCount();
                            int incValue = getMessageResult.getMsgCount4Commercial() * commercialBaseCount;
    
                            context.setCommercialRcvStats(BrokerStatsManager.StatsType.RCV_SUCCESS);
                            context.setCommercialRcvTimes(incValue);
                            context.setCommercialRcvSize(getMessageResult.getBufferTotalSize());
                            context.setCommercialOwner(owner);
    
                            break;
                        case ResponseCode.PULL_NOT_FOUND:
                            if (!brokerAllowSuspend) {
    
                                context.setCommercialRcvStats(BrokerStatsManager.StatsType.RCV_EPOLLS);
                                context.setCommercialRcvTimes(1);
                                context.setCommercialOwner(owner);
    
                            }
                            break;
                        case ResponseCode.PULL_RETRY_IMMEDIATELY:
                        case ResponseCode.PULL_OFFSET_MOVED:
                            context.setCommercialRcvStats(BrokerStatsManager.StatsType.RCV_EPOLLS);
                            context.setCommercialRcvTimes(1);
                            context.setCommercialOwner(owner);
                            break;
                        default:
                            assert false;
                            break;
                    }

### 9.步骤九 ###

**如果commitlog标记可用并且当前节点为主节点，则更新消息消费进度。**

**服务端消息拉取处理完毕，将返回结果到拉取消息调用方。在调用方，需要重点关注PULL\_RETRY\_IMMEDIATELY、PULL\_OFFSET\_MOVED、PULL\_NOT\_FOUND等情况下如何校正拉取偏移量。**

boolean storeOffsetEnable = brokerAllowSuspend;
            storeOffsetEnable = storeOffsetEnable && hasCommitOffsetFlag;
            storeOffsetEnable = storeOffsetEnable
                && this.brokerController.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE;
            if (storeOffsetEnable) {
                this.brokerController.getConsumerOffsetManager().commitOffset(RemotingHelper.parseChannelRemoteAddr(channel),
                    requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getCommitOffset());
            }

## 消息拉取客户端处理消息 ##

**NettyRemotingClient在收到服务端响应结构后会回调PullCallback的onSuccess或onException, PullCallBack对象在DefaultMQPushConsumerImpl\#pullMessage中创建。**

### 1.步骤一 ###

**根据响应结果解码成PullResultExt对象，此时只是从网络中读取消息列表到byte\[\] messageBinary属性。**

PullStatus pullStatus = PullStatus.NO_NEW_MSG;
            switch (response.getCode()) {
                case ResponseCode.SUCCESS:
                    pullStatus = PullStatus.FOUND;
                    break;
                case ResponseCode.PULL_NOT_FOUND:
                    pullStatus = PullStatus.NO_NEW_MSG;
                    break;
                case ResponseCode.PULL_RETRY_IMMEDIATELY:
                    pullStatus = PullStatus.NO_MATCHED_MSG;
                    break;
                case ResponseCode.PULL_OFFSET_MOVED:
                    pullStatus = PullStatus.OFFSET_ILLEGAL;
                    break;
    
                default:
                    throw new MQBrokerException(response.getCode(), response.getRemark(), addr);
            }
    
            PullMessageResponseHeader responseHeader =
                (PullMessageResponseHeader) response.decodeCommandCustomHeader(PullMessageResponseHeader.class);
    
            return new PullResultExt(pullStatus, responseHeader.getNextBeginOffset(), responseHeader.getMinOffset(),
                responseHeader.getMaxOffset(), null, responseHeader.getSuggestWhichBrokerId(), response.getBody());

### 2.步骤二 ###

**调用pullAPIWrapper的processPullResult将消息字节数组解码成消息列表填充msgFoundList，并对消息进行消息过滤（TAG）模式。**

**DefaultMQPushConsumerImpl$PullCallBack\#onSuccess**

pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
                            subscriptionData);

### 3.步骤三 ###

**更新PullRequest的下一次拉取偏移量，如果msgFoundList为空，则立即将PullReqeuest放入到PullMessageService的pullRequestQueue，以便PullMessageSerivce能及时唤醒并再次执行消息拉取。**

long prevRequestOffset = pullRequest.getNextOffset();
                                pullRequest.setNextOffset(pullResult.getNextBeginOffset());
                                long pullRT = System.currentTimeMillis() - beginTimestamp;
                                DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullRT(pullRequest.getConsumerGroup(),
                                    pullRequest.getMessageQueue().getTopic(), pullRT);

### 4.步骤四 ###

**首先将拉取到的消息存入ProcessQueue，然后将拉取到的消息提交到ConsumeMessageService中供消费者消费，该方法是一个异步方法，也就是PullCallBack将消息提交到ConsumeMessageService中就会立即返回，至于这些消息如何消费，PullCallBack不关注。**

boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
                                    DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
                                        pullResult.getMsgFoundList(),
                                        processQueue,
                                        pullRequest.getMessageQueue(),
                                        dispatchToConsume);

### 5.步骤五 ###

**将消息提交给消费者线程之后PullCallBack将立即返回，可以说本次消息拉取顺利完成，然后根据pullInterval参数，如果pullInterval>0，则等待pullInterval毫秒后将PullRequest对象放入到PullMessageService的pullRequestQueue中，该消息队列的下次拉取即将被激活，达到持续消息拉取，实现准实时拉取消息的效果。**

if (DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval() > 0) {
                                        DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest,
                                            DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval());
                                    } else {
                                        DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                                    }

### 消息拉取长轮询机制分析 ###

**RocketMQ并没有真正实现推模式，而是消费者主动向消息服务器拉取消息，RocketMQ推模式是循环向消息服务端发送消息拉取请求，如果消息消费者向RocketMQ发送消息拉取时，消息并未到达消费队列，如果不启用长轮询机制，则会在服务端等待shortPollingTimeMills时间后（挂起）再去判断消息是否已到达消息队列，如果消息未到达则提示消息拉取客户端PULL\_NOT\_FOUND（消息不存在），如果开启长轮询模式，RocketMQ一方面会每5s轮询检查一次消息是否可达，同时一有新消息到达后立马通知挂起线程再次验证新消息是否是自己感兴趣的消息，如果是则从commitlog文件提取消息返回给消息拉取客户端，否则直到挂起超时，超时时间由消息拉取方在消息拉取时封装在请求参数中，PUSH模式默认为15s, PULL模式通过DefaultMQPullConsumer\#setBrokerSuspendMaxTimeMillis设置。**

**RocketMQ通过在Broker端配置longPollingEnable为true来开启长轮询模式。消息拉取时服务端从Commitlog未找到消息时的处理逻辑如下。**

**PullMessageProcessor\#processRequest**

**RocketMQ轮询机制由两个线程共同来完成。**

1.  **PullRequestHoldService：每隔5s重试一次。**
2.  **DefaultMessageStore\#ReputMessageService，每处理一次重新拉取，Thread.sleep（1），继续下一次检查。**

private RemotingCommand processRequest(final Channel channel, RemotingCommand request, boolean brokerAllowSuspend)
            throws RemotingCommandException {
    
    ……
                    case ResponseCode.PULL_NOT_FOUND:
    
                        if (brokerAllowSuspend && hasSuspendFlag) {
                            long pollingTimeMills = suspendTimeoutMillisLong;
                            if (!this.brokerController.getBrokerConfig().isLongPollingEnable()) {
                                pollingTimeMills = this.brokerController.getBrokerConfig().getShortPollingTimeMills();
                            }
    
                            String topic = requestHeader.getTopic();
                            long offset = requestHeader.getQueueOffset();
                            int queueId = requestHeader.getQueueId();
                            PullRequest pullRequest = new PullRequest(request, channel, pollingTimeMills,
                                this.brokerController.getMessageStore().now(), offset, subscriptionData, messageFilter);
                            this.brokerController.getPullRequestHoldService().suspendPullRequest(topic, queueId, pullRequest);
                            response = null;
                            break;
                        }
    ……
        }

## PullRequestHoldService线程详解 ##

@Override
        public void run() {
            log.info("{} service started", this.getServiceName());
            while (!this.isStopped()) {
                try {
                    if (this.brokerController.getBrokerConfig().isLongPollingEnable()) {
                        this.waitForRunning(5 * 1000);
                    } else {
                        this.waitForRunning(this.brokerController.getBrokerConfig().getShortPollingTimeMills());
                    }
    
                    long beginLockTimestamp = this.systemClock.now();
                    this.checkHoldRequest();
                    long costTime = this.systemClock.now() - beginLockTimestamp;
                    if (costTime > 5 * 1000) {
                        log.info("[NOTIFYME] check hold request cost {} ms.", costTime);
                    }
                } catch (Throwable e) {
                    log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }
    
            log.info("{} service end", this.getServiceName());
        }

**如果开启长轮询，每5s尝试一次，判断新消息是否到达。如果未开启长轮询，则默认等待1s再次尝试，可以通过BrokerConfig\#shortPollingTimeMills改变等待时间。PullRequestHoldService的核心逻辑见PullRequestHoldService\#checkHoldRequest。**

private void checkHoldRequest() {
            for (String key : this.pullRequestTable.keySet()) {
                String[] kArray = key.split(TOPIC_QUEUEID_SEPARATOR);
                if (2 == kArray.length) {
                    String topic = kArray[0];
                    int queueId = Integer.parseInt(kArray[1]);
                    final long offset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId);
                    try {
                        this.notifyMessageArriving(topic, queueId, offset);
                    } catch (Throwable e) {
                        log.error("check hold request failed. topic={}, queueId={}", topic, queueId, e);
                    }
                }
            }
        }

**遍历拉取任务表，根据主题与队列获取消息消费队列最大偏移量，如果该偏移量大于待拉取偏移量，说明有新的消息到达，调用notifyMessageArriving触发消息拉取。**

public void notifyMessageArriving(final String topic, final int queueId, final long maxOffset, final Long tagsCode,
            long msgStoreTime, byte[] filterBitMap, Map<String, String> properties) {
            String key = this.buildKey(topic, queueId);
            ManyPullRequest mpr = this.pullRequestTable.get(key);
            if (mpr != null) {
                List<PullRequest> requestList = mpr.cloneListAndClear();
                if (requestList != null) {
                    List<PullRequest> replayList = new ArrayList<PullRequest>();
    
                    for (PullRequest request : requestList) {
                        long newestOffset = maxOffset;
                        if (newestOffset <= request.getPullFromThisOffset()) {
                            newestOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId);
                        }
    
                        if (newestOffset > request.getPullFromThisOffset()) {
                            boolean match = request.getMessageFilter().isMatchedByConsumeQueue(tagsCode,
                                new ConsumeQueueExt.CqExtUnit(tagsCode, msgStoreTime, filterBitMap));
                            // match by bit map, need eval again when properties is not null.
                            if (match && properties != null) {
                                match = request.getMessageFilter().isMatchedByCommitLog(null, properties);
                            }
    
                            if (match) {
                                try {
                                    this.brokerController.getPullMessageProcessor().executeRequestWhenWakeup(request.getClientChannel(),
                                        request.getRequestCommand());
                                } catch (Throwable e) {
                                    log.error("execute request when wakeup failed.", e);
                                }
                                continue;
                            }
                        }
    
                        if (System.currentTimeMillis() >= (request.getSuspendTimestamp() + request.getTimeoutMillis())) {
                            try {
                                this.brokerController.getPullMessageProcessor().executeRequestWhenWakeup(request.getClientChannel(),
                                    request.getRequestCommand());
                            } catch (Throwable e) {
                                log.error("execute request when wakeup failed.", e);
                            }
                            continue;
                        }
    
                        replayList.add(request);
                    }
    
                    if (!replayList.isEmpty()) {
                        mpr.addPullRequest(replayList);
                    }
                }
            }
        }

1.  **首先从ManyPullRequest中获取当前该主题、队列所有的挂起拉取任务。值得注意的是该方法使用了synchronized，说明该数据结构会存在并发访问，该属性是PullRequestHoldService线程的私有属性。**
2.  **如果消息队列的最大偏移量大于待拉取偏移量，如果消息匹配则调用executeRequestWhenWakeup将消息返回给消息拉取客户端，否则等待下一次尝试。**
3.  **如果挂起超时时间超时，则不继续等待将直接返回客户消息未找到。**

public void executeRequestWhenWakeup(final Channel channel,
            final RemotingCommand request) throws RemotingCommandException {
            Runnable run = new Runnable() {
                @Override
                public void run() {
                    try {
                        final RemotingCommand response = PullMessageProcessor.this.processRequest(channel, request, false);

**其核心是设置brokerAllowSuspend为false，表示不支持拉取线程挂起，即当根据偏移量无法获取消息时将不挂起线程等待新消息到来，而是直接返回告诉客户端本次消息拉取未找到消息。**

**参考《RocketMQ技术内幕：RocketMQ架构设计与实现原理》**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70]: /images/20221021/cf3ad765af9849dbbf65c2e45060d7f1.png