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

Browse Source
This commit is contained in:
Redkale
2020-06-14 01:01:38 +08:00
parent 13217a11c0
commit 0377dac71f
1 changed files with 5 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/org/redkale/mq/MessageClient.java
View File

@@ -65,17 +65,18 @@ public abstract class MessageClient {
synchronized (this) { synchronized (this) {
if (this.respConsumerid == null) this.respConsumerid = "consumer-" + this.respTopic; if (this.respConsumerid == null) this.respConsumerid = "consumer-" + this.respTopic;
if (this.consumer == null) { if (this.consumer == null) {
MessageProcessor processor = (msg,callback) -> { MessageProcessor processor = (msg, callback) -> {
MessageRespFutureNode node = respNodes.get(msg.getSeqid()); MessageRespFutureNode node = respNodes.get(msg.getSeqid());
if (node == null) { if (node == null) {
messageAgent.logger.log(Level.WARNING, MessageClient.this.getClass().getSimpleName() + " process " + msg + " error"); messageAgent.logger.log(Level.WARNING, MessageClient.this.getClass().getSimpleName() + " process " + msg + " error， not found msgnode");
return; return;
} }
if (node.getCounter() != null) node.getCounter().decrementAndGet(); if (node.getCounter() != null) node.getCounter().decrementAndGet();
node.future.complete(msg); node.future.complete(msg);
}; };
this.consumer = messageAgent.createConsumer(new String[]{respTopic}, respConsumerid, processor); MessageConsumer one = messageAgent.createConsumer(new String[]{respTopic}, respConsumerid, processor);
this.consumer.startup().join(); one.startup().join();
this.consumer = one;
} }
} }
} }