+ * support other message selection, such as {@link org.apache.rocketmq.common.filter.ExpressionType#SQL92} + *
+ * + * @param mq from which message queue + * @param selector message selector({@link MessageSelector}), can be null. + * @param offset from where to pull + * @param maxNums max pulling numbers + * @return The resulting {@code PullRequest} + */ + PullResult pull(final MessageQueue mq, final MessageSelector selector, final long offset, + final int maxNums) throws MQClientException, RemotingException, MQBrokerException, + InterruptedException; + + /** + * Pulling the messages in the specified timeout + *+ * support other message selection, such as {@link org.apache.rocketmq.common.filter.ExpressionType#SQL92} + *
+ * + * @param mq from which message queue + * @param selector message selector({@link MessageSelector}), can be null. + * @param offset from where to pull + * @param maxNums max pulling numbers + * @param timeout Pulling the messages in the specified timeout + * @return The resulting {@code PullRequest} + */ + PullResult pull(final MessageQueue mq, final MessageSelector selector, final long offset, + final int maxNums, final long timeout) throws MQClientException, RemotingException, MQBrokerException, + InterruptedException; + /** * Pulling the messages in a async. way */ @@ -80,6 +113,20 @@ void pull(final MessageQueue mq, final String subExpression, final long offset, final PullCallback pullCallback, long timeout) throws MQClientException, RemotingException, InterruptedException; + /** + * Pulling the messages in a async. way. Support message selection + */ + void pull(final MessageQueue mq, final MessageSelector selector, final long offset, final int maxNums, + final PullCallback pullCallback) throws MQClientException, RemotingException, + InterruptedException; + + /** + * Pulling the messages in a async. way. Support message selection + */ + void pull(final MessageQueue mq, final MessageSelector selector, final long offset, final int maxNums, + final PullCallback pullCallback, long timeout) throws MQClientException, RemotingException, + InterruptedException; + /** * Pulling the messages,if no message arrival,blocking some time * diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java index 420d89b2fd0..39c43d592d7 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java @@ -26,6 +26,7 @@ import org.apache.rocketmq.client.QueryResult; import org.apache.rocketmq.client.Validators; import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer; +import org.apache.rocketmq.client.consumer.MessageSelector; import org.apache.rocketmq.client.consumer.PullCallback; import org.apache.rocketmq.client.consumer.PullResult; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; @@ -46,6 +47,7 @@ import org.apache.rocketmq.common.ServiceState; import org.apache.rocketmq.common.UtilAll; import org.apache.rocketmq.common.consumer.ConsumeFromWhere; +import org.apache.rocketmq.common.filter.ExpressionType; import org.apache.rocketmq.common.filter.FilterAPI; import org.apache.rocketmq.common.help.FAQUrl; import org.apache.rocketmq.logging.InternalLogger; @@ -158,17 +160,58 @@ public PullResult pull(MessageQueue mq, String subExpression, long offset, int m public PullResult pull(MessageQueue mq, String subExpression, long offset, int maxNums, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { - return this.pullSyncImpl(mq, subExpression, offset, maxNums, false, timeout); + SubscriptionData subscriptionData = getSubscriptionData(mq, subExpression); + return this.pullSyncImpl(mq, subscriptionData, offset, maxNums, false, timeout); } - private PullResult pullSyncImpl(MessageQueue mq, String subExpression, long offset, int maxNums, boolean block, + public PullResult pull(MessageQueue mq, MessageSelector messageSelector, long offset, int maxNums) + throws MQClientException, RemotingException, MQBrokerException, InterruptedException { + return pull(mq, messageSelector, offset, maxNums, this.defaultMQPullConsumer.getConsumerPullTimeoutMillis()); + } + + public PullResult pull(MessageQueue mq, MessageSelector messageSelector, long offset, int maxNums, long timeout) + throws MQClientException, RemotingException, MQBrokerException, InterruptedException { + SubscriptionData subscriptionData = getSubscriptionData(mq, messageSelector); + return this.pullSyncImpl(mq, subscriptionData, offset, maxNums, false, timeout); + } + + private SubscriptionData getSubscriptionData(MessageQueue mq, String subExpression) + throws MQClientException { + + if (null == mq) { + throw new MQClientException("mq is null", null); + } + + try { + return FilterAPI.buildSubscriptionData(this.defaultMQPullConsumer.getConsumerGroup(), + mq.getTopic(), subExpression); + } catch (Exception e) { + throw new MQClientException("parse subscription error", e); + } + } + + private SubscriptionData getSubscriptionData(MessageQueue mq, MessageSelector messageSelector) + throws MQClientException { + + if (null == mq) { + throw new MQClientException("mq is null", null); + } + + try { + return FilterAPI.build(mq.getTopic(), + messageSelector.getExpression(), messageSelector.getExpressionType()); + } catch (Exception e) { + throw new MQClientException("parse subscription error", e); + } + } + + private PullResult pullSyncImpl(MessageQueue mq, SubscriptionData subscriptionData, long offset, int maxNums, boolean block, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { this.makeSureStateOK(); if (null == mq) { throw new MQClientException("mq is null", null); - } if (offset < 0) { @@ -183,20 +226,14 @@ private PullResult pullSyncImpl(MessageQueue mq, String subExpression, long offs int sysFlag = PullSysFlag.buildSysFlag(false, block, true, false); - SubscriptionData subscriptionData; - try { - subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPullConsumer.getConsumerGroup(), - mq.getTopic(), subExpression); - } catch (Exception e) { - throw new MQClientException("parse subscription error", e); - } - long timeoutMillis = block ? this.defaultMQPullConsumer.getConsumerTimeoutMillisWhenSuspend() : timeout; + boolean isTagType = ExpressionType.isTagType(subscriptionData.getExpressionType()); PullResult pullResult = this.pullAPIWrapper.pullKernelImpl( mq, subscriptionData.getSubString(), - 0L, + subscriptionData.getExpressionType(), + isTagType ? 0L : subscriptionData.getSubVersion(), offset, maxNums, sysFlag, @@ -369,12 +406,27 @@ public void pull(MessageQueue mq, String subExpression, long offset, int maxNums public void pull(MessageQueue mq, String subExpression, long offset, int maxNums, PullCallback pullCallback, long timeout) throws MQClientException, RemotingException, InterruptedException { - this.pullAsyncImpl(mq, subExpression, offset, maxNums, pullCallback, false, timeout); + SubscriptionData subscriptionData = getSubscriptionData(mq, subExpression); + this.pullAsyncImpl(mq, subscriptionData, offset, maxNums, pullCallback, false, timeout); + } + + public void pull(MessageQueue mq, MessageSelector messageSelector, long offset, int maxNums, + PullCallback pullCallback) + throws MQClientException, RemotingException, InterruptedException { + pull(mq, messageSelector, offset, maxNums, pullCallback, this.defaultMQPullConsumer.getConsumerPullTimeoutMillis()); + } + + public void pull(MessageQueue mq, MessageSelector messageSelector, long offset, int maxNums, + PullCallback pullCallback, + long timeout) + throws MQClientException, RemotingException, InterruptedException { + SubscriptionData subscriptionData = getSubscriptionData(mq, messageSelector); + this.pullAsyncImpl(mq, subscriptionData, offset, maxNums, pullCallback, false, timeout); } private void pullAsyncImpl( final MessageQueue mq, - final String subExpression, + final SubscriptionData subscriptionData, final long offset, final int maxNums, final PullCallback pullCallback, @@ -403,20 +455,14 @@ private void pullAsyncImpl( try { int sysFlag = PullSysFlag.buildSysFlag(false, block, true, false); - final SubscriptionData subscriptionData; - try { - subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPullConsumer.getConsumerGroup(), - mq.getTopic(), subExpression); - } catch (Exception e) { - throw new MQClientException("parse subscription error", e); - } - long timeoutMillis = block ? this.defaultMQPullConsumer.getConsumerTimeoutMillisWhenSuspend() : timeout; + boolean isTagType = ExpressionType.isTagType(subscriptionData.getExpressionType()); this.pullAPIWrapper.pullKernelImpl( mq, subscriptionData.getSubString(), - 0L, + subscriptionData.getExpressionType(), + isTagType ? 0L : subscriptionData.getSubVersion(), offset, maxNums, sysFlag, @@ -444,7 +490,8 @@ public void onException(Throwable e) { public PullResult pullBlockIfNotFound(MessageQueue mq, String subExpression, long offset, int maxNums) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { - return this.pullSyncImpl(mq, subExpression, offset, maxNums, true, this.getDefaultMQPullConsumer().getConsumerPullTimeoutMillis()); + SubscriptionData subscriptionData = getSubscriptionData(mq, subExpression); + return this.pullSyncImpl(mq, subscriptionData, offset, maxNums, true, this.getDefaultMQPullConsumer().getConsumerPullTimeoutMillis()); } public DefaultMQPullConsumer getDefaultMQPullConsumer() { @@ -454,7 +501,8 @@ public DefaultMQPullConsumer getDefaultMQPullConsumer() { public void pullBlockIfNotFound(MessageQueue mq, String subExpression, long offset, int maxNums, PullCallback pullCallback) throws MQClientException, RemotingException, InterruptedException { - this.pullAsyncImpl(mq, subExpression, offset, maxNums, pullCallback, true, + SubscriptionData subscriptionData = getSubscriptionData(mq, subExpression); + this.pullAsyncImpl(mq, subscriptionData, offset, maxNums, pullCallback, true, this.getDefaultMQPullConsumer().getConsumerPullTimeoutMillis()); } diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullAPIWrapper.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullAPIWrapper.java index b650e35e02f..1d2d24fa3b9 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullAPIWrapper.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullAPIWrapper.java @@ -209,34 +209,6 @@ public PullResult pullKernelImpl( throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null); } - public PullResult pullKernelImpl( - final MessageQueue mq, - final String subExpression, - 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 { - return pullKernelImpl( - mq, - subExpression, - ExpressionType.TAG, - subVersion, offset, - maxNums, - sysFlag, - commitOffset, - brokerSuspendMaxTimeMillis, - timeoutMillis, - communicationMode, - pullCallback - ); - } - public long recalculatePullFromWhichNode(final MessageQueue mq) { if (this.isConnectBrokerByUser()) { return this.defaultBrokerId; diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java b/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java index 4b055e0fc2f..88f06ea4ca6 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java @@ -1046,6 +1046,19 @@ public int findBrokerVersion(String brokerName, String brokerAddr) { if (this.brokerVersionTable.get(brokerName).containsKey(brokerAddr)) { return this.brokerVersionTable.get(brokerName).get(brokerAddr); } + } else { + HeartbeatData heartbeatData = prepareHeartbeatData(); + try { + int version = this.mQClientAPIImpl.sendHearbeat(brokerAddr, heartbeatData, 3000); + return version; + } catch (Exception e) { + if (this.isBrokerInNameServer(brokerAddr)) { + log.info("send heart beat to broker[{} {}] failed", brokerName, brokerAddr); + } else { + log.info("send heart beat to broker[{} {}] exception, because the broker not up, forget it", brokerName, + brokerAddr); + } + } } return 0; } diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java index 7ace9d5b07d..90f4f787635 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java @@ -30,8 +30,10 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.rocketmq.client.QueryResult; import org.apache.rocketmq.client.Validators; import org.apache.rocketmq.client.common.ClientErrorCode; @@ -101,6 +103,10 @@ public class DefaultMQProducerImpl implements MQProducerInner { private MQFaultStrategy mqFaultStrategy = new MQFaultStrategy(); + private final BlockingQueue