defaultmqpushconsumer - 腾讯云开发者社区

文章/答案/技术大牛

发布

rocketmq使用DefaultMQPushConsumer创建消费者客户端

样例 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("organization"); consumer.setNamesrvAddr...CONSUME_FROM_FIRST_OFFSET 从最早的消息开始消费 CONSUME_FROM_TIMESTAMP 从指定时间开始消费注：如果使用CONSUME_FROM_TIMESTAMP ,需设置参数 DefaultMQPushConsumer.setConsumeTimestamp...(“20131223171201”) 时间戳字符串格式为yyyyMMddHHmmss DefaultMQPushConsumer.subscribe(String topic, String subExpression...) subExpression参数为tag选择表达式语法：不过滤tag："*" 或者null 根据多个tag过滤：“tag1 || tag2 || tag3” DefaultMQPushConsumer.registerMessageListener

2K2 0

聊聊rocketmq的pullThresholdForQueue

/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java public class DefaultMQPushConsumer extends...()必须大于等于1且小于等于65535；defaultMQPushConsumer.getPullThresholdSizeForQueue()必须大于等于1且小于等于1024 pullMessage...(), this.defaultMQPushConsumer.getConsumerGroup()); this.executePullRequestLater(pullRequest...()必须大于等于1且小于等于65535；defaultMQPushConsumer.getPullThresholdSizeForQueue()必须大于等于1且小于等于1024 cachedMessageCount...若大于defaultMQPushConsumer.getPullThresholdForQueue()或者cachedMessageSizeInMiB大于defaultMQPushConsumer.getPullThresholdSizeForQueue

1.2K1 0

您找到你想要的搜索结果了吗？

是的

没有找到

聊聊rocketmq的pullThresholdForQueue

1.1K0 0

聊聊rocketmq的consumeThread

序本文主要研究一下rocketmq的consumeThread DefaultMQPushConsumer rocketmq-client-4.5.2-sources.jar!.../org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java public class DefaultMQPushConsumer extends...() < 1 || this.defaultMQPushConsumer.getConsumeThreadMin() > 1000) { throw new..."consumeThreadMin (" + this.defaultMQPushConsumer.getConsumeThreadMin() + ") " +...及submitConsumeRequestLater方法均会往consumeExecutor提交consumeRequest doc DefaultMQPushConsumer

1.4K1 0

聊聊rocketmq的consumeThread

序本文主要研究一下rocketmq的consumeThread th (38).jpeg DefaultMQPushConsumer rocketmq-client-4.5.2-sources.jar.../org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java public class DefaultMQPushConsumer extends...() < 1 || this.defaultMQPushConsumer.getConsumeThreadMin() > 1000) { throw new..."consumeThreadMin (" + this.defaultMQPushConsumer.getConsumeThreadMin() + ") " +...及submitConsumeRequestLater方法均会往consumeExecutor提交consumeRequest doc DefaultMQPushConsumer

2.5K0 0

聊聊rocketmq的consumeMessageBatchMaxSize

/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java public class DefaultMQPushConsumer extends...{ this.consumeMessageBatchMaxSize = consumeMessageBatchMaxSize; } //...... } DefaultMQPushConsumer...private void checkConfig() throws MQClientException { Validators.checkGroup(this.defaultMQPushConsumer.getConsumerGroup...() < 1 || this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() > 1024) {...consumeExecutor执行，若出现RejectedExecutionException则将剩余的msg添加到msgThis，然后执行submitConsumeRequestLater doc DefaultMQPushConsumer

2.5K0 0

RocketMQ 消费者启动流程

Pull Consumer Push Consumer DefaultMQPushConsumer DefaultMQPushCOnsumerImpl 通过 start 方法启动 org.apache.rocketmq.client.consumer.DefaultMQPushConsumer...DefaultMQPushConsumer#start 启动代码： public void start() throws MQClientException { setConsumerGroup...(), this.defaultMQPushConsumer.getMessageModel(), this.defaultMQPushConsumer.isUnitMode...() == MessageModel.CLUSTERING) { this.defaultMQPushConsumer.changeInstanceNameToPID...if (this.defaultMQPushConsumer.getOffsetStore() !

9791 0

聊聊rocketmq的registerConsumer与unregisterConsumer

(), this.defaultMQPushConsumer.getMessageModel(), this.defaultMQPushConsumer.isUnitMode...() == MessageModel.CLUSTERING) { this.defaultMQPushConsumer.changeInstanceNameToPID...()); this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel())...", this.defaultMQPushConsumer.getConsumerGroup()); this.serviceState = ServiceState.RUNNING...(), this)返回false的时候，会抛出MQClientException("The consumer group[" + this.defaultMQPushConsumer.getConsumerGroup

8140 0

聊聊rocketmq的pullThresholdForTopic

/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java public class DefaultMQPushConsumer extends...= -1) { if (this.defaultMQPushConsumer.getPullThresholdForTopic() defaultMQPushConsumer.getPullThresholdForTopic...() defaultMQPushConsumer.getPullThresholdSizeForTopic() > 102400) { throw...()在值非-1时必须大于等于1且小于等于6553500；defaultMQPushConsumer.getPullThresholdSizeForTopic()在值非-1时必须大于等于1且小于等于102400...()在值非-1时必须大于等于1且小于等于6553500；defaultMQPushConsumer.getPullThresholdSizeForTopic()在值非-1时必须大于等于1且小于等于102400

6330 0

聊聊rocketmq的pullThresholdForTopic

7341 0

聊聊rocketmq的consumeTimeout

/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java public class DefaultMQPushConsumer extends...defaultMQPushConsumer; private final MessageListenerConcurrently messageListener; private final...(), this.defaultMQPushConsumer.getConsumeThreadMax(), 1000 * 60,...(), this.defaultMQPushConsumer.getConsumeTimeout(), TimeUnit.MINUTES); } private void cleanExpireMsg...public void cleanExpiredMsg(DefaultMQPushConsumer pushConsumer) { if (pushConsumer.getDefaultMQPushConsumerImpl

1.1K3 0

聊聊rocketmq的pullBatchSize

/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java public class DefaultMQPushConsumer extends...()); //...... // pullBatchSize if (this.defaultMQPushConsumer.getPullBatchSize...() defaultMQPushConsumer.getPullBatchSize() > 1024) { throw new MQClientException...= null) { if (this.defaultMQPushConsumer.isPostSubscriptionWhenPull() && !...()传递进来的参数对应于maxNums值小结 DefaultMQPushConsumer定义了pullBatchSize属性，默认为32；DefaultMQPushConsumerImpl的checkConfig

7160 0

聊聊rocketmq的pullInterval

/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java public class DefaultMQPushConsumer extends...()); //...... // pullInterval if (this.defaultMQPushConsumer.getPullInterval...() defaultMQPushConsumer.getPullInterval() > 65535) { throw new MQClientException...方法会校验defaultMQPushConsumer.getPullInterval()必须大于等于0且小于等于65535；pullCallback的onSuccess方法判断defaultMQPushConsumer.getPullInterval...())，否则这执行executePullRequestImmediately(pullRequest) doc DefaultMQPushConsumer

