src/org/redkale/mq/MessageAgent.java

@@ -62,8 +62,7 @@ public abstract class MessageAgent {
         if (this.sncpRespStartms >= 0) return;
         long s = System.currentTimeMillis();
         if (this.sncpRespConsumer != null) {
-            this.sncpRespConsumer.start();
-            this.sncpRespConsumer.waitFor();
+            this.sncpRespConsumer.startup().join();
         }
         this.sncpRespStartms = System.currentTimeMillis() - s;
     }
@@ -78,8 +77,7 @@ public abstract class MessageAgent {
         }
         this.messageNodes.values().forEach(node -> {
             long s = System.currentTimeMillis();
-            node.consumer.start();
-            node.consumer.waitFor();
+            node.consumer.startup().join();
             sb.append("MessageConsumer(topic=").append(fillString(node.consumer.topic, maxlen.get())).append(") init and start in ").append(System.currentTimeMillis() - s).append(" ms\r\n");
         });
         return CompletableFuture.completedFuture(null);
@@ -87,7 +85,7 @@ public abstract class MessageAgent {
 
     public CompletableFuture<Void> stop() {
         this.messageNodes.values().forEach(node -> {
-            node.consumer.close();
+            node.consumer.shutdown().join();
         });
         return CompletableFuture.completedFuture(null);
     }
@@ -122,8 +120,7 @@ public abstract class MessageAgent {
     public synchronized MessageProducer getProducer() {
         if (this.producer == null) {
             this.producer = createProducer();
-            this.producer.start();
-            this.producer.waitFor();
+            this.producer.startup().join();
         }
         return this.producer;
     }

src/org/redkale/mq/MessageConsumer.java

@@ -6,6 +6,7 @@
 package org.redkale.mq;
 
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
 import java.util.logging.Logger;
 
 /**
@@ -18,7 +19,7 @@ import java.util.logging.Logger;
  *
  * @since 2.1.0
  */
-public abstract class MessageConsumer extends Thread {
+public abstract class MessageConsumer {
 
     protected final String topic;
 
@@ -47,11 +48,11 @@ public abstract class MessageConsumer extends Thread {
         return topic;
     }
 
-    public abstract void waitFor();
+    public abstract CompletableFuture<Void> startup();
 
     protected boolean isClosed() {
         return closed;
     }
 
-    protected abstract void close();
+    protected abstract CompletableFuture<Void> shutdown();
 }

src/org/redkale/mq/MessageProducer.java

@@ -17,7 +17,7 @@ import java.util.logging.Logger;
  *
  * @since 2.1.0
  */
-public abstract class MessageProducer extends Thread {
+public abstract class MessageProducer  {
 
     protected final Logger logger = Logger.getLogger(this.getClass().getSimpleName());
 
@@ -25,11 +25,11 @@ public abstract class MessageProducer extends Thread {
 
     public abstract CompletableFuture<Void> apply(MessageRecord message);
 
-    protected abstract void waitFor();
+    public abstract CompletableFuture<Void> startup();
 
     protected boolean isClosed() {
         return closed;
     }
 
-    protected abstract void close();
+    protected abstract CompletableFuture<Void> shutdown();
 }