RocketMQ源码分析——Broker接收拉消息

引：前面讲到了Broker对于不同的消息有自己不同的Processor，对于PushConsumer的Pull请求的时候，它对应的就是PullMessageProcessor~

PullMessageProcessor

通过前面的分析，我们自然而然地就会想到去看org.apache.rocketmq.broker.processor.PullMessageProcessor#processRequest()方法：

private RemotingCommand processRequest(final Channel channel, RemotingCommand request, boolean brokerAllowSuspend)
        throws RemotingCommandException {
        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);
        // 判断Broker当前是否可读
        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;
        }
        // 判断consumerGroup是否存在
        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;
        }
        // 判断consumerGroup的消费状态是否Enable
        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;
        // 判断请求的topic是否存在，并且可读，请求的queueId是否合法
        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;
        // 有设置subscribe flag,表示第一次pull或者需要更新filter
        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 {
            // 没有设置subscribe flag，表示之前已经订阅过了,对比订阅条件是否一致
            // 对于同一个ConsumerGroup下的多个consumer客户端，Broker要求订阅参数设置必须要是一致的，要不然会造成数据混乱
            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());
        }
        // 通过messageStore读取消息（带上topic和queueId和offset和过滤器）
        final GetMessageResult getMessageResult =
            this.brokerController.getMessageStore().getMessage(requestHeader.getConsumerGroup(), requestHeader.getTopic(),
                requestHeader.getQueueId(), requestHeader.getQueueOffset(), requestHeader.getMaxMsgNums(), messageFilter);
        if (getMessageResult != null) {
            response.setRemark(getMessageResult.getStatus().name());
            responseHeader.setNextBeginOffset(getMessageResult.getNextBeginOffset());
            responseHeader.setMinOffset(getMessageResult.getMinOffset());
            responseHeader.setMaxOffset(getMessageResult.getMaxOffset());
            // master读取比较慢的话，默认设置推荐从slave读取
            if (getMessageResult.isSuggestPullingFromSlave()) {
                responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getWhichBrokerWhenConsumeSlowly());
            } else {
                responseHeader.setSuggestWhichBrokerId(MixAll.MASTER_ID);
            }

            switch (this.brokerController.getMessageStoreConfig().getBrokerRole()) {
                case ASYNC_MASTER:
                case SYNC_MASTER:
                    break;
                case SLAVE:
                    // 如果当前broker是slave，并且不支持read，则提示客户端从master读
                    if (!this.brokerController.getBrokerConfig().isSlaveReadEnable()) {
                        response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
                        responseHeader.setSuggestWhichBrokerId(MixAll.MASTER_ID);
                    }
                    break;
            }
            // 如果slave可读，覆盖之前的设置
            if (this.brokerController.getBrokerConfig().isSlaveReadEnable()) {
                // consume too slow ,redirect to another machine
                if (getMessageResult.isSuggestPullingFromSlave()) {
                    responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getWhichBrokerWhenConsumeSlowly());
                }
                // consume ok
                else {
                    responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getBrokerId());
                }
            } else {
                responseHeader.setSuggestWhichBrokerId(MixAll.MASTER_ID);
            }
            // 将MessageStore读取的返回状态映射成repsonse的状态
            switch (getMessageResult.getStatus()) {
                case FOUND:
                    response.setCode(ResponseCode.SUCCESS);
                    break;
                case MESSAGE_WAS_REMOVING:
                    response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
                    break;
                case NO_MATCHED_LOGIC_QUEUE:
                case NO_MESSAGE_IN_QUEUE:
                    if (0 != requestHeader.getQueueOffset()) {
                        response.setCode(ResponseCode.PULL_OFFSET_MOVED);

                        // XXX: warn and notify me
                        log.info("the broker store no queue data, fix the request offset {} to {}, Topic: {} QueueId: {} Consumer Group: {}",
                            requestHeader.getQueueOffset(),
                            getMessageResult.getNextBeginOffset(),
                            requestHeader.getTopic(),
                            requestHeader.getQueueId(),
                            requestHeader.getConsumerGroup()
                        );
                    } else {
                        response.setCode(ResponseCode.PULL_NOT_FOUND);
                    }
                    break;
                case NO_MATCHED_MESSAGE:
                    response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
                    break;
                case OFFSET_FOUND_NULL:
                    response.setCode(ResponseCode.PULL_NOT_FOUND);
                    break;
                case OFFSET_OVERFLOW_BADLY:
                    response.setCode(ResponseCode.PULL_OFFSET_MOVED);
                    // XXX: warn and notify me
                    log.info("the request offset: {} over flow badly, broker max offset: {}, consumer: {}",
                        requestHeader.getQueueOffset(), getMessageResult.getMaxOffset(), channel.remoteAddress());
                    break;
                case OFFSET_OVERFLOW_ONE:
                    response.setCode(ResponseCode.PULL_NOT_FOUND);
                    break;
                case OFFSET_TOO_SMALL:
                    response.setCode(ResponseCode.PULL_OFFSET_MOVED);
                    log.info("the request offset too small. group={}, topic={}, requestOffset={}, brokerMinOffset={}, clientIp={}",
                        requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueOffset(),
                        getMessageResult.getMinOffset(), channel.remoteAddress());
                    break;
                default:
                    assert false;
                    break;
            }
            // 如果有回调函数
            // ...
            // 处理返回结果
            switch (response.getCode()) {
                case ResponseCode.SUCCESS:
                    // 成功后更新统计信息
                    this.brokerController.getBrokerStatsManager().incGroupGetNums(requestHeader.getConsumerGroup(), requestHeader.getTopic(),
                        getMessageResult.getMessageCount());

                    this.brokerController.getBrokerStatsManager().incGroupGetSize(requestHeader.getConsumerGroup(), requestHeader.getTopic(),
                        getMessageResult.getBufferTotalSize());

                    this.brokerController.getBrokerStatsManager().incBrokerGetNums(getMessageResult.getMessageCount());
                    // 如果是通过堆，则从文件中读取消息set到response body返回
                    if (this.brokerController.getBrokerConfig().isTransferMsgByHeap()) {
                        final long beginTimeMills = this.brokerController.getMessageStore().now();
                        final byte[] r = this.readGetMessageResult(getMessageResult, requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId());
                        this.brokerController.getBrokerStatsManager().incGroupGetLatency(requestHeader.getConsumerGroup(),
                            requestHeader.getTopic(), requestHeader.getQueueId(),
                            (int) (this.brokerController.getMessageStore().now() - beginTimeMills));
                        response.setBody(r);
                    } else {
                        // 否则，利用netty直接读取内存映射文件，少一次copy
                        try {
                            FileRegion fileRegion =
                                new ManyMessageTransfer(response.encodeHeader(getMessageResult.getBufferTotalSize()), getMessageResult);
                            channel.writeAndFlush(fileRegion).addListener(new ChannelFutureListener() {
                                @Override
                                public void operationComplete(ChannelFuture future) throws Exception {
                                    getMessageResult.release();
                                    if (!future.isSuccess()) {
                                        log.error("transfer many message by pagecache failed, {}", channel.remoteAddress(), future.cause());
                                    }
                                }
                            });
                        } catch (Throwable e) {
                            log.error("transfer many message by pagecache exception", e);
                            getMessageResult.release();
                        }

                        response = null;
                    }
                    break;
                // 没有读取到消息，则hold住请求，有新消息时唤醒
                // 等待超时后还是没读到brokerAllowSuspend=false
                case ResponseCode.PULL_NOT_FOUND:

                    if (brokerAllowSuspend && hasSuspendFlag) {
                      	// 默认长轮询时间为设置的挂起超时时间
                        long pollingTimeMills = suspendTimeoutMillisLong;
                        if (!this.brokerController.getBrokerConfig().isLongPollingEnable()) {
                          	// 没有开启长轮询，则轮询时间为1s
                            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);
                        // 利用PullRequestHoldService hold住PullRequest
                        this.brokerController.getPullRequestHoldService().suspendPullRequest(topic, queueId, pullRequest);
                        response = null;
                        break;
                    }

                case ResponseCode.PULL_RETRY_IMMEDIATELY:
                    break;
                // 消费开始的offset不正确
                case ResponseCode.PULL_OFFSET_MOVED:
                    if (this.brokerController.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE
                        || this.brokerController.getMessageStoreConfig().isOffsetCheckInSlave()) {
                        MessageQueue mq = new MessageQueue();
                        mq.setTopic(requestHeader.getTopic());
                        mq.setQueueId(requestHeader.getQueueId());
                        mq.setBrokerName(this.brokerController.getBrokerConfig().getBrokerName());

                        OffsetMovedEvent event = new OffsetMovedEvent();
                        event.setConsumerGroup(requestHeader.getConsumerGroup());
                        event.setMessageQueue(mq);
                        event.setOffsetRequest(requestHeader.getQueueOffset());
                        event.setOffsetNew(getMessageResult.getNextBeginOffset());
                        this.generateOffsetMovedEvent(event);
                        log.warn(
                            "PULL_OFFSET_MOVED:correction offset. topic={}, groupId={}, requestOffset={}, newOffset={}, suggestBrokerId={}",
                            requestHeader.getTopic(), requestHeader.getConsumerGroup(), event.getOffsetRequest(), event.getOffsetNew(),
                            responseHeader.getSuggestWhichBrokerId());
                    } else {
                        responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getBrokerId());
                        response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
                        log.warn("PULL_OFFSET_MOVED:none correction. topic={}, groupId={}, requestOffset={}, suggestBrokerId={}",
                            requestHeader.getTopic(), requestHeader.getConsumerGroup(), requestHeader.getQueueOffset(),
                            responseHeader.getSuggestWhichBrokerId());
                    }

                    break;
                default:
                    assert false;
            }
        } else {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("store getMessage return null");
        }
        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());
        }
        return response;
    }

