lxy/redkale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

mq

Browse Source
This commit is contained in:
redkale
2023-10-14 15:29:38 +08:00
parent c2e3636062
commit 17fbb29957
2 changed files with 8 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/java/org/redkale/mq/MessageAgent.java
View File

@@ -110,8 +110,8 @@ public abstract class MessageAgent implements Resourcable {
this.workExecutor = threads > 0 ? WorkThread.createExecutor(threads, "Redkale-MessageConsumerThread-[" + name + "]-%s") this.workExecutor = threads > 0 ? WorkThread.createExecutor(threads, "Redkale-MessageConsumerThread-[" + name + "]-%s")
: WorkThread.createWorkExecutor(Utility.cpus(), "Redkale-MessageConsumerThread-[" + name + "]-%s"); : WorkThread.createWorkExecutor(Utility.cpus(), "Redkale-MessageConsumerThread-[" + name + "]-%s");
} }
this.httpMessageClient = new MessageClient(this, this.httpAppRespTopic); this.httpMessageClient = new MessageClient("http", this, this.httpAppRespTopic);
this.sncpMessageClient = new MessageClient(this, this.sncpAppRespTopic); this.sncpMessageClient = new MessageClient("sncp", this, this.sncpAppRespTopic);
String coderType = config.getValue("coder", ""); String coderType = config.getValue("coder", "");
if (!coderType.trim().isEmpty()) { if (!coderType.trim().isEmpty()) {
@@ -198,8 +198,6 @@ public abstract class MessageAgent implements Resourcable {
this.messageConsumerList.clear(); this.messageConsumerList.clear();
this.messageConsumerMap.clear(); this.messageConsumerMap.clear();
//-------------- MessageClient -------------- //-------------- MessageClient --------------
this.httpMessageClient.stop();
this.sncpMessageClient.stop();
if (this.messageClientProducer != null) { if (this.messageClientProducer != null) {
this.messageClientProducer.stop(); this.messageClientProducer.stop();
} }

11
src/main/java/org/redkale/mq/MessageClient.java
View File

@@ -48,12 +48,15 @@ public class MessageClient implements ClusterRpcClient<MessageRecord, MessageRec
protected final AtomicLong msgSeqno; protected final AtomicLong msgSeqno;
protected final String protocol;
//key: reqTopic //key: reqTopic
private final HashMap<String, MessageProcessor> messageProcessors = new HashMap<>(); private final HashMap<String, MessageProcessor> messageProcessors = new HashMap<>();
final ConcurrentHashMap<Long, MessageRespFuture> respQueue = new ConcurrentHashMap<>(); final ConcurrentHashMap<Long, MessageRespFuture> respQueue = new ConcurrentHashMap<>();
protected MessageClient(MessageAgent messageAgent, String appRespTopic) { protected MessageClient(String protocol, MessageAgent messageAgent, String appRespTopic) {
this.protocol = protocol;
this.messageAgent = messageAgent; this.messageAgent = messageAgent;
this.appRespTopic = appRespTopic; this.appRespTopic = appRespTopic;
this.msgSeqno = messageAgent.msgSeqno; this.msgSeqno = messageAgent.msgSeqno;
@@ -206,10 +209,8 @@ public class MessageClient implements ClusterRpcClient<MessageRecord, MessageRec
return ctype; return ctype;
} }
public void start() { public String getProtocol() {
} return protocol;
public void stop() {
} }
public MessageAgent getMessageAgent() { public MessageAgent getMessageAgent() {