11、RocketMQ的Comsumer的拉取消息概念以及入口

迷南。 2024-02-05 14:44 4阅读 0赞

消息的拉取分为Push模式和Pull模式，具体的实现类为Push模式对应DefaultMQPushConsumer而Pull模式对应DefaultMQPullConsumer，但是在4.9.0之后就把这个类废弃了。

Push模式下由Broker服务主动将消息推送给消费者，时效性高只要Broker有消息就会进行推送，但是这样也加重的Broker的负担。

Pull模式下就是由消费者主动到Broker端拉取消息，消费者可以根据自己的消费能力来控制拉取消息，灵活性比较高。

根据对源码的理解其实Push模式和Pull模式在RocketMQ的实现层面都是Pull模式，好处当然是为了减少网络开销以及提高消息的推送率同时满足所有pull模式的好处，

这时候就有很多人会问：“那么它是如何保证时效性呢？”

这就是RocketMQ的特色点长轮询机制，简答理解为当Broker接收到Consumer的Pull请求，判断如果没有对应的消息，不用直接给Consumer响应，而是将这个Pull请求缓存起来，当Producer发送消息过来时，增加一个步骤去检查是否有对应的已缓存的Pull请求，如果有就及时将请求从缓存中拉取出来并将消息通知给Consumer（具体如下图）。

![在这里插入图片描述][e20635c61aa84c448de5470685fcdcdb.png]  
Push模式

Push模式下，消息的拉取由PullMessageService服务来实现，在前面消费者启动的时候会去启动this.pullMessageService.start();拉取消息的服务，该类是线程类调用start方法其实就是调用run方法。

public void run() {
        
        log.info(this.getServiceName() + " service started");
        // 服务没停止一直取获取，如果没有就等待
        while (!this.isStopped()) {
        
            try {
        
                PullRequest pullRequest = this.pullRequestQueue.take();
                // 存在就会调用pullMessage方法拉取消息
                this.pullMessage(pullRequest);
            } catch (InterruptedException ignored) {
        
            } catch (Exception e) {
        
                log.error("Pull Message Service Run Method exception", e);
            }
        }
    
        log.info(this.getServiceName() + " service end");
    }

真正去拉取消息会调用DefaultMQPushConsumerImpl的pullMessage，该方法处理过程很长这里我只贴出关键核心的代码整体理解一下代码即可。

public void pullMessage(final PullRequest pullRequest) {
        
      // 判断状态,确保Consumer状态正常
        this.makeSureStateOK();
    
        // 流量控制，防止消费者消息积压 默认1000
        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_CACHE_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;
        }
    
      // 消息大小限流，默认100M
        if (cachedMessageSizeInMiB > this.defaultMQPushConsumer.getPullThresholdSizeForQueue()) {
        
            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_CACHE_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;
        }
      // 并发消息：如果内存中消息的offset的最大跨度大于设置的阈值，默认2000
        // 顺序消息
        if (!this.consumeOrderly) {
        
            if (processQueue.getMaxSpan() > this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan()) {
        
                this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_CACHE_FLOW_CONTROL);
                if ((queueMaxSpanFlowControlTimes++ % 1000) == 0) {
        
                    log.warn(
                        "the queue's messages, span too long, so do flow control, minOffset={}, maxOffset={}, maxSpan={}, pullRequest={}, flowControlTimes={}",
                        processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), processQueue.getMaxSpan(),
                        pullRequest, queueMaxSpanFlowControlTimes);
                }
                return;
            }
        }
        // 创建拉取消息的回调函数，成功onSuccess，失败onException
       PullCallback pullCallback = new PullCallback() {
        
                @Override
               // 成功执行
                public void onSuccess(PullResult pullResult) {
        
                    // 拉取消息
                    DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                    // 提交拉取消息的请求
                    DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
                        pullResult.getMsgFoundList(),
                        processQueue,
                        pullRequest.getMessageQueue(),
                        dispatchToConsume);
                }
          @Override
                // 失败执行
                public void onException(Throwable e) {
        
                    if (!pullRequest.getMessageQueue().getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
        
                        log.warn("execute the pull request exception", e);
                    }
    
                    if (e instanceof MQBrokerException && ((MQBrokerException) e).getResponseCode() == ResponseCode.FLOW_CONTROL) {
        
                       // 如果返回code是流程控制就延迟20s
                        DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_BROKER_FLOW_CONTROL);
                    } else {
        
                        // 延迟3s钟，将PullRequest对象再次放入队列pullRequestQueue中，等待再次take()
                        DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
                    }
                }   
         }
    
        // 判断是否需要上报点位
        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;
            }
        }
      // 真正去拉取消息
        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
        );
    }

校验拉取是否超出设定值，超出则延迟拉取。

判断consumer是否为正常状态，异常则延迟。

判断流量控制当已缓存的消息大于1000条或者消息内容大小大于100M就会导致延迟。

判断顺序消费与并发消费，并发消费为消息的offset大于2000就会延迟而顺序消息如果未锁定就会延迟。