5202 0

聊聊rocketmq的pullInterval

2K0 0

聊聊rocketmq的PushConsumerImpl

consumer/PushConsumerImpl.java public class PushConsumerImpl implements PushConsumer { private final DefaultMQPushConsumer...(), this.defaultMQPushConsumer.getMessageModel(), this.defaultMQPushConsumer.isUnitMode...() == MessageModel.CLUSTERING) { this.defaultMQPushConsumer.changeInstanceNameToPID...()); this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel())...", this.defaultMQPushConsumer.getConsumerGroup()); this.serviceState = ServiceState.RUNNING

1.3K2 0

聊聊rocketmq的consumeTimeout

/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java public class DefaultMQPushConsumer extends...; } public TraceDispatcher getTraceDispatcher() { return traceDispatcher; } } DefaultMQPushConsumer...defaultMQPushConsumer; private final MessageListenerConcurrently messageListener; private final...(), this.defaultMQPushConsumer.getConsumeThreadMax(), 1000 * 60,...(), this.defaultMQPushConsumer.getConsumeTimeout(), TimeUnit.MINUTES); } private void cleanExpireMsg

1.5K0 0

聊聊rocketmq的consumeMessageBatchMaxSize

/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java public class DefaultMQPushConsumer extends...) { this.consumeMessageBatchMaxSize = consumeMessageBatchMaxSize; } //...... } DefaultMQPushConsumer...private void checkConfig() throws MQClientException { Validators.checkGroup(this.defaultMQPushConsumer.getConsumerGroup...() < 1 || this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() > 1024) {...consumeExecutor执行，若出现RejectedExecutionException则将剩余的msg添加到msgThis，然后执行submitConsumeRequestLater doc DefaultMQPushConsumer

