src/org/redkale/boot/Application.java

@@ -778,6 +778,13 @@ public final class Application {
         runServers(timecd, others);
         runServers(timecd, watchs); //必须在所有服务都启动后再启动WATCH服务
         timecd.await();
+        if (this.messageAgents != null) {
+            long s = System.currentTimeMillis();
+            for (NodeServer ns : servers) {
+                ns.messageAgents.values().forEach(agent -> agent.start().join());
+            }
+            logger.info(this.getClass().getSimpleName() + " messageagent init in " + (System.currentTimeMillis() - s) + " ms\r\n");
+        }
         //if (!singletonrun) signalHandle();
         //if (!singletonrun) clearPersistData();
         logger.info(this.getClass().getSimpleName() + " started in " + (System.currentTimeMillis() - startTime) + " ms\r\n");

src/org/redkale/boot/NodeServer.java

@@ -550,9 +550,6 @@ public abstract class NodeServer {
 
     //Server.start执行之后调用
     protected void postStartServer(Set<Service> localServices, Set<Service> remoteServices) {
-        if (!this.messageAgents.isEmpty()) { //MQ
-            this.messageAgents.values().forEach(agent -> agent.start());
-        }
     }
 
     protected abstract ClassFilter<Filter> createFilterClassFilter();

src/org/redkale/mq/MessageAgent.java

@@ -48,7 +48,11 @@ public abstract class MessageAgent {
     }
 
     public CompletableFuture<Void> start() {
-        return null;
+        this.httpNodes.values().forEach(node -> {
+            node.consumer.start();
+            node.consumer.waitFor();
+        });
+        return CompletableFuture.completedFuture(null);
     }
 
     public CompletableFuture<Void> stop() {