如果是第一次锁定就得设置消费点位。然后就是去创建PullCallback对象，实现消息请求成功的回调方法onSuccess与请求失败回调方法onException。

（简单理解就是一堆约束性判断，最后去构建一个PullCallback对象）。

判断是否需要上报点位，其实就是需要同步offset，是广播模式offset在本地则更新本地，是集群模式offset在远程的Broker则通知Broker进行持久化。最后真正去执行消息拉取方法this.pullAPIWrapper.pullKernelImpl()。

pullKernelImpl与pullMessage方法

public PullResult pullKernelImpl(
            final MessageQueue mq,final String subExpression,final String expressionType,
            final long subVersion,final long offset,final int maxNums,final int sysFlag,
            final long commitOffset,final long brokerSuspendMaxTimeMillis,final long timeoutMillis,
            final CommunicationMode communicationMode,final PullCallback pullCallback
        ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        
        // 获取指定brokerName的Broker地址
        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);
            }
    
            if (findBrokerResult != null) {
        
                {
        
                    // check version
                    if (!ExpressionType.isTagType(expressionType)
                        && findBrokerResult.getBrokerVersion() < MQVersion.Version.V4_1_0_SNAPSHOT.ordinal()) {
        
                        throw new MQClientException("The broker[" + mq.getBrokerName() + ", "
                            + findBrokerResult.getBrokerVersion() + "] does not upgrade to support for filter message by " + expressionType, null);
                    }
                }
                int sysFlagInner = sysFlag;
              // 如果拉取消息的不是Master节点，则不进行消息确认
                if (findBrokerResult.isSlave()) {
        
                    sysFlagInner = PullSysFlag.clearCommitOffsetFlag(sysFlagInner);
                }
              // 构建请求头
                PullMessageRequestHeader requestHeader = new PullMessageRequestHeader();
                // 消费者组
                requestHeader.setConsumerGroup(this.consumerGroup);
                // 主题
                requestHeader.setTopic(mq.getTopic());
                // 队列
                requestHeader.setQueueId(mq.getQueueId());
                // 队列offset
                requestHeader.setQueueOffset(offset);
                requestHeader.setMaxMsgNums(maxNums);
                requestHeader.setSysFlag(sysFlagInner);
                requestHeader.setCommitOffset(commitOffset);
                requestHeader.setSuspendTimeoutMillis(brokerSuspendMaxTimeMillis);
                requestHeader.setSubscription(subExpression);
                requestHeader.setSubVersion(subVersion);
                requestHeader.setExpressionType(expressionType);
                requestHeader.setBname(mq.getBrokerName());
    
                String brokerAddr = findBrokerResult.getBrokerAddr();
                if (PullSysFlag.hasClassFilterFlag(sysFlagInner)) {
        
                    brokerAddr = computePullFromWhichFilterServer(mq.getTopic(), brokerAddr);
                }
              // 发送请求去拉取消息
                PullResult pullResult = this.mQClientFactory.getMQClientAPIImpl().pullMessage(
                    brokerAddr,
                    requestHeader,
                    timeoutMillis,
                    communicationMode,
                    pullCallback);
    
                return pullResult;
            }
    
            throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
        }

根据指定BrokerName获取Broker的地址，如果非Master就更新topic在获取，找到Broker后通过Broker的相关信息以及传入的相关信息构建一个PullMessageRequestHeader类的请求头，用来进行消息的发送进行消息的拉取，得到请求体后调用pullMessage方法进行消息的拉取。

public PullResult pullMessage(
            final String addr,
            final PullMessageRequestHeader requestHeader,
            final long timeoutMillis,
            final CommunicationMode communicationMode,
            final PullCallback pullCallback
    ) throws RemotingException, MQBrokerException, InterruptedException {
        
        // 构建远程消息处理的Code，Broker根据对应code找出Processor进行消息的处理
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, requestHeader);
      // 判断方式
        switch (communicationMode) {
        
                // 单向不处理
            case ONEWAY:
                assert false;
                return null;
                // 异步拉取消息,将拉取后的消息交给前面构造的PullCallBack处理
            case ASYNC:
                this.pullMessageAsync(addr, request, timeoutMillis, pullCallback);
                return null;
                // 同步拉取消息
            case SYNC:
                return this.pullMessageSync(addr, request, timeoutMillis);
            default:
                assert false;
                break;
        }
    
        return null;
    }

构建消息体，指定拉取消息在Broker端处理的Code，在后面Broker处理时就可以根据指定的Code来找出对应的Processor，进行拉取消息的处理。

判断拉取消息的方式，单向不处理，异步拉取消息执行pullMessageAsync方法当处理完成就会去执行PullCallBack的回调方法，如果处理成功则调用onSuccess方法，如果处理失败则调用onException方法，同步拉取直接调用pullMessageSync方法。

[e20635c61aa84c448de5470685fcdcdb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/30/87adbca97c3a40cbb31926aec2695cef.png