从上面我们可以看出：我们会根据topic和queueId和offset和过滤器从MessageStore中读取消息，如果成功，我们可以从指向内存文件的ByteBuffer得到数据，根据配置有两种返回数据的方式，第一种是从ByteBuffer中将数据读取到response中（经过堆），然后返回。第二种是让netty直接读取ByteBuffer，将消息写给客户端，相对前一种，不需要将ByteBuffer中的数据copy到java Heap中，少一次内存copy。但是第二种方式无法记录监控信息，比如consumer消费延时（从代码看出）等指标；如果没有读取到消息，会判断是否是PushConsumer（参数hasSuspendFlag）以及是否允许挂起（参数brokerAllowSuspend），如果是的话，则将请求挂起，方式就是封装成PullRequest提交给PullRequestHoldService。

#MessageStore#getMessage

我们看看这个方法的代码：

public GetMessageResult getMessage(final String group, final String topic, final int queueId, final long offset,
    final int maxMsgNums,
    final MessageFilter messageFilter) {
    if (this.shutdown) {
        log.warn("message store has shutdown, so getMessage is forbidden");
        return null;
    }

    if (!this.runningFlags.isReadable()) {
        log.warn("message store is not readable, so getMessage is forbidden " + this.runningFlags.getFlagBits());
        return null;
    }

    long beginTime = this.getSystemClock().now();
    // 默认没有消息在队列
    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();
    // 找到对应的ConsumeQueue
    ConsumeQueue consumeQueue = findConsumeQueue(topic, queueId);
    if (consumeQueue != null) {
        minOffset = consumeQueue.getMinOffsetInQueue();
        maxOffset = consumeQueue.getMaxOffsetInQueue();
        // 一些offset不合法的情况
        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);
            }
        } else {
            // 根据offset找到对应的ConsumeQueue的MappedFile（包含ByteBuf）
            SelectMappedBufferResult bufferConsumeQueue = consumeQueue.getIndexBuffer(offset);
            if (bufferConsumeQueue != null) {
                try {
                    status = GetMessageStatus.NO_MATCHED_MESSAGE;

                    long nextPhyFileStartOffset = Long.MIN_VALUE;
                    long maxPhyOffsetPulling = 0;

                    int i = 0;
                    // 能返回的最大信息大小，不能大于16M
                    final int maxFilterMessageCount = Math.max(16000, maxMsgNums * ConsumeQueue.CQ_STORE_UNIT_SIZE);
                    final boolean diskFallRecorded = this.messageStoreConfig.isDiskFallRecorded();
                    // 封装CqExtUnit
                    ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();
                    for (; i < bufferConsumeQueue.getSize() && i < maxFilterMessageCount; i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
                        // 对应了CQUnit的结构
                        long offsetPy = bufferConsumeQueue.getByteBuffer().getLong();
                        int sizePy = bufferConsumeQueue.getByteBuffer().getInt();
                        long tagsCode = bufferConsumeQueue.getByteBuffer().getLong();

                        maxPhyOffsetPulling = offsetPy;
                        // 不是末尾的情况，offsetPy不正常则返回
                        if (nextPhyFileStartOffset != Long.MIN_VALUE) {
                            if (offsetPy < nextPhyFileStartOffset)
                                continue;
                        }

                        boolean isInDisk = checkInDiskByCommitOffset(offsetPy, maxOffsetPy);

                        if (this.isTheBatchFull(sizePy, maxMsgNums, getResult.getBufferTotalSize(), getResult.getMessageCount(),
                            isInDisk)) {
                            break;
                        }

                        boolean extRet = false, isTagsCodeLegal = true;
                        if (consumeQueue.isExtAddr(tagsCode)) {
                            extRet = consumeQueue.getExt(tagsCode, cqExtUnit);
                            if (extRet) {
                                tagsCode = cqExtUnit.getTagsCode();
                            } else {
                                // can't find ext content.Client will filter messages by tag also.
                                log.error("[BUG] can't find consume queue extend file content!addr={}, offsetPy={}, sizePy={}, topic={}, group={}",
                                    tagsCode, offsetPy, sizePy, topic, group);
                                isTagsCodeLegal = false;
                            }
                        }
                      	// 关于过滤，可以在参考出找到大佬的文章膜拜一下
                        // 是否有根据tagsCode匹配的消息
                        if (messageFilter != null
                            && !messageFilter.isMatchedByConsumeQueue(isTagsCodeLegal ? tagsCode : null, extRet ? cqExtUnit : null)) {
                            if (getResult.getBufferTotalSize() == 0) {
                                status = GetMessageStatus.NO_MATCHED_MESSAGE;
                            }

                            continue;
                        }
                        // 根据offset和size从CommitLog拿到具体的Message
                        SelectMappedBufferResult selectResult = this.commitLog.getMessage(offsetPy, sizePy);
                        if (null == selectResult) {
                            if (getResult.getBufferTotalSize() == 0) {
                                status = GetMessageStatus.MESSAGE_WAS_REMOVING;
                            }

                            nextPhyFileStartOffset = this.commitLog.rollNextFile(offsetPy);
                            continue;
                        }
                        // 是否有根据内容匹配的消息
                        if (messageFilter != null
                            && !messageFilter.isMatchedByCommitLog(selectResult.getByteBuffer().slice(), null)) {
                            if (getResult.getBufferTotalSize() == 0) {
                                status = GetMessageStatus.NO_MATCHED_MESSAGE;
                            }
                            // release...
                            selectResult.release();
                            continue;
                        }
                        // 增加一些统计信息
                        this.storeStatsService.getGetMessageTransferedMsgCount().incrementAndGet();
                        // 将Message放入结果集
                        getResult.addMessage(selectResult);
                        status = GetMessageStatus.FOUND;
                        nextPhyFileStartOffset = Long.MIN_VALUE;
                    }

                    if (diskFallRecorded) {
                        long fallBehind = maxOffsetPy - maxPhyOffsetPulling;
                        brokerStatsManager.recordDiskFallBehindSize(group, topic, queueId, fallBehind);
                    }
                    // 更新offset
                    nextBeginOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);

                    long diff = maxOffsetPy - maxPhyOffsetPulling;
                    long memory = (long) (StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE
                        * (this.messageStoreConfig.getAccessMessageInMemoryMaxRatio() / 100.0));
                    getResult.setSuggestPullingFromSlave(diff > memory);
                } finally {

                    bufferConsumeQueue.release();
                }
            } else {
                status = GetMessageStatus.OFFSET_FOUND_NULL;
                nextBeginOffset = nextOffsetCorrection(offset, consumeQueue.rollNextFile(offset));
                log.warn("consumer request topic: " + topic + "offset: " + offset + " minOffset: " + minOffset + " maxOffset: "
                    + maxOffset + ", but access logic queue failed.");
            }
        }
    } else {
        status = GetMessageStatus.NO_MATCHED_LOGIC_QUEUE;
        nextBeginOffset = nextOffsetCorrection(offset, 0);
    }

    if (GetMessageStatus.FOUND == status) {
        this.storeStatsService.getGetMessageTimesTotalFound().incrementAndGet();
    } else {
        this.storeStatsService.getGetMessageTimesTotalMiss().incrementAndGet();
    }
    long eclipseTime = this.getSystemClock().now() - beginTime;
    this.storeStatsService.setGetMessageEntireTimeMax(eclipseTime);

    getResult.setStatus(status);
    getResult.setNextBeginOffset(nextBeginOffset);
    getResult.setMaxOffset(maxOffset);
    getResult.setMinOffset(minOffset);
    return getResult;
}

