当前位置：首页 > article >正文

DefaultMQPushConsumer的整体流程

article 2024/10/24 19:27:00

我们使用DefaultMQPushConsumer的时候，一般流程是设置好GroupName、NameServer地址，以及订阅的Topic名称，然后填充Message处理函数，最后调用start（）。本节基于这个流程来分析源码。

1 上层接口类

DefaultMQPushConsumer类在org.apache.rocketmq.client.consumer包中，这个类担任着上层接口的角色，具体实现都在DefaultMQPushConsumerImpl类中，如代码清单11-1所示。

代码清单11-1 DefaultMQPushConsumer接口类

/**
* Default constructor.
*/
public DefaultMQPushConsumer() {
    this(MixAll.DEFAULT_CONSUMER_GROUP, null, new
        AllocateMessageQueueAveragely());
}
/**
* Constructor specifying consumer group, RPC hook and message queue
* allocating algorithm.
*
* @param consumerGroup Consume queue.
* @param rpcHook RPC hook to execute before each remoting command.
* @param allocateMessageQueueStrategy message queue allocating algorithm.
*/
public DefaultMQPushConsumer(final String consumerGroup, RPCHook rpcHook,
    AllocateMessageQueueStrategy allocateMessageQueueStrategy) {
    this.consumerGroup = consumerGroup;
  this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
    defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(this,
        rpcHook);
}
/**
* Constructor specifying RPC hook.
*
* @param rpcHook RPC hook to execute before each remoting command.
*/
public DefaultMQPushConsumer(RPCHook rpcHook) {
    this(MixAll.DEFAULT_CONSUMER_GROUP, rpcHook, new
        AllocateMessageQueueAveragely());
}

/**
* Constructor specifying consumer group.
*
* @param consumerGroup Consumer group.
*/
public DefaultMQPushConsumer(final String consumerGroup) {
    this(consumerGroup, null, new AllocateMessageQueueAveragely());
}

我们常用的是最后这个构造函数，只传入一个consumer Group名称作为参数，这个构造函数会把RPCHoop设为空，把负载均衡策略设置成平均策略。在构造函数的实现中，主要工作是创建DefaultMQPushConsumerImpl对象。

2 DefaultMQPushConsumer的实现者

DefaultMQPushConsumerImpl具体实现了DefaultMQPushConsumer的业务逻辑，DefaultMQPushConsumerImpl.java在org.apache.rocketmq.client.impl.consumer这个包里，本节接下来从start方法着手分析。

首先是初始化MQClientInstance，并且设置好rebalance策略和pullApiWraper，有这些结构后才能发送pull请求获取消息，如代码清单11-2所示。

代码清单11-2 初始化MQClientInstance和pullApiWraper

this.mQClientFactory = MQClientManager.getInstance()
    .getAndCreateMQClientInstance(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);

然后是确定OffsetStore。OffsetStore里存储的是当前消费者所消费的消息在队列中的偏移量，如代码清单11-3所示。

代码清单11-3 初始化OffsetStore

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

根据消费消息方式的不同，OffsetStore的类型也不同。如果是BROADCASTING模式，使用的是LocalFileOffsetStore，Offset存到本地；如果是CLUSTERING模式，使用的是RemoteBrokerOffsetStore，Offset存到Broker机器上。

然后是初始化consumeMessageService，根据对消息顺序需求的不同，使用不同的Service类型，如代码清单11-4所示。

代码清单11-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();

最后调用MQClientInstance的start方法，开始获取数据。

3 获取消息逻辑

获取消息的逻辑实现在public void pullMessage（final PullRequest pullRequest）函数中，这是一个很大的函数，前半部分是进行一些判断，是进行流量控制的逻辑（见代码清单11-5）；中间是对返回消息结果做处理的逻辑；后面是发送获取消息请求的逻辑。

代码清单11-5 流量控制逻辑

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

通过判断未处理消息的个数和总大小来控制是否继续请求消息。对于顺序消息还有一些特殊判断逻辑。获取的消息返回后，根据返回状态，调用相应的处理方法，如代码清单11-6所示。

代码清单11-6 对返回消息结果做处理

switch (pullResult.getPullStatus()) {
    case FOUND:
        long prevRequestOffset = pullRequest
            .getNextOffset();
        pullRequest.setNextOffset(pullResult
            .getNextBeginOffset());
        ……
        break;
    case NO_NEW_MSG:
        pullRequest.setNextOffset(pullResult
            .getNextBeginOffset());
        DefaultMQPushConsumerImpl.this.correctTagsOffset
            (pullRequest);
        DefaultMQPushConsumerImpl.this
            .executePullRequestImmediately(pullRequest);
        break;
    case NO_MATCHED_MSG:
        pullRequest.setNextOffset(pullResult
            .getNextBeginOffset());
        DefaultMQPushConsumerImpl.this.correctTagsOffset
            (pullRequest);
        DefaultMQPushConsumerImpl.this
            .executePullRequestImmediately(pullRequest);
        break;
    case OFFSET_ILLEGAL:
        log.warn("the pull request offset illegal, {} {}",
            pullRequest.toString(), pullResult.toString());
        pullRequest.setNextOffset(pullResult
            .getNextBeginOffset());
……        break;
    default:
        break;
}

最后是发送获取消息请求，这三个阶段不停地循环执行，直到程序被停止，如代码清单11-7所示。

代码清单11-7 发送pull请求

try {
    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
    );
} catch (Exception e) {
    log.error("pullKernelImpl exception", e);
    this.executePullRequestLater(pullRequest,
        PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
}