@Test public void checkConfigTest() throws MQClientException { thrown.expect(MQClientException.class); thrown.expectMessage("consumeThreadMin (10) is larger than consumeThreadMax (9)"); DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_consumer_group"); consumer.setConsumeThreadMin(10); consumer.setConsumeThreadMax(9); consumer.registerMessageListener(new MessageListenerConcurrently() { public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { System.out.println(" Receive New Messages: " + msgs); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); DefaultMQPushConsumerImpl defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(consumer, null); defaultMQPushConsumerImpl.start(); }
private void checkConfig() throws MQClientException { Validators.checkGroup(this.defaultMQPushConsumer.getConsumerGroup()); if (null == this.defaultMQPushConsumer.getConsumerGroup()) { throw new MQClientException( "consumerGroup is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumerGroup().equals(MixAll.DEFAULT_CONSUMER_GROUP)) { throw new MQClientException( "consumerGroup can not equal " + MixAll.DEFAULT_CONSUMER_GROUP + ", please specify another one." + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getMessageModel()) { throw new MQClientException( "messageModel is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getConsumeFromWhere()) { throw new MQClientException( "consumeFromWhere is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } Date dt = UtilAll.parseDate(this.defaultMQPushConsumer.getConsumeTimestamp(), UtilAll.YYYYMMDDHHMMSS); if (null == dt) { throw new MQClientException( "consumeTimestamp is invalid, the valid format is yyyyMMddHHmmss,but received " + this.defaultMQPushConsumer.getConsumeTimestamp() + " " + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getAllocateMessageQueueStrategy()) { throw new MQClientException( "allocateMessageQueueStrategy is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getSubscription()) { throw new MQClientException( "subscription is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getMessageListener()) { throw new MQClientException( "messageListener is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } boolean orderly = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerOrderly; boolean concurrently = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerConcurrently; if (!orderly && !concurrently) { throw new MQClientException( "messageListener must be instanceof MessageListenerOrderly or MessageListenerConcurrently" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeThreadMin() < 1 || this.defaultMQPushConsumer.getConsumeThreadMin() > 1000) { throw new MQClientException( "consumeThreadMin Out of range [1, 1000]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeThreadMax() < 1 || this.defaultMQPushConsumer.getConsumeThreadMax() > 1000) { throw new MQClientException( "consumeThreadMax Out of range [1, 1000]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeThreadMin() > this.defaultMQPushConsumer.getConsumeThreadMax()) { throw new MQClientException( "consumeThreadMin (" + this.defaultMQPushConsumer.getConsumeThreadMin() + ") " + "is larger than consumeThreadMax (" + this.defaultMQPushConsumer.getConsumeThreadMax() + ")", null); } if (this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan() < 1 || this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan() > 65535) { throw new MQClientException( "consumeConcurrentlyMaxSpan Out of range [1, 65535]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } 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); } if (this.defaultMQPushConsumer.getPullThresholdForTopic() != -1) { if (this.defaultMQPushConsumer.getPullThresholdForTopic() < 1 || this.defaultMQPushConsumer.getPullThresholdForTopic() > 6553500) { throw new MQClientException( "pullThresholdForTopic Out of range [1, 6553500]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } } 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); } if (this.defaultMQPushConsumer.getPullThresholdSizeForTopic() != -1) { if (this.defaultMQPushConsumer.getPullThresholdSizeForTopic() < 1 || this.defaultMQPushConsumer.getPullThresholdSizeForTopic() > 102400) { throw new MQClientException( "pullThresholdSizeForTopic Out of range [1, 102400]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } } if (this.defaultMQPushConsumer.getPullInterval() < 0 || this.defaultMQPushConsumer.getPullInterval() > 65535) { throw new MQClientException( "pullInterval Out of range [0, 65535]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() < 1 || this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() > 1024) { throw new MQClientException( "consumeMessageBatchMaxSize Out of range [1, 1024]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getPullBatchSize() < 1 || this.defaultMQPushConsumer.getPullBatchSize() > 1024) { throw new MQClientException( "pullBatchSize Out of range [1, 1024]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } }
DefaultMQPushConsumerImpl implements MQConsumerInner { private void checkConfig() throws MQClientException { Validators.checkGroup(this.defaultMQPushConsumer.getConsumerGroup()); if (null == this.defaultMQPushConsumer.getConsumerGroup()) { throw new MQClientException( "consumerGroup is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumerGroup().equals(MixAll.DEFAULT_CONSUMER_GROUP)) { throw new MQClientException( "consumerGroup can not equal " + MixAll.DEFAULT_CONSUMER_GROUP + ", please specify another one." + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getMessageModel()) { throw new MQClientException( "messageModel is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getConsumeFromWhere()) { throw new MQClientException( "consumeFromWhere is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } Date dt = UtilAll.parseDate(this.defaultMQPushConsumer.getConsumeTimestamp(), UtilAll.YYYYMMDDHHMMSS); if (null == dt) { throw new MQClientException( "consumeTimestamp is invalid, the valid format is yyyyMMddHHmmss,but received " + this.defaultMQPushConsumer.getConsumeTimestamp() + " " + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getAllocateMessageQueueStrategy()) { throw new MQClientException( "allocateMessageQueueStrategy is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getSubscription()) { throw new MQClientException( "subscription is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getMessageListener()) { throw new MQClientException( "messageListener is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } boolean orderly = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerOrderly; boolean concurrently = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerConcurrently; if (!orderly && !concurrently) { throw new MQClientException( "messageListener must be instanceof MessageListenerOrderly or MessageListenerConcurrently" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeThreadMin() < 1 || this.defaultMQPushConsumer.getConsumeThreadMin() > 1000) { throw new MQClientException( "consumeThreadMin Out of range [1, 1000]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeThreadMax() < 1 || this.defaultMQPushConsumer.getConsumeThreadMax() > 1000) { throw new MQClientException( "consumeThreadMax Out of range [1, 1000]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeThreadMin() > this.defaultMQPushConsumer.getConsumeThreadMax()) { throw new MQClientException( "consumeThreadMin (" + this.defaultMQPushConsumer.getConsumeThreadMin() + ") " + "is larger than consumeThreadMax (" + this.defaultMQPushConsumer.getConsumeThreadMax() + ")", null); } if (this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan() < 1 || this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan() > 65535) { throw new MQClientException( "consumeConcurrentlyMaxSpan Out of range [1, 65535]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } 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); } if (this.defaultMQPushConsumer.getPullThresholdForTopic() != -1) { if (this.defaultMQPushConsumer.getPullThresholdForTopic() < 1 || this.defaultMQPushConsumer.getPullThresholdForTopic() > 6553500) { throw new MQClientException( "pullThresholdForTopic Out of range [1, 6553500]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } } 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); } if (this.defaultMQPushConsumer.getPullThresholdSizeForTopic() != -1) { if (this.defaultMQPushConsumer.getPullThresholdSizeForTopic() < 1 || this.defaultMQPushConsumer.getPullThresholdSizeForTopic() > 102400) { throw new MQClientException( "pullThresholdSizeForTopic Out of range [1, 102400]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } } if (this.defaultMQPushConsumer.getPullInterval() < 0 || this.defaultMQPushConsumer.getPullInterval() > 65535) { throw new MQClientException( "pullInterval Out of range [0, 65535]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() < 1 || this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() > 1024) { throw new MQClientException( "consumeMessageBatchMaxSize Out of range [1, 1024]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getPullBatchSize() < 1 || this.defaultMQPushConsumer.getPullBatchSize() > 1024) { throw new MQClientException( "pullBatchSize Out of range [1, 1024]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } } }
DefaultMQPushConsumerImpl implements MQConsumerInner { private void checkConfig() throws MQClientException { Validators.checkGroup(this.defaultMQPushConsumer.getConsumerGroup()); if (null == this.defaultMQPushConsumer.getConsumerGroup()) { throw new MQClientException( "consumerGroup is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumerGroup().equals(MixAll.DEFAULT_CONSUMER_GROUP)) { throw new MQClientException( "consumerGroup can not equal " + MixAll.DEFAULT_CONSUMER_GROUP + ", please specify another one." + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getMessageModel()) { throw new MQClientException( "messageModel is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getConsumeFromWhere()) { throw new MQClientException( "consumeFromWhere is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } Date dt = UtilAll.parseDate(this.defaultMQPushConsumer.getConsumeTimestamp(), UtilAll.YYYYMMDDHHMMSS); if (null == dt) { throw new MQClientException( "consumeTimestamp is invalid, the valid format is yyyyMMddHHmmss,but received " + this.defaultMQPushConsumer.getConsumeTimestamp() + " " + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getAllocateMessageQueueStrategy()) { throw new MQClientException( "allocateMessageQueueStrategy is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getSubscription()) { throw new MQClientException( "subscription is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getMessageListener()) { throw new MQClientException( "messageListener is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } boolean orderly = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerOrderly; boolean concurrently = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerConcurrently; if (!orderly && !concurrently) { throw new MQClientException( "messageListener must be instanceof MessageListenerOrderly or MessageListenerConcurrently" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeThreadMin() < 1 || this.defaultMQPushConsumer.getConsumeThreadMin() > 1000) { throw new MQClientException( "consumeThreadMin Out of range [1, 1000]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeThreadMax() < 1 || this.defaultMQPushConsumer.getConsumeThreadMax() > 1000) { throw new MQClientException( "consumeThreadMax Out of range [1, 1000]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeThreadMin() > this.defaultMQPushConsumer.getConsumeThreadMax()) { throw new MQClientException( "consumeThreadMin (" + this.defaultMQPushConsumer.getConsumeThreadMin() + ") " + "is larger than consumeThreadMax (" + this.defaultMQPushConsumer.getConsumeThreadMax() + ")", null); } if (this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan() < 1 || this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan() > 65535) { throw new MQClientException( "consumeConcurrentlyMaxSpan Out of range [1, 65535]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } 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); } if (this.defaultMQPushConsumer.getPullThresholdForTopic() != -1) { if (this.defaultMQPushConsumer.getPullThresholdForTopic() < 1 || this.defaultMQPushConsumer.getPullThresholdForTopic() > 6553500) { throw new MQClientException( "pullThresholdForTopic Out of range [1, 6553500]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } } 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); } if (this.defaultMQPushConsumer.getPullThresholdSizeForTopic() != -1) { if (this.defaultMQPushConsumer.getPullThresholdSizeForTopic() < 1 || this.defaultMQPushConsumer.getPullThresholdSizeForTopic() > 102400) { throw new MQClientException( "pullThresholdSizeForTopic Out of range [1, 102400]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } } if (this.defaultMQPushConsumer.getPullInterval() < 0 || this.defaultMQPushConsumer.getPullInterval() > 65535) { throw new MQClientException( "pullInterval Out of range [0, 65535]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() < 1 || this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() > 1024) { throw new MQClientException( "consumeMessageBatchMaxSize Out of range [1, 1024]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getPullBatchSize() < 1 || this.defaultMQPushConsumer.getPullBatchSize() > 1024) { throw new MQClientException( "pullBatchSize Out of range [1, 1024]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } } DefaultMQPushConsumerImpl(DefaultMQPushConsumer defaultMQPushConsumer, RPCHook rpcHook); }
DefaultMQPushConsumerImpl implements MQConsumerInner { private void checkConfig() throws MQClientException { Validators.checkGroup(this.defaultMQPushConsumer.getConsumerGroup()); if (null == this.defaultMQPushConsumer.getConsumerGroup()) { throw new MQClientException( "consumerGroup is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumerGroup().equals(MixAll.DEFAULT_CONSUMER_GROUP)) { throw new MQClientException( "consumerGroup can not equal " + MixAll.DEFAULT_CONSUMER_GROUP + ", please specify another one." + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getMessageModel()) { throw new MQClientException( "messageModel is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getConsumeFromWhere()) { throw new MQClientException( "consumeFromWhere is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } Date dt = UtilAll.parseDate(this.defaultMQPushConsumer.getConsumeTimestamp(), UtilAll.YYYYMMDDHHMMSS); if (null == dt) { throw new MQClientException( "consumeTimestamp is invalid, the valid format is yyyyMMddHHmmss,but received " + this.defaultMQPushConsumer.getConsumeTimestamp() + " " + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getAllocateMessageQueueStrategy()) { throw new MQClientException( "allocateMessageQueueStrategy is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getSubscription()) { throw new MQClientException( "subscription is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getMessageListener()) { throw new MQClientException( "messageListener is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } boolean orderly = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerOrderly; boolean concurrently = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerConcurrently; if (!orderly && !concurrently) { throw new MQClientException( "messageListener must be instanceof MessageListenerOrderly or MessageListenerConcurrently" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeThreadMin() < 1 || this.defaultMQPushConsumer.getConsumeThreadMin() > 1000) { throw new MQClientException( "consumeThreadMin Out of range [1, 1000]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeThreadMax() < 1 || this.defaultMQPushConsumer.getConsumeThreadMax() > 1000) { throw new MQClientException( "consumeThreadMax Out of range [1, 1000]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeThreadMin() > this.defaultMQPushConsumer.getConsumeThreadMax()) { throw new MQClientException( "consumeThreadMin (" + this.defaultMQPushConsumer.getConsumeThreadMin() + ") " + "is larger than consumeThreadMax (" + this.defaultMQPushConsumer.getConsumeThreadMax() + ")", null); } if (this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan() < 1 || this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan() > 65535) { throw new MQClientException( "consumeConcurrentlyMaxSpan Out of range [1, 65535]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } 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); } if (this.defaultMQPushConsumer.getPullThresholdForTopic() != -1) { if (this.defaultMQPushConsumer.getPullThresholdForTopic() < 1 || this.defaultMQPushConsumer.getPullThresholdForTopic() > 6553500) { throw new MQClientException( "pullThresholdForTopic Out of range [1, 6553500]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } } 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); } if (this.defaultMQPushConsumer.getPullThresholdSizeForTopic() != -1) { if (this.defaultMQPushConsumer.getPullThresholdSizeForTopic() < 1 || this.defaultMQPushConsumer.getPullThresholdSizeForTopic() > 102400) { throw new MQClientException( "pullThresholdSizeForTopic Out of range [1, 102400]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } } if (this.defaultMQPushConsumer.getPullInterval() < 0 || this.defaultMQPushConsumer.getPullInterval() > 65535) { throw new MQClientException( "pullInterval Out of range [0, 65535]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() < 1 || this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() > 1024) { throw new MQClientException( "consumeMessageBatchMaxSize Out of range [1, 1024]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getPullBatchSize() < 1 || this.defaultMQPushConsumer.getPullBatchSize() > 1024) { throw new MQClientException( "pullBatchSize Out of range [1, 1024]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } } DefaultMQPushConsumerImpl(DefaultMQPushConsumer defaultMQPushConsumer, RPCHook rpcHook); void registerFilterMessageHook(final FilterMessageHook hook); boolean hasHook(); void registerConsumeMessageHook(final ConsumeMessageHook hook); void executeHookBefore(final ConsumeMessageContext context); void executeHookAfter(final ConsumeMessageContext context); void createTopic(String key, String newTopic, int queueNum); void createTopic(String key, String newTopic, int queueNum, int topicSysFlag); Set<MessageQueue> fetchSubscribeMessageQueues(String topic); Set<MessageQueue> parseSubscribeMessageQueues(Set<MessageQueue> messageQueueList); DefaultMQPushConsumer getDefaultMQPushConsumer(); long earliestMsgStoreTime(MessageQueue mq); long maxOffset(MessageQueue mq); long minOffset(MessageQueue mq); OffsetStore getOffsetStore(); void setOffsetStore(OffsetStore offsetStore); void pullMessage(final PullRequest pullRequest); boolean isPause(); void setPause(boolean pause); ConsumerStatsManager getConsumerStatsManager(); void executePullRequestImmediately(final PullRequest pullRequest); void executeTaskLater(final Runnable r, final long timeDelay); QueryResult queryMessage(String topic, String key, int maxNum, long begin, long end); MessageExt queryMessageByUniqKey(String topic, String uniqKey); void registerMessageListener(MessageListener messageListener); void resume(); void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName); void shutdown(); synchronized void shutdown(long awaitTerminateMillis); synchronized void start(); MessageListener getMessageListenerInner(); ConcurrentMap<String, SubscriptionData> getSubscriptionInner(); void subscribe(String topic, String subExpression); void subscribe(String topic, String fullClassName, String filterClassSource); void subscribe(final String topic, final MessageSelector messageSelector); void suspend(); void unsubscribe(String topic); void updateConsumeOffset(MessageQueue mq, long offset); void updateCorePoolSize(int corePoolSize); MessageExt viewMessage(String msgId); RebalanceImpl getRebalanceImpl(); boolean isConsumeOrderly(); void setConsumeOrderly(boolean consumeOrderly); void resetOffsetByTimeStamp(long timeStamp); long searchOffset(MessageQueue mq, long timestamp); @Override String groupName(); @Override MessageModel messageModel(); @Override ConsumeType consumeType(); @Override ConsumeFromWhere consumeFromWhere(); @Override Set<SubscriptionData> subscriptions(); @Override void doRebalance(); @Override void persistConsumerOffset(); @Override void updateTopicSubscribeInfo(String topic, Set<MessageQueue> info); @Override boolean isSubscribeTopicNeedUpdate(String topic); @Override boolean isUnitMode(); @Override ConsumerRunningInfo consumerRunningInfo(); MQClientInstance getmQClientFactory(); void setmQClientFactory(MQClientInstance mQClientFactory); ServiceState getServiceState(); @Deprecated synchronized void setServiceState(ServiceState serviceState); void adjustThreadPool(); List<QueueTimeSpan> queryConsumeTimeSpan(final String topic); void resetRetryAndNamespace(final List<MessageExt> msgs, String consumerGroup); ConsumeMessageService getConsumeMessageService(); void setConsumeMessageService(ConsumeMessageService consumeMessageService); void setPullTimeDelayMillsWhenException(long pullTimeDelayMillsWhenException); }
DefaultMQPushConsumerImpl implements MQConsumerInner { private void checkConfig() throws MQClientException { Validators.checkGroup(this.defaultMQPushConsumer.getConsumerGroup()); if (null == this.defaultMQPushConsumer.getConsumerGroup()) { throw new MQClientException( "consumerGroup is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumerGroup().equals(MixAll.DEFAULT_CONSUMER_GROUP)) { throw new MQClientException( "consumerGroup can not equal " + MixAll.DEFAULT_CONSUMER_GROUP + ", please specify another one." + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getMessageModel()) { throw new MQClientException( "messageModel is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getConsumeFromWhere()) { throw new MQClientException( "consumeFromWhere is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } Date dt = UtilAll.parseDate(this.defaultMQPushConsumer.getConsumeTimestamp(), UtilAll.YYYYMMDDHHMMSS); if (null == dt) { throw new MQClientException( "consumeTimestamp is invalid, the valid format is yyyyMMddHHmmss,but received " + this.defaultMQPushConsumer.getConsumeTimestamp() + " " + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getAllocateMessageQueueStrategy()) { throw new MQClientException( "allocateMessageQueueStrategy is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getSubscription()) { throw new MQClientException( "subscription is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (null == this.defaultMQPushConsumer.getMessageListener()) { throw new MQClientException( "messageListener is null" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } boolean orderly = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerOrderly; boolean concurrently = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerConcurrently; if (!orderly && !concurrently) { throw new MQClientException( "messageListener must be instanceof MessageListenerOrderly or MessageListenerConcurrently" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeThreadMin() < 1 || this.defaultMQPushConsumer.getConsumeThreadMin() > 1000) { throw new MQClientException( "consumeThreadMin Out of range [1, 1000]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeThreadMax() < 1 || this.defaultMQPushConsumer.getConsumeThreadMax() > 1000) { throw new MQClientException( "consumeThreadMax Out of range [1, 1000]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeThreadMin() > this.defaultMQPushConsumer.getConsumeThreadMax()) { throw new MQClientException( "consumeThreadMin (" + this.defaultMQPushConsumer.getConsumeThreadMin() + ") " + "is larger than consumeThreadMax (" + this.defaultMQPushConsumer.getConsumeThreadMax() + ")", null); } if (this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan() < 1 || this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan() > 65535) { throw new MQClientException( "consumeConcurrentlyMaxSpan Out of range [1, 65535]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } 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); } if (this.defaultMQPushConsumer.getPullThresholdForTopic() != -1) { if (this.defaultMQPushConsumer.getPullThresholdForTopic() < 1 || this.defaultMQPushConsumer.getPullThresholdForTopic() > 6553500) { throw new MQClientException( "pullThresholdForTopic Out of range [1, 6553500]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } } 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); } if (this.defaultMQPushConsumer.getPullThresholdSizeForTopic() != -1) { if (this.defaultMQPushConsumer.getPullThresholdSizeForTopic() < 1 || this.defaultMQPushConsumer.getPullThresholdSizeForTopic() > 102400) { throw new MQClientException( "pullThresholdSizeForTopic Out of range [1, 102400]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } } if (this.defaultMQPushConsumer.getPullInterval() < 0 || this.defaultMQPushConsumer.getPullInterval() > 65535) { throw new MQClientException( "pullInterval Out of range [0, 65535]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() < 1 || this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() > 1024) { throw new MQClientException( "consumeMessageBatchMaxSize Out of range [1, 1024]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } if (this.defaultMQPushConsumer.getPullBatchSize() < 1 || this.defaultMQPushConsumer.getPullBatchSize() > 1024) { throw new MQClientException( "pullBatchSize Out of range [1, 1024]" + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), null); } } DefaultMQPushConsumerImpl(DefaultMQPushConsumer defaultMQPushConsumer, RPCHook rpcHook); void registerFilterMessageHook(final FilterMessageHook hook); boolean hasHook(); void registerConsumeMessageHook(final ConsumeMessageHook hook); void executeHookBefore(final ConsumeMessageContext context); void executeHookAfter(final ConsumeMessageContext context); void createTopic(String key, String newTopic, int queueNum); void createTopic(String key, String newTopic, int queueNum, int topicSysFlag); Set<MessageQueue> fetchSubscribeMessageQueues(String topic); Set<MessageQueue> parseSubscribeMessageQueues(Set<MessageQueue> messageQueueList); DefaultMQPushConsumer getDefaultMQPushConsumer(); long earliestMsgStoreTime(MessageQueue mq); long maxOffset(MessageQueue mq); long minOffset(MessageQueue mq); OffsetStore getOffsetStore(); void setOffsetStore(OffsetStore offsetStore); void pullMessage(final PullRequest pullRequest); boolean isPause(); void setPause(boolean pause); ConsumerStatsManager getConsumerStatsManager(); void executePullRequestImmediately(final PullRequest pullRequest); void executeTaskLater(final Runnable r, final long timeDelay); QueryResult queryMessage(String topic, String key, int maxNum, long begin, long end); MessageExt queryMessageByUniqKey(String topic, String uniqKey); void registerMessageListener(MessageListener messageListener); void resume(); void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName); void shutdown(); synchronized void shutdown(long awaitTerminateMillis); synchronized void start(); MessageListener getMessageListenerInner(); ConcurrentMap<String, SubscriptionData> getSubscriptionInner(); void subscribe(String topic, String subExpression); void subscribe(String topic, String fullClassName, String filterClassSource); void subscribe(final String topic, final MessageSelector messageSelector); void suspend(); void unsubscribe(String topic); void updateConsumeOffset(MessageQueue mq, long offset); void updateCorePoolSize(int corePoolSize); MessageExt viewMessage(String msgId); RebalanceImpl getRebalanceImpl(); boolean isConsumeOrderly(); void setConsumeOrderly(boolean consumeOrderly); void resetOffsetByTimeStamp(long timeStamp); long searchOffset(MessageQueue mq, long timestamp); @Override String groupName(); @Override MessageModel messageModel(); @Override ConsumeType consumeType(); @Override ConsumeFromWhere consumeFromWhere(); @Override Set<SubscriptionData> subscriptions(); @Override void doRebalance(); @Override void persistConsumerOffset(); @Override void updateTopicSubscribeInfo(String topic, Set<MessageQueue> info); @Override boolean isSubscribeTopicNeedUpdate(String topic); @Override boolean isUnitMode(); @Override ConsumerRunningInfo consumerRunningInfo(); MQClientInstance getmQClientFactory(); void setmQClientFactory(MQClientInstance mQClientFactory); ServiceState getServiceState(); @Deprecated synchronized void setServiceState(ServiceState serviceState); void adjustThreadPool(); List<QueueTimeSpan> queryConsumeTimeSpan(final String topic); void resetRetryAndNamespace(final List<MessageExt> msgs, String consumerGroup); ConsumeMessageService getConsumeMessageService(); void setConsumeMessageService(ConsumeMessageService consumeMessageService); void setPullTimeDelayMillsWhenException(long pullTimeDelayMillsWhenException);  }