如果理解了Broker的信息存储，上面的这些代码粗略的看还是比较好懂的。

主要是先找到topic+queue对应的ConsumeQueue，根据PullRequest传入的offset找到
对应的MappedFile；从MappedFile中里面读取指定数量的CQUnit，根据TagsCode做下过滤，然后得到过滤后的Commit log的offset；然后根据offset从CommitLog中获取具体的Message；最后再根据消息内容再做一次过滤，然后返回结果。

PullRequestHoldService#suspendPullRequest

从Processor的处理来看，如果没有读取到消息，会判断是否是PushConsumer（参数hasSuspendFlag）以及是否允许挂起（参数brokerAllowSuspend），如果是的话，则将请求挂起，方式就是封装成PullRequest提交给PullRequestHoldService。所以我们这里看看它是如何挂起请求处理的：

  // org.apache.rocketmq.broker.longpolling.PullRequestHoldService
public void suspendPullRequest(final String topic, final int queueId, final PullRequest pullRequest) {
      String key = this.buildKey(topic, queueId);
      // pullRequest被放入一个以topic+queue为key的Map中， value为一个队列ManyPullRequest
      ManyPullRequest mpr = this.pullRequestTable.get(key);
      if (null == mpr) {
          mpr = new ManyPullRequest();
          ManyPullRequest prev = this.pullRequestTable.putIfAbsent(key, mpr);
          if (prev != null) {
              mpr = prev;
          }
      }

      mpr.addPullRequest(pullRequest);
  }

