消息拉取

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, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
    return;
}

if (this.isPause()) {
    this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_SUSPEND);
    return;
}

long size = processQueue.getMsgCount().get();
if (size > this.defaultMQPushConsumer.getPullThresholdForQueue()) {
    this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
    if ((flowControlTimes1++ % 1000) == 0) {
        // 省略流控输出语句
    }
    return;
}

if (processQueue.getMaxSpan() > this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan()) {
    this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
    if ((flowControlTimes2++ % 1000) == 0) {
        // 省略流控输出语句
    }
    return;
}

final SubscriptionData subscriptionData = this.rebalanceImpl
    .getSubscriptionInner()
    .get(pullRequest.getMessageQueue().getTopic());

if (null == subscriptionData) {
    this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
    return;
}

boolean commitOffsetEnable = false;
long commitOffsetValue = 0L;

if (MessageModel.CLUSTERING == this.defaultMQPushConsumer.getMessageModel()) {
    commitOffsetValue = this.offsetStore.readOffset(
        pullRequest.getMessageQueue(), ReadOffsetType.READ_FROM_MEMORY
    );
    if (commitOffsetValue > 0) {
        commitOffsetEnable = true;
    }
}

String subExpression = null;
boolean classFilter = false;

if (sd != null) {
    if (this.defaultMQPushConsumer.isPostSubscriptionWhenPull() && !sd.isClassFilterMode()) {
        subExpression = sd.getSubString();
    }
    classFilter = sd.isClassFilterMode();
}

int sysFlag = PullSysFlag.buildSysFlag(
    commitOffsetEnable, // commit offset
    true,
    subExpression != null,
    classFilter // class filter
);

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
);

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
    );
}

String brokerAddr = findBrokerResult.getBrokerAddr();

if (PullSysFlag.hasClassFilterFlag(sysFlagInner)) {
    brokerAddr = computPullFromWhichFilterServer(mq.getTopic(), brokerAddr);
}

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

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 = findConsumeQueue(topic, queueId);

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);
    }
}

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

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()
    );
}

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());
}

return new PullResultExt(); // （省略PullResultExt拉取结果的解码）

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

long prevRequestOffset = pullRequest.getNextOffset();
pullRequest.setNextOffset(pullResult.getNextBeginOffset());

if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {
    DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
}

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

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

case NO_NEW_MSG:
case NO_MATCHED_MSG:
    pullRequest.setNextOffset(pullResult.getNextBeginOffset());
    DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);
    DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
    break;

pullRequest.setNextOffset(pullResult.getNextBeginOffset());
pullRequest.getProcessQueue().setDropped(true);

DefaultMQPushConsumerImpl.this.executeTaskLater(new Runnable() {
    public void run() {
        try {
            DefaultMQPushConsumerImpl.this.offsetStore.updateOffset(
                pullRequest.getMessageQueue(), pullRequest.getNextOffset(), false
            );
            DefaultMQPushConsumerImpl.this.offsetStore.persist(pullRequest.getMessageQueue());
            DefaultMQPushConsumerImpl.this.rebalanceImpl.removeProcessQueue(pullRequest.getMessageQueue());
        } catch (Throwable e) {
            // Handle the exception if necessary
        }
    }
}, 10000);

private RemotingCommand processRequest(final Channel channel, RemotingCommand request, boolean brokerAllowSuspend) {
    // 省略相关代码

    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;
        }
}