聊聊rocketmq的pullThresholdForQueue
- 2019 年 11 月 25 日
- 筆記
序
本文主要研究一下rocketmq的pullThresholdForQueue
pullThresholdForQueue
rocketmq-client-4.5.2-sources.jar!/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java
public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsumer { //...... /** * Flow control threshold on queue level, each message queue will cache at most 1000 messages by default, * Consider the {@code pullBatchSize}, the instantaneous value may exceed the limit */ private int pullThresholdForQueue = 1000; /** * Limit the cached message size on queue level, each message queue will cache at most 100 MiB messages by default, * Consider the {@code pullBatchSize}, the instantaneous value may exceed the limit * * <p> * The size of a message only measured by message body, so it's not accurate */ private int pullThresholdSizeForQueue = 100; public int getPullThresholdForQueue() { return pullThresholdForQueue; } public void setPullThresholdForQueue(int pullThresholdForQueue) { this.pullThresholdForQueue = pullThresholdForQueue; } public int getPullThresholdSizeForQueue() { return pullThresholdSizeForQueue; } public void setPullThresholdSizeForQueue(final int pullThresholdSizeForQueue) { this.pullThresholdSizeForQueue = pullThresholdSizeForQueue; } //...... }
- DefaultMQPushConsumer定义了pullThresholdForQueue(
默认值1000)、pullThresholdSizeForQueue(默认值100)属性
checkConfig
rocketmq-client-4.5.2-sources.jar!/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
public class DefaultMQPushConsumerImpl implements MQConsumerInner { //...... private void checkConfig() throws MQClientException { Validators.checkGroup(this.defaultMQPushConsumer.getConsumerGroup()); //...... // pullThresholdForQueue if (this.defaultMQPushConsumer.getPullThresholdForQueue() < 1 || this.defaultMQPushConsumer.getPullThresholdForQueue() > 65535) { throw new MQClientException( "pullThresholdForQueue Out of range [1, 65535]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } // pullThresholdSizeForQueue if (this.defaultMQPushConsumer.getPullThresholdSizeForQueue() < 1 || this.defaultMQPushConsumer.getPullThresholdSizeForQueue() > 1024) { throw new MQClientException( "pullThresholdSizeForQueue Out of range [1, 1024]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } //...... } //...... }
- checkConfig方法要求defaultMQPushConsumer.getPullThresholdForQueue()必须大于等于1且小于等于65535；defaultMQPushConsumer.getPullThresholdSizeForQueue()必须大于等于1且小于等于1024
pullMessage
rocketmq-client-4.5.2-sources.jar!/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
public class DefaultMQPushConsumerImpl implements MQConsumerInner { //...... /** * Flow control interval */ private static final long PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL = 50; //...... public void pullMessage(final PullRequest pullRequest) { 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()) { 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; } 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; } //...... } private void executePullRequestLater(final PullRequest pullRequest, final long timeDelay) { this.mQClientFactory.getPullMessageService().executePullRequestLater(pullRequest, timeDelay); } //...... }
- pullMessage方法会判断cachedMessageCount(
processQueue.getMsgCount())是否大于defaultMQPushConsumer.getPullThresholdForQueue()，大于的话则执行executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL)然后提前返回；之后会判断cachedMessageSizeInMiB(processQueue.getMsgSize().get() / (1024 * 1024))是否大于defaultMQPushConsumer.getPullThresholdSizeForQueue()，大于的话也会执行executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL)然后提前返回
小结
- DefaultMQPushConsumer定义了pullThresholdForQueue(
cachedMessageCount，默认值1000)、pullThresholdSizeForQueue(cachedMessageSizeInMiB，默认值100)属性 - checkConfig方法要求defaultMQPushConsumer.getPullThresholdForQueue()必须大于等于1且小于等于65535；defaultMQPushConsumer.getPullThresholdSizeForQueue()必须大于等于1且小于等于1024
- cachedMessageCount若大于defaultMQPushConsumer.getPullThresholdForQueue()或者cachedMessageSizeInMiB大于defaultMQPushConsumer.getPullThresholdSizeForQueue()，都会执行executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL)然后提前返回
doc
- DefaultMQPushConsumer