然后我们回想一下，我们是不是在BrokerController启动中看到了PullRequestHoldService(继承ServiceThread)被启动了，所以我们应该去看看PullRequestHoldService做了什么，所以首先去看他的run方法：

   public void run() {
       log.info("{} service started", this.getServiceName());
       while (!this.isStopped()) {
           try {
               // 是否开启长轮询, 长轮询等待5秒，短轮询等待1秒
         			// 如何过被通知有消息了就不会等，没有就等待
               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());
   }
   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);
               }
           }
       }
   }
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);
       // 拿到topic+queueId对应的所有PullRequest
       ManyPullRequest mpr = this.pullRequestTable.get(key);
       if (mpr != null) {
           // clone等待的List，同时会清空等待列表
           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);
                       }
                       // 匹配到消息，重新通过PullRequestProcessor执行消息读取
                       if (match) {
                           try {
                               this.brokerController.getPullMessageProcessor().executeRequestWhenWakeup(request.getClientChannel(),
                                   request.getRequestCommand());
                           } catch (Throwable e) {
                               log.error("execute request when wakeup failed.", e);
                           }
                           continue;
                       }
                   }
                   // 如果requst等待超时，无论前一步是否符合条件，肯定会发给processor处理
                   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;
                   }
                   // 未超时和没有匹配到消息的request，重新放入队列等待
                   replayList.add(request);
               }

               if (!replayList.isEmpty()) {
                   mpr.addPullRequest(replayList);
               }
           }
       }
   }
	// org.apache.rocketmq.broker.processor.PullMessageProcessor
   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);
									// 不为空就直接推给消费者
                   if (response != null) {
                       response.setOpaque(request.getOpaque());
                       response.markResponseType();
                       try {
                           channel.writeAndFlush(response).addListener(new ChannelFutureListener() {
                               @Override
                               public void operationComplete(ChannelFuture future) throws Exception {
                                   if (!future.isSuccess()) {
                                       log.error("processRequestWrapper response to {} failed",
                                           future.channel().remoteAddress(), future.cause());
                                       log.error(request.toString());
                                       log.error(response.toString());
                                   }
                               }
                           });
                       } catch (Throwable e) {
                           log.error("processRequestWrapper process request over, but response failed", e);
                           log.error(request.toString());
                           log.error(response.toString());
                       }
                   }
               } catch (RemotingCommandException e1) {
                   log.error("excuteRequestWhenWakeup run", e1);
               }
           }
       };
       this.brokerController.getPullMessageExecutor().submit(new RequestTask(run, channel, request));
   }

然后我们在回想一下PullRequestHoldService#notifyMessageArriving()方法，我们是否似曾相识，对了，就是在org.apache.rocketmq.store.DefaultMessageStore.ReputMessageService#doReput()方法时：

if (BrokerRole.SLAVE != DefaultMessageStore.this.getMessageStoreConfig().getBrokerRole()
                                        && DefaultMessageStore.this.brokerConfig.isLongPollingEnable()) {
	// 唤醒等待的PullReqeust接收消息                              
  DefaultMessageStore.this.messageArrivingListener.arriving(dispatchRequest.getTopic(),
                                              dispatchRequest.getQueueId(), dispatchRequest.getConsumeQueueOffset() + 1,
                                              dispatchRequest.getTagsCode(), dispatchRequest.getStoreTimestamp(),
                                              dispatchRequest.getBitMap(), dispatchRequest.getPropertiesMap());
}

OK！到这里，Broker就已经完成了PullMessage的接收啦！！

参考

1. RocketMQ源码解析(十一)-Broker#消息消费
2. RocketMQ 消息过滤流程