You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by du...@apache.org on 2019/01/25 02:51:21 UTC
[rocketmq] branch snode updated: Add support for one client
instance with multi consumer group in push module
This is an automated email from the ASF dual-hosted git repository.
duhengforever pushed a commit to branch snode
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/snode by this push:
new c7fdb76 Add support for one client instance with multi consumer group in push module
c7fdb76 is described below
commit c7fdb765ec8099c3844c8d961a384347a7d8390f
Author: duhenglucky <du...@gmail.com>
AuthorDate: Fri Jan 25 10:50:40 2019 +0800
Add support for one client instance with multi consumer group in push module
---
acl/pom.xml | 3 +-
.../apache/rocketmq/broker/BrokerController.java | 4 +-
.../broker/processor/PullMessageProcessor.java | 17 +++----
.../processor/SnodePullMessageProcessor.java | 2 +-
.../client/impl/consumer/RebalancePushImpl.java | 10 ++--
.../flowcontrol/AbstractFlowControlService.java | 2 +-
.../common/protocol/header/PushMessageHeader.java | 11 +++++
.../common/protocol/heartbeat/ConsumerData.java | 11 -----
.../protocol/heartbeat/PushSubscriptionData.java | 18 +++++--
.../protocol/heartbeat/SubscriptionData.java | 16 -------
.../apache/rocketmq/example/filter/Consumer.java | 12 ++---
.../apache/rocketmq/example/filter/Producer.java | 29 ++++++------
.../rocketmq/example/filter/SqlConsumer.java | 28 ++++-------
.../org/apache/rocketmq/snode/client/Client.java | 46 +++++++++---------
.../rocketmq/snode/client/ClientManager.java | 2 +-
.../rocketmq/snode/client/SlowConsumerService.java | 5 +-
.../rocketmq/snode/client/SubscriptionManager.java | 6 ++-
.../snode/client/impl/ClientManagerImpl.java | 16 +++----
.../snode/client/impl/SlowConsumerServiceImpl.java | 27 +++--------
.../snode/client/impl/SubscriptionManagerImpl.java | 55 +++++++++-------------
.../snode/processor/HeartbeatProcessor.java | 34 +++++++------
.../rocketmq/snode/service/AdminService.java | 1 +
.../snode/service/impl/PushServiceImpl.java | 50 +++++++++++++-------
23 files changed, 187 insertions(+), 218 deletions(-)
diff --git a/acl/pom.xml b/acl/pom.xml
index e7d24d4..d592304 100644
--- a/acl/pom.xml
+++ b/acl/pom.xml
@@ -14,6 +14,7 @@
<artifactId>rocketmq-all</artifactId>
<version>4.4.0-SNAPSHOT</version>
</parent>
+
<modelVersion>4.0.0</modelVersion>
<packaging>jar</packaging>
<artifactId>rocketmq-acl</artifactId>
@@ -69,7 +70,7 @@
<scope>test</scope>
</dependency>
<dependency>
- <groupId>org.apache.rocketmq</groupId>
+ <groupId>${project.groupId}</groupId>
<artifactId>rocketmq-remoting</artifactId>
</dependency>
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
index 1218956..4077bf4 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
@@ -513,8 +513,8 @@ public class BrokerController {
*/
this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE, this.pullMessageProcessor, this.pullMessageExecutor);
this.pullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);
- this.remotingServer.registerProcessor(RequestCode.SNODE_PULL_MESSAGE, this.snodePullMessageProcessor, pullMessageExecutor);
- this.snodePullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);
+// this.remotingServer.registerProcessor(RequestCode.SNODE_PULL_MESSAGE, this.snodePullMessageProcessor, pullMessageExecutor);
+// this.snodePullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);
/**
* QueryMessageProcessor
*/
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java
index 60b9ad8..d7f7d4b 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java
@@ -19,7 +19,6 @@ package org.apache.rocketmq.broker.processor;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
-import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.FileRegion;
import java.nio.ByteBuffer;
import java.util.List;
@@ -41,8 +40,6 @@ import org.apache.rocketmq.common.constant.PermName;
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;
-import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.ResponseCode;
@@ -53,11 +50,13 @@ import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.common.protocol.topic.OffsetMovedEvent;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.common.sysflag.PullSysFlag;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.RemotingChannel;
+import org.apache.rocketmq.remoting.RequestProcessor;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.common.RemotingUtil;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
-import org.apache.rocketmq.remoting.RequestProcessor;
import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.netty.RequestTask;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
@@ -78,12 +77,10 @@ public class PullMessageProcessor implements RequestProcessor {
}
@Override
- public RemotingCommand processRequest(RemotingChannel remotingChannel,
+ public RemotingCommand processRequest(final RemotingChannel remotingChannel,
RemotingCommand request) throws RemotingCommandException {
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = (NettyChannelHandlerContextImpl) remotingChannel;
- ChannelHandlerContext ctx = nettyChannelHandlerContext.getChannelHandlerContext();
-
- return this.processRequest(ctx.channel(), request, true);
+ Channel channel = ((NettyChannelHandlerContextImpl)remotingChannel).getChannelHandlerContext().channel();
+ return this.processRequest(channel, request, true);
}
@Override
@@ -100,7 +97,7 @@ public class PullMessageProcessor implements RequestProcessor {
response.setOpaque(request.getOpaque());
- log.info("receive PullMessage request command, {}", request);
+ log.debug("receive PullMessage request command, {}", request);
if (!PermName.isReadable(this.brokerController.getBrokerConfig().getBrokerPermission())) {
response.setCode(ResponseCode.NO_PERMISSION);
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/SnodePullMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/SnodePullMessageProcessor.java
index c08ebcd..74a0176 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/SnodePullMessageProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/SnodePullMessageProcessor.java
@@ -115,7 +115,7 @@ public class SnodePullMessageProcessor implements RequestProcessor {
log.warn("Parse the consumer's subscription[{}] failed, group: {}", requestHeader.getSubscription(),
requestHeader.getConsumerGroup());
response.setCode(ResponseCode.SUBSCRIPTION_PARSE_FAILED);
- response.setRemark("parse the consumer's subscription failed");
+ response.setRemark(e.getMessage());
return response;
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalancePushImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalancePushImpl.java
index 79e1a64..b772b84 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalancePushImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalancePushImpl.java
@@ -31,6 +31,7 @@ import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
+import org.apache.rocketmq.common.protocol.heartbeat.PushSubscriptionData;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
public class RebalancePushImpl extends RebalanceImpl {
@@ -54,16 +55,17 @@ public class RebalancePushImpl extends RebalanceImpl {
* When rebalance result changed, should update subscription's version to notify broker.
* Fix: inconsistency subscription may lead to consumer miss messages.
*/
- SubscriptionData subscriptionData = this.subscriptionInner.get(topic);
+ SubscriptionData sub = this.subscriptionInner.get(topic);
+ PushSubscriptionData subscriptionData = (PushSubscriptionData) sub;
long newVersion = System.currentTimeMillis();
log.info("{} Rebalance changed, also update version: {}, {}", topic, subscriptionData.getSubVersion(), newVersion);
subscriptionData.setSubVersion(newVersion);
- Set<Integer> queueIdSet = new HashSet<Integer>();
+ Set<MessageQueue> queueIdSet = new HashSet<MessageQueue>();
for (MessageQueue messageQueue : mqAll) {
- queueIdSet.add(messageQueue.getQueueId());
+ queueIdSet.add(messageQueue);
}
- subscriptionData.setQueueIdSet(queueIdSet);
+ subscriptionData.setMessageQueueSet(queueIdSet);
int currentQueueCount = this.processQueueTable.size();
if (currentQueueCount != 0) {
int pullThresholdForTopic = this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getPullThresholdForTopic();
diff --git a/common/src/main/java/org/apache/rocketmq/common/flowcontrol/AbstractFlowControlService.java b/common/src/main/java/org/apache/rocketmq/common/flowcontrol/AbstractFlowControlService.java
index 3cdfa65..8713a5d 100644
--- a/common/src/main/java/org/apache/rocketmq/common/flowcontrol/AbstractFlowControlService.java
+++ b/common/src/main/java/org/apache/rocketmq/common/flowcontrol/AbstractFlowControlService.java
@@ -131,7 +131,7 @@ public abstract class AbstractFlowControlService implements Interceptor {
}
}
FlowRuleManager.loadRules(sentinelRules);
- log.warn("Load Rules: {}" + FlowRuleManager.getRules());
+ log.warn("Load Rules: {}", FlowRuleManager.getRules());
}
}
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/PushMessageHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/PushMessageHeader.java
index f7288e6..40277eb 100644
--- a/common/src/main/java/org/apache/rocketmq/common/protocol/header/PushMessageHeader.java
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/PushMessageHeader.java
@@ -33,6 +33,9 @@ public class PushMessageHeader implements CommandCustomHeader {
@CFNotNull
private String topic;
+ @CFNotNull
+ private String consumerGroup;
+
@Override
public void checkFields() throws RemotingCommandException {
@@ -69,4 +72,12 @@ public class PushMessageHeader implements CommandCustomHeader {
public void setMessageId(String messageId) {
this.messageId = messageId;
}
+
+ public String getConsumerGroup() {
+ return consumerGroup;
+ }
+
+ public void setConsumerGroup(String consumerGroup) {
+ this.consumerGroup = consumerGroup;
+ }
}
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ConsumerData.java b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ConsumerData.java
index 505eeed..604ce04 100644
--- a/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ConsumerData.java
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ConsumerData.java
@@ -31,7 +31,6 @@ public class ConsumerData {
private ConsumeFromWhere consumeFromWhere;
private Set<SubscriptionData> subscriptionDataSet = new HashSet<SubscriptionData>();
private boolean unitMode;
- private boolean realPushEnable = true;
public String getGroupName() {
return groupName;
@@ -80,15 +79,6 @@ public class ConsumerData {
public void setUnitMode(boolean isUnitMode) {
this.unitMode = isUnitMode;
}
-
- public boolean isRealPushEnable() {
- return realPushEnable;
- }
-
- public void setRealPushEnable(boolean realPushEnable) {
- this.realPushEnable = realPushEnable;
- }
-
@Override
public String toString() {
return "ConsumerData{" +
@@ -98,7 +88,6 @@ public class ConsumerData {
", consumeFromWhere=" + consumeFromWhere +
", subscriptionDataSet=" + subscriptionDataSet +
", unitMode=" + unitMode +
- ", realPushEnable=" + realPushEnable +
'}';
}
}
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/client/SlowConsumerService.java b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/PushSubscriptionData.java
similarity index 60%
copy from snode/src/main/java/org/apache/rocketmq/snode/client/SlowConsumerService.java
copy to common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/PushSubscriptionData.java
index 1fdea74..02245d2 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/client/SlowConsumerService.java
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/PushSubscriptionData.java
@@ -14,12 +14,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.rocketmq.snode.client;
+package org.apache.rocketmq.common.protocol.heartbeat;
-import org.apache.rocketmq.remoting.RemotingChannel;
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.rocketmq.common.message.MessageQueue;
-public interface SlowConsumerService {
+public class PushSubscriptionData extends SubscriptionData {
+ private Set<MessageQueue> messageQueueSet = new HashSet<MessageQueue>();
+
+ public Set<MessageQueue> getMessageQueueSet() {
+ return messageQueueSet;
+ }
- boolean isSlowConsumer(long latestLogicOffset, String topic, int queueId, RemotingChannel remotingChannel,
- String enodeName);
+ public void setMessageQueueSet(Set<MessageQueue> messageQueueSet) {
+ this.messageQueueSet = messageQueueSet;
+ }
}
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/SubscriptionData.java b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/SubscriptionData.java
index c6ae712..4b0a303 100644
--- a/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/SubscriptionData.java
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/SubscriptionData.java
@@ -31,7 +31,6 @@ public class SubscriptionData implements Comparable<SubscriptionData> {
private boolean classFilterMode = false;
private String topic;
private String subString;
- private Set<Integer> queueIdSet = new HashSet<Integer>();
private Set<String> tagsSet = new HashSet<String>();
private Set<Integer> codeSet = new HashSet<Integer>();
private long subVersion = System.currentTimeMillis();
@@ -114,14 +113,6 @@ public class SubscriptionData implements Comparable<SubscriptionData> {
this.expressionType = expressionType;
}
- public Set<Integer> getQueueIdSet() {
- return queueIdSet;
- }
-
- public void setQueueIdSet(Set<Integer> queueIdSet) {
- this.queueIdSet = queueIdSet;
- }
-
@Override
public int hashCode() {
final int prime = 31;
@@ -129,7 +120,6 @@ public class SubscriptionData implements Comparable<SubscriptionData> {
result = prime * result + (classFilterMode ? 1231 : 1237);
result = prime * result + ((codeSet == null) ? 0 : codeSet.hashCode());
result = prime * result + ((subString == null) ? 0 : subString.hashCode());
- result = prime * result + ((queueIdSet == null) ? 0 : queueIdSet.hashCode());
result = prime * result + ((tagsSet == null) ? 0 : tagsSet.hashCode());
result = prime * result + ((topic == null) ? 0 : topic.hashCode());
result = prime * result + ((expressionType == null) ? 0 : expressionType.hashCode());
@@ -164,11 +154,6 @@ public class SubscriptionData implements Comparable<SubscriptionData> {
return false;
} else if (!tagsSet.equals(other.tagsSet))
return false;
- if (queueIdSet == null) {
- if (other.queueIdSet != null)
- return false;
- } else if (!queueIdSet.equals(other.queueIdSet))
- return false;
if (topic == null) {
if (other.topic != null)
return false;
@@ -187,7 +172,6 @@ public class SubscriptionData implements Comparable<SubscriptionData> {
"classFilterMode=" + classFilterMode +
", topic='" + topic + '\'' +
", subString='" + subString + '\'' +
- ", queueIdSet=" + queueIdSet +
", tagsSet=" + tagsSet +
", codeSet=" + codeSet +
", subVersion=" + subVersion +
diff --git a/example/src/main/java/org/apache/rocketmq/example/filter/Consumer.java b/example/src/main/java/org/apache/rocketmq/example/filter/Consumer.java
index bb491ac..8fd50cc 100644
--- a/example/src/main/java/org/apache/rocketmq/example/filter/Consumer.java
+++ b/example/src/main/java/org/apache/rocketmq/example/filter/Consumer.java
@@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.example.filter;
-import java.io.File;
import java.io.IOException;
import java.util.List;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
@@ -24,20 +23,14 @@ import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.message.MessageExt;
public class Consumer {
public static void main(String[] args) throws InterruptedException, MQClientException, IOException {
- DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupNamecc4");
+ DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
- ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
- File classFile = new File(classLoader.getResource("MessageFilterImpl.java").getFile());
-
- String filterCode = MixAll.file2String(classFile);
- consumer.subscribe("TopicTest", "org.apache.rocketmq.example.filter.MessageFilterImpl",
- filterCode);
+ consumer.subscribe("TagFilterTest", "TagA || TagC");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@@ -53,4 +46,5 @@ public class Consumer {
System.out.printf("Consumer Started.%n");
}
+
}
diff --git a/example/src/main/java/org/apache/rocketmq/example/filter/Producer.java b/example/src/main/java/org/apache/rocketmq/example/filter/Producer.java
index 2a0da65..48aac56 100644
--- a/example/src/main/java/org/apache/rocketmq/example/filter/Producer.java
+++ b/example/src/main/java/org/apache/rocketmq/example/filter/Producer.java
@@ -16,31 +16,30 @@
*/
package org.apache.rocketmq.example.filter;
-import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
public class Producer {
- public static void main(String[] args) throws MQClientException, InterruptedException {
- DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
+ public static void main(String[] args) throws Exception {
+ DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
+
producer.start();
- try {
- for (int i = 0; i < 6000000; i++) {
- Message msg = new Message("TopicFilter7",
- "TagA",
- "OrderID001",
- "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
+ String[] tags = new String[] {"TagA", "TagB", "TagC"};
+
+ for (int i = 0; i < 10; i++) {
+ Message msg = new Message("TagFilterTest",
+ tags[i % tags.length],
+ ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
+ );
+ msg.putUserProperty("a", String.valueOf(i));
- msg.putUserProperty("SequenceId", String.valueOf(i));
- SendResult sendResult = producer.send(msg);
- System.out.printf("%s%n", sendResult);
- }
- } catch (Exception e) {
- e.printStackTrace();
+ SendResult sendResult = producer.send(msg);
+ System.out.printf("%s%n", sendResult);
}
+
producer.shutdown();
}
}
diff --git a/example/src/main/java/org/apache/rocketmq/example/filter/SqlConsumer.java b/example/src/main/java/org/apache/rocketmq/example/filter/SqlConsumer.java
index c41c9c1..0093842 100644
--- a/example/src/main/java/org/apache/rocketmq/example/filter/SqlConsumer.java
+++ b/example/src/main/java/org/apache/rocketmq/example/filter/SqlConsumer.java
@@ -17,28 +17,23 @@
package org.apache.rocketmq.example.filter;
+import java.util.List;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
-import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
-import java.util.List;
-
public class SqlConsumer {
- public static void main(String[] args) {
- DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
- try {
- consumer.subscribe("TopicTest",
- MessageSelector.bySql("(TAGS is not null and TAGS in ('TagA', 'TagB'))" +
- "and (a is not null and a between 0 3)"));
- } catch (MQClientException e) {
- e.printStackTrace();
- return;
- }
+ public static void main(String[] args) throws Exception{
+ DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
+
+ // Don't forget to set enablePropertyFilter=true in broker
+ consumer.subscribe("SqlFilterTest",
+ MessageSelector.bySql("(TAGS is not null and TAGS in ('TagA', 'TagB'))" +
+ "and (a is not null and a between 0 and 3)"));
consumer.registerMessageListener(new MessageListenerConcurrently() {
@@ -50,12 +45,7 @@ public class SqlConsumer {
}
});
- try {
- consumer.start();
- } catch (MQClientException e) {
- e.printStackTrace();
- return;
- }
+ consumer.start();
System.out.printf("Consumer Started.%n");
}
}
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/client/Client.java b/snode/src/main/java/org/apache/rocketmq/snode/client/Client.java
index 2376b70..4130f6b 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/client/Client.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/client/Client.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.snode.client;
import java.util.Objects;
+import java.util.Set;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.serialize.LanguageCode;
import org.apache.rocketmq.snode.client.impl.ClientRole;
@@ -24,10 +25,10 @@ import org.apache.rocketmq.snode.client.impl.ClientRole;
public class Client {
private ClientRole clientRole;
- private String groupId;
-
private String clientId;
+ private Set<String> groups;
+
private RemotingChannel remotingChannel;
private int heartbeatInterval;
@@ -46,32 +47,23 @@ public class Client {
this.clientRole = clientRole;
}
- @Override
- public boolean equals(Object o) {
+ @Override public boolean equals(Object o) {
if (this == o)
return true;
- if (o == null || getClass() != o.getClass())
+ if (!(o instanceof Client))
return false;
Client client = (Client) o;
- return version == client.version &&
- clientRole == client.clientRole &&
- Objects.equals(groupId, client.groupId) &&
- Objects.equals(clientId, client.clientId) &&
- Objects.equals(remotingChannel, client.remotingChannel) &&
- language == client.language;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(clientRole, groupId, clientId, remotingChannel, version, language);
- }
-
- public String getGroupId() {
- return groupId;
+ return
+ version == client.version &&
+ clientRole == client.clientRole &&
+ Objects.equals(clientId, client.clientId) &&
+ Objects.equals(groups, client.groups) &&
+ Objects.equals(remotingChannel, client.remotingChannel) &&
+ language == client.language;
}
- public void setGroupId(String groupId) {
- this.groupId = groupId;
+ @Override public int hashCode() {
+ return Objects.hash(clientRole, clientId, groups, remotingChannel, heartbeatInterval, lastUpdateTimestamp, version, language);
}
public RemotingChannel getRemotingChannel() {
@@ -122,11 +114,19 @@ public class Client {
this.language = language;
}
+ public Set<String> getGroups() {
+ return groups;
+ }
+
+ public void setGroups(Set<String> groups) {
+ this.groups = groups;
+ }
+
@Override public String toString() {
return "Client{" +
"clientRole=" + clientRole +
- ", groupId='" + groupId + '\'' +
", clientId='" + clientId + '\'' +
+ ", groups=" + groups +
", remotingChannel=" + remotingChannel +
", heartbeatInterval=" + heartbeatInterval +
", lastUpdateTimestamp=" + lastUpdateTimestamp +
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/client/ClientManager.java b/snode/src/main/java/org/apache/rocketmq/snode/client/ClientManager.java
index dead603..b6fda0a 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/client/ClientManager.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/client/ClientManager.java
@@ -20,7 +20,7 @@ import java.util.List;
import org.apache.rocketmq.remoting.RemotingChannel;
public interface ClientManager {
- boolean register(Client client);
+ boolean register(String groupId, Client client);
void unRegister(String groupId, RemotingChannel remotingChannel);
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/client/SlowConsumerService.java b/snode/src/main/java/org/apache/rocketmq/snode/client/SlowConsumerService.java
index 1fdea74..369b245 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/client/SlowConsumerService.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/client/SlowConsumerService.java
@@ -16,10 +16,7 @@
*/
package org.apache.rocketmq.snode.client;
-import org.apache.rocketmq.remoting.RemotingChannel;
-
public interface SlowConsumerService {
- boolean isSlowConsumer(long latestLogicOffset, String topic, int queueId, RemotingChannel remotingChannel,
- String enodeName);
+ boolean isSlowConsumer(long latestLogicOffset, String topic, int queueId, String consumerGroup, String enodeName);
}
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/client/SubscriptionManager.java b/snode/src/main/java/org/apache/rocketmq/snode/client/SubscriptionManager.java
index a40bbb6..a164fc4 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/client/SubscriptionManager.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/client/SubscriptionManager.java
@@ -18,6 +18,7 @@ package org.apache.rocketmq.snode.client;
import java.util.Set;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
@@ -35,9 +36,10 @@ public interface SubscriptionManager {
Subscription getSubscription(String groupId);
- void registerPushSession(Set<SubscriptionData> subscriptionDataSet, RemotingChannel remotingChannel, String groupId);
+ void registerPushSession(Set<SubscriptionData> subscriptionDataSet, RemotingChannel remotingChannel,
+ String groupId);
void removePushSession(RemotingChannel remotingChannel);
- Set<RemotingChannel> getPushableChannel(String topic, Integer queueId);
+ Set<RemotingChannel> getPushableChannel(MessageQueue messageQueue);
}
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/client/impl/ClientManagerImpl.java b/snode/src/main/java/org/apache/rocketmq/snode/client/impl/ClientManagerImpl.java
index 34d7948..3d2bf21 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/client/impl/ClientManagerImpl.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/client/impl/ClientManagerImpl.java
@@ -82,7 +82,7 @@ public abstract class ClientManagerImpl implements ClientManager {
iter.remove();
client.getRemotingChannel().close();
log.warn("SCAN: Remove expired channel from {}ClientTable. channel={}, group={}", client.getClientRole(),
- RemotingHelper.parseChannelRemoteAddr(client.getRemotingChannel().remoteAddress()), client.getGroupId());
+ RemotingHelper.parseChannelRemoteAddr(client.getRemotingChannel().remoteAddress()), group);
if (channelTable.isEmpty()) {
iterator.remove();
log.warn("SCAN: Remove group={} channel from {}ClientTable.", group, client.getClientRole());
@@ -93,13 +93,13 @@ public abstract class ClientManagerImpl implements ClientManager {
}
@Override
- public boolean register(Client client) {
+ public boolean register(String groupId, Client client) {
boolean updated = false;
if (client != null) {
- ConcurrentHashMap<RemotingChannel, Client> channelTable = groupClientTable.get(client.getGroupId());
+ ConcurrentHashMap<RemotingChannel, Client> channelTable = groupClientTable.get(groupId);
if (channelTable == null) {
channelTable = new ConcurrentHashMap();
- ConcurrentHashMap prev = groupClientTable.putIfAbsent(client.getGroupId(), channelTable);
+ ConcurrentHashMap prev = groupClientTable.putIfAbsent(groupId, channelTable);
channelTable = prev != null ? prev : channelTable;
}
@@ -107,14 +107,14 @@ public abstract class ClientManagerImpl implements ClientManager {
if (oldClient == null) {
Client prev = channelTable.put(client.getRemotingChannel(), client);
if (prev != null) {
- log.info("New client connected, group: {} {} {} channel: {}", client.getGroupId(), client.toString());
+ log.info("New client connected, group: {} {} {} channel: {}", groupId, client.toString());
updated = true;
}
oldClient = client;
} else {
if (!oldClient.getClientId().equals(client.getClientId())) {
log.error("[BUG] client channel exist in snode, but clientId not equal. GROUP: {} OLD: {} NEW: {} ",
- client.getGroupId(),
+ groupId,
oldClient.toString(),
channelTable.toString());
channelTable.put(client.getRemotingChannel(), client);
@@ -122,8 +122,8 @@ public abstract class ClientManagerImpl implements ClientManager {
}
oldClient.setLastUpdateTimestamp(System.currentTimeMillis());
}
- log.debug("Register client role: {}, group: {}, last: {}", client.getClientRole(), client.getGroupId(), client.getLastUpdateTimestamp());
- onRegister(client.getGroupId(), client.getRemotingChannel());
+ log.debug("Register client role: {}, group: {}, last: {}", client.getClientRole(), groupId, client.getLastUpdateTimestamp());
+ onRegister(groupId, client.getRemotingChannel());
return updated;
}
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/client/impl/SlowConsumerServiceImpl.java b/snode/src/main/java/org/apache/rocketmq/snode/client/impl/SlowConsumerServiceImpl.java
index acf6d50..f7485c2 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/client/impl/SlowConsumerServiceImpl.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/client/impl/SlowConsumerServiceImpl.java
@@ -16,17 +16,11 @@
*/
package org.apache.rocketmq.snode.client.impl;
-import io.netty.channel.Channel;
-import io.netty.util.Attribute;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
-import org.apache.rocketmq.remoting.RemotingChannel;
-import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
import org.apache.rocketmq.snode.SnodeController;
-import org.apache.rocketmq.snode.client.Client;
import org.apache.rocketmq.snode.client.SlowConsumerService;
-import org.apache.rocketmq.snode.constant.SnodeConstant;
public class SlowConsumerServiceImpl implements SlowConsumerService {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.SNODE_LOGGER_NAME);
@@ -39,22 +33,13 @@ public class SlowConsumerServiceImpl implements SlowConsumerService {
@Override
public boolean isSlowConsumer(long latestLogicOffset, String topic, int queueId,
- RemotingChannel remotingChannel, String enodeName) {
- Client client = null;
- if (remotingChannel instanceof NettyChannelImpl) {
- Channel channel = ((NettyChannelImpl) remotingChannel).getChannel();
- Attribute<Client> clientAttribute = channel.attr(SnodeConstant.NETTY_CLIENT_ATTRIBUTE_KEY);
- if (clientAttribute != null) {
- client = clientAttribute.get();
- }
- }
- if (client != null) {
- long ackedOffset = this.snodeController.getConsumerOffsetManager().queryOffset(enodeName, client.getGroupId(), topic, queueId);
- if (latestLogicOffset - ackedOffset > snodeController.getSnodeConfig().getSlowConsumerThreshold()) {
- log.warn("[SlowConsumer] group: {}, lastAckedOffset:{} nowOffset:{} ", client.getGroupId(), ackedOffset, latestLogicOffset);
- return true;
- }
+ String consumerGroup, String enodeName) {
+ long ackedOffset = this.snodeController.getConsumerOffsetManager().queryOffset(enodeName, consumerGroup, topic, queueId);
+ if (latestLogicOffset - ackedOffset > snodeController.getSnodeConfig().getSlowConsumerThreshold()) {
+ log.warn("[SlowConsumer] group: {}, lastAckedOffset:{} nowOffset:{} ", consumerGroup, ackedOffset, latestLogicOffset);
+ return true;
}
+
return false;
}
}
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/client/impl/SubscriptionManagerImpl.java b/snode/src/main/java/org/apache/rocketmq/snode/client/impl/SubscriptionManagerImpl.java
index 9011c5f..729f282 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/client/impl/SubscriptionManagerImpl.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/client/impl/SubscriptionManagerImpl.java
@@ -23,8 +23,10 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
+import org.apache.rocketmq.common.protocol.heartbeat.PushSubscriptionData;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
@@ -36,38 +38,27 @@ public class SubscriptionManagerImpl implements SubscriptionManager {
private ConcurrentHashMap<String/*Consumer group*/, Subscription> groupSubscriptionTable = new ConcurrentHashMap<>(1024);
- private ConcurrentHashMap<String/*Topic#QueueId*/, Set<RemotingChannel>> pushTable = new ConcurrentHashMap();
+ private ConcurrentHashMap<MessageQueue, Set<RemotingChannel>> pushTable = new ConcurrentHashMap();
- private ConcurrentHashMap<RemotingChannel, Set<String>/*Topic#QueueId*/> clientSubscriptionTable = new ConcurrentHashMap<>(2048);
-
- private final String pushKeySeparator = "#";
-
- private String buildPushKey(String topic, Integer queueId) {
- if (topic != null && queueId != null) {
- StringBuffer stringBuffer = new StringBuffer(32);
- stringBuffer.append(topic).append(pushKeySeparator).append(queueId);
- return stringBuffer.toString();
- }
- return null;
- }
+ private ConcurrentHashMap<RemotingChannel, Set<MessageQueue>> clientSubscriptionTable = new ConcurrentHashMap<>(2048);
@Override
public void registerPushSession(Set<SubscriptionData> subscriptionDataSet, RemotingChannel remotingChannel,
String groupId) {
- Set<String> prevSubSet = this.clientSubscriptionTable.get(remotingChannel);
- Set<String> keySet = new HashSet<>();
- for (SubscriptionData subscriptionData : subscriptionDataSet) {
- if (subscriptionData.getTopic() != null && subscriptionData.getQueueIdSet() != null && remotingChannel != null) {
- for (Integer queueId : subscriptionData.getQueueIdSet()) {
- String key = buildPushKey(subscriptionData.getTopic(), queueId);
- keySet.add(key);
- Set<RemotingChannel> clientSet = pushTable.get(key);
+ Set<MessageQueue> prevSubSet = this.clientSubscriptionTable.get(remotingChannel);
+ Set<MessageQueue> keySet = new HashSet<>();
+ for (SubscriptionData tmp : subscriptionDataSet) {
+ PushSubscriptionData subscriptionData = (PushSubscriptionData) tmp;
+ if (subscriptionData.getTopic() != null && subscriptionData.getMessageQueueSet() != null && remotingChannel != null) {
+ for (MessageQueue messageQueue : subscriptionData.getMessageQueueSet()) {
+ keySet.add(messageQueue);
+ Set<RemotingChannel> clientSet = pushTable.get(messageQueue);
if (clientSet == null) {
clientSet = new HashSet<>();
- Set<RemotingChannel> prev = pushTable.putIfAbsent(key, clientSet);
+ Set<RemotingChannel> prev = pushTable.putIfAbsent(messageQueue, clientSet);
clientSet = prev != null ? prev : clientSet;
}
- log.info("Register push session key: {}, group: {} remoting: {}", key, groupId, remotingChannel.remoteAddress());
+ log.info("Register push session message queue: {}, group: {} remoting: {}", messageQueue, groupId, remotingChannel.remoteAddress());
clientSet.add(remotingChannel);
}
}
@@ -76,12 +67,12 @@ public class SubscriptionManagerImpl implements SubscriptionManager {
this.clientSubscriptionTable.putIfAbsent(remotingChannel, keySet);
}
if (prevSubSet != null) {
- for (String key : prevSubSet) {
- if (!keySet.contains(key)) {
- Set clientSet = pushTable.get(key);
+ for (MessageQueue messageQueue : prevSubSet) {
+ if (!keySet.contains(messageQueue)) {
+ Set clientSet = pushTable.get(messageQueue);
if (clientSet != null) {
clientSet.remove(remotingChannel);
- log.info("Remove key:{}", key);
+ log.info("Remove subscription message queue:{}", messageQueue);
}
}
}
@@ -90,9 +81,9 @@ public class SubscriptionManagerImpl implements SubscriptionManager {
@Override
public void removePushSession(RemotingChannel remotingChannel) {
- Set<String> subSet = this.clientSubscriptionTable.get(remotingChannel);
+ Set<MessageQueue> subSet = this.clientSubscriptionTable.get(remotingChannel);
if (subSet != null) {
- for (String key : subSet) {
+ for (MessageQueue key : subSet) {
Set clientSet = pushTable.get(key);
if (clientSet != null) {
log.info("Remove push key:{} remoting:{}", key, remotingChannel.remoteAddress());
@@ -107,10 +98,8 @@ public class SubscriptionManagerImpl implements SubscriptionManager {
}
@Override
- public Set<RemotingChannel> getPushableChannel(String topic, Integer queueId) {
- String key = buildPushKey(topic, queueId);
- log.info("Get pushableChannel by key: {}", key);
- return pushTable.get(key);
+ public Set<RemotingChannel> getPushableChannel(MessageQueue messageQueue) {
+ return pushTable.get(messageQueue);
}
private Subscription getSubscription(String groupId, ConsumeType consumeType,
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/processor/HeartbeatProcessor.java b/snode/src/main/java/org/apache/rocketmq/snode/processor/HeartbeatProcessor.java
index 3d2c03a..435d165 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/processor/HeartbeatProcessor.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/processor/HeartbeatProcessor.java
@@ -18,6 +18,8 @@ package org.apache.rocketmq.snode.processor;
import io.netty.channel.Channel;
import io.netty.util.Attribute;
+import java.util.HashSet;
+import java.util.Set;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.ResponseCode;
@@ -72,30 +74,34 @@ public class HeartbeatProcessor implements RequestProcessor {
Client client = new Client();
client.setClientId(heartbeatData.getClientID());
client.setRemotingChannel(remotingChannel);
-
+ Set<String> groupSet = new HashSet<>();
for (ProducerData producerData : heartbeatData.getProducerDataSet()) {
- client.setGroupId(producerData.getGroupName());
client.setClientRole(ClientRole.Producer);
- this.snodeController.getProducerManager().register(client);
+ groupSet.add(producerData.getGroupName());
+ this.snodeController.getProducerManager().register(producerData.getGroupName(), client);
}
- for (ConsumerData data : heartbeatData.getConsumerDataSet()) {
- client.setGroupId(data.getGroupName());
+ log.info("Heartbeat consumerData: {}", heartbeatData.getConsumerDataSet());
+ for (ConsumerData consumerData : heartbeatData.getConsumerDataSet()) {
client.setClientRole(ClientRole.Consumer);
- boolean channelChanged = this.snodeController.getConsumerManager().register(client);
- boolean subscriptionChanged = this.snodeController.getSubscriptionManager().subscribe(data.getGroupName(),
- data.getSubscriptionDataSet(),
- data.getConsumeType(),
- data.getMessageModel(),
- data.getConsumeFromWhere());
- if (data.getConsumeType() == ConsumeType.CONSUME_PUSH) {
+ groupSet.add(consumerData.getGroupName());
+ boolean channelChanged = this.snodeController.getConsumerManager().register(consumerData.getGroupName(), client);
+ boolean subscriptionChanged = this.snodeController.getSubscriptionManager().subscribe(consumerData.getGroupName(),
+ consumerData.getSubscriptionDataSet(),
+ consumerData.getConsumeType(),
+ consumerData.getMessageModel(),
+ consumerData.getConsumeFromWhere());
+ if (consumerData.getConsumeType() == ConsumeType.CONSUME_PUSH) {
NettyChannelImpl nettyChannel = new NettyChannelImpl(channel);
- this.snodeController.getSubscriptionManager().registerPushSession(data.getSubscriptionDataSet(), nettyChannel, data.getGroupName());
+ this.snodeController.getSubscriptionManager().registerPushSession(consumerData.getSubscriptionDataSet(), nettyChannel, consumerData.getGroupName());
}
if (subscriptionChanged || channelChanged) {
- this.snodeController.getClientService().notifyConsumer(data.getGroupName());
+ this.snodeController.getClientService().notifyConsumer(consumerData.getGroupName());
}
}
+ if (groupSet.size() > 0) {
+ client.setGroups(groupSet);
+ }
clientAttribute.setIfAbsent(client);
RemotingCommand response = RemotingCommand.createResponseCommand(null);
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/service/AdminService.java b/snode/src/main/java/org/apache/rocketmq/snode/service/AdminService.java
index b19a474..560233a 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/service/AdminService.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/service/AdminService.java
@@ -16,4 +16,5 @@ package org.apache.rocketmq.snode.service;/*
*/
public interface AdminService {
+
}
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/service/impl/PushServiceImpl.java b/snode/src/main/java/org/apache/rocketmq/snode/service/impl/PushServiceImpl.java
index f6a3ac6..9883dd0 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/service/impl/PushServiceImpl.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/service/impl/PushServiceImpl.java
@@ -16,12 +16,15 @@
*/
package org.apache.rocketmq.snode.service.impl;
+import io.netty.channel.Channel;
+import io.netty.util.Attribute;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.header.PushMessageHeader;
import org.apache.rocketmq.common.protocol.header.SendMessageResponseHeader;
@@ -29,8 +32,11 @@ import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.RemotingChannel;
+import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.snode.SnodeController;
+import org.apache.rocketmq.snode.client.Client;
+import org.apache.rocketmq.snode.client.impl.Subscription;
import org.apache.rocketmq.snode.constant.SnodeConstant;
import org.apache.rocketmq.snode.service.PushService;
@@ -80,24 +86,37 @@ public class PushServiceImpl implements PushService {
pushMessageHeader.setQueueId(queueId);
RemotingCommand pushMessage = RemotingCommand.createRequestCommand(RequestCode.SNODE_PUSH_MESSAGE, pushMessageHeader);
pushMessage.setBody(message);
- Set<RemotingChannel> consumerTable = snodeController.getSubscriptionManager().getPushableChannel(topic, queueId);
+ MessageQueue messageQueue = new MessageQueue(topic, enodeName, queueId);
+ Set<RemotingChannel> consumerTable = snodeController.getSubscriptionManager().getPushableChannel(messageQueue);
if (consumerTable != null) {
for (RemotingChannel remotingChannel : consumerTable) {
- if (remotingChannel.isWritable()) {
- boolean slowConsumer = snodeController.getSlowConsumerService().isSlowConsumer(sendMessageResponseHeader.getQueueOffset(), topic, queueId, remotingChannel, enodeName);
- if (slowConsumer) {
- log.warn("[SlowConsumer]: {} closed as slow consumer", remotingChannel);//TODO metrics
- remotingChannel.close();
- continue;
+ Client client = null;
+ if (remotingChannel instanceof NettyChannelImpl) {
+ Channel channel = ((NettyChannelImpl) remotingChannel).getChannel();
+ Attribute<Client> clientAttribute = channel.attr(SnodeConstant.NETTY_CLIENT_ATTRIBUTE_KEY);
+ if (clientAttribute != null) {
+ client = clientAttribute.get();
+ }
+ }
+ if (client != null) {
+ for (String consumerGroup : client.getGroups()) {
+ Subscription subscription = snodeController.getSubscriptionManager().getSubscription(consumerGroup);
+ if (subscription.getSubscriptionData(topic) != null) {
+ boolean slowConsumer = snodeController.getSlowConsumerService().isSlowConsumer(sendMessageResponseHeader.getQueueOffset(), topic, queueId, consumerGroup, enodeName);
+ if (slowConsumer) {
+ log.warn("[SlowConsumer]: {} closed as slow consumer", remotingChannel);//TODO metrics
+ remotingChannel.close();//FIXME this action should be discussed
+ continue;
+ }
+ snodeController.getSnodeServer().push(remotingChannel, pushMessage, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
+ }
}
- log.debug("Push message to remotingChannel: {}", remotingChannel.remoteAddress());
- snodeController.getSnodeServer().push(remotingChannel, pushMessage, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
} else {
- log.warn("Remoting channel is not writable: {}", remotingChannel.remoteAddress());
+ log.error("[NOTIFYME] Remoting channel: {} related client is null", remotingChannel.remoteAddress());
}
}
} else {
- log.warn("Get client info to topic: {} queueId: {} is null", topic, queueId);
+ log.info("No online registered as push consumer and online for messageQueue: {} ", messageQueue);
}
} catch (Exception ex) {
log.warn("Push message to topic: {} queueId: {}", topic, queueId, ex);
@@ -116,13 +135,8 @@ public class PushServiceImpl implements PushService {
@Override
public void pushMessage(final String enodeName, final String topic, final Integer queueId, final byte[] message,
final RemotingCommand response) {
- Set<RemotingChannel> pushableChannels = this.snodeController.getSubscriptionManager().getPushableChannel(topic, queueId);
- if (pushableChannels != null) {
- PushTask pushTask = new PushTask(topic, queueId, message, response, enodeName);
- pushMessageExecutorService.submit(pushTask);
- } else {
- log.info("Topic: {} QueueId: {} no need to push", topic, queueId);
- }
+ PushTask pushTask = new PushTask(topic, queueId, message, response, enodeName);
+ pushMessageExecutorService.submit(pushTask);
}
@Override