1.7K3 0

聊聊rocketmq的suspendCurrentQueueTimeMillis

/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java public class DefaultMQPushConsumer extends...{ this.suspendCurrentQueueTimeMillis = suspendCurrentQueueTimeMillis; } //...... } DefaultMQPushConsumer...timeMillis, TimeUnit.MILLISECONDS); } //...... } submitConsumeRequestLater方法在timeMillis为-1时会读取defaultMQPushConsumer.getSuspendCurrentQueueTimeMillis...如果该值小于10则重置为10，如果该值大于30000则重置为30000；然后使用scheduledExecutorService延时timeMillis执行submitConsumeRequest方法小结 DefaultMQPushConsumer...如果该值小于10则重置为10，如果该值大于30000则重置为30000；然后使用scheduledExecutorService延时timeMillis执行submitConsumeRequest方法 doc DefaultMQPushConsumer

7723 0

聊聊rocketmq的suspendCurrentQueueTimeMillis

/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java public class DefaultMQPushConsumer extends...this.suspendCurrentQueueTimeMillis = suspendCurrentQueueTimeMillis; } //...... } DefaultMQPushConsumer...timeMillis, TimeUnit.MILLISECONDS); } //...... } submitConsumeRequestLater方法在timeMillis为-1时会读取defaultMQPushConsumer.getSuspendCurrentQueueTimeMillis...如果该值小于10则重置为10，如果该值大于30000则重置为30000；然后使用scheduledExecutorService延时timeMillis执行submitConsumeRequest方法小结 DefaultMQPushConsumer...如果该值小于10则重置为10，如果该值大于30000则重置为30000；然后使用scheduledExecutorService延时timeMillis执行submitConsumeRequest方法 doc DefaultMQPushConsumer

8630 0

聊聊rocketmq的pullBatchSize

1.7K1 0

点击加载更多

rocketmq使用DefaultMQPushConsumer创建消费者客户端

聊聊rocketmq的pullThresholdForQueue

聊聊rocketmq的pullThresholdForQueue

聊聊rocketmq的consumeThread

聊聊rocketmq的consumeThread

聊聊rocketmq的consumeMessageBatchMaxSize

RocketMQ 消费者启动流程

聊聊rocketmq的registerConsumer与unregisterConsumer

聊聊rocketmq的pullThresholdForTopic

聊聊rocketmq的pullThresholdForTopic

聊聊rocketmq的consumeTimeout

聊聊rocketmq的pullBatchSize

聊聊rocketmq的pullInterval

聊聊rocketmq的pullInterval

聊聊rocketmq的PushConsumerImpl

聊聊rocketmq的consumeTimeout

聊聊rocketmq的consumeMessageBatchMaxSize

聊聊rocketmq的suspendCurrentQueueTimeMillis

聊聊rocketmq的suspendCurrentQueueTimeMillis

聊聊rocketmq的pullBatchSize

相关资讯

热门标签

活动推荐

运营活动

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