feat(cluster): 优化集群创建器并完善集群消息广播功能

ChunMengLu · ChunMengLu · commit 85bcfb3c80fa · 2026-03-16T10:56:18.000+08:00
- 新增 MqttBroker 入口类简化 Broker 创建过程
- MqttClusterConfig 支持链式调用，提升配置便捷性
- 优化 MqttClusterBrokerCreator 支持 start() 直接启动集群和 MQTT 服务
- 集群管理器新增 publish 方法实现集群广播消息的智能转发
- 调整 ClusterTest 集群测试用例，增加定时广播消息示例
- 集群消息类型处理代码简化，提升可读性和一致性
- MqttClusterIntegrationTest 补充广播消息接收的断言验证
- 优化导入包，去除冗余并统一消息常量引用方式
diff --git a/mica-mqtt-broker/src/main/java/org/dromara/mica/mqtt/broker/MqttBroker.java b/mica-mqtt-broker/src/main/java/org/dromara/mica/mqtt/broker/MqttBroker.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & dreamlu.net).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.dromara.mica.mqtt.broker;
+
+import org.dromara.mica.mqtt.broker.cluster.MqttClusterBrokerCreator;
+import org.dromara.mica.mqtt.core.server.MqttServer;
+import org.dromara.mica.mqtt.core.server.MqttServerCreator;
+
+/**
+ * Mqtt Broker
+ * <p>
+ * Entry point for creating MQTT broker instances, including cluster support.
+ * </p>
+ *
+ * @author L.cm
+ * @author opencode
+ */
+public class MqttBroker {
+
+    /**
+     * Create a new MqttClusterBrokerCreator using an existing MqttServerCreator
+     *
+     * @param serverCreator the underlying MqttServerCreator
+     * @return MqttClusterBrokerCreator
+     */
+    public static MqttClusterBrokerCreator create(MqttServerCreator serverCreator) {
+        return new MqttClusterBrokerCreator(serverCreator);
+    }
+
+    /**
+     * Create a new MqttClusterBrokerCreator using a new MqttServerCreator
+     *
+     * @return MqttClusterBrokerCreator
+     */
+    public static MqttClusterBrokerCreator create() {
+        return new MqttClusterBrokerCreator(MqttServer.create());
+    }
+
+}
diff --git a/mica-mqtt-broker/src/main/java/org/dromara/mica/mqtt/broker/cluster/MqttClusterBrokerCreator.java b/mica-mqtt-broker/src/main/java/org/dromara/mica/mqtt/broker/cluster/MqttClusterBrokerCreator.java
@@ -16,11 +16,11 @@
 
 package org.dromara.mica.mqtt.broker.cluster;
 
+import org.dromara.mica.mqtt.broker.cluster.dispatcher.ClusterMessageDispatcher;
 import org.dromara.mica.mqtt.core.server.MqttServer;
 import org.dromara.mica.mqtt.core.server.MqttServerCreator;
 import org.dromara.mica.mqtt.core.server.session.IMqttSessionManager;
 import org.dromara.mica.mqtt.core.server.session.InMemoryMqttSessionManager;
-import org.dromara.mica.mqtt.broker.cluster.dispatcher.ClusterMessageDispatcher;
 
 /**
  * 集群模式的 Broker 创建器
@@ -40,7 +40,7 @@ public MqttClusterBrokerCreator clusterConfig(MqttClusterConfig config) {
         return this;
     }
 
-    public MqttServer build() throws Exception {
+    public MqttServer build() {
         if (clusterConfig == null || !clusterConfig.isEnabled()) {
             return serverCreator.build();
         }
@@ -71,12 +71,27 @@ public MqttServer build() throws Exception {
         ClusterMessageDispatcher dispatcher = new ClusterMessageDispatcher(mqttServer, clusterManager, clusterSessionManager);
         serverCreator.getMessagePipeline().addHandler(dispatcher);
 
-        // 6. 启动集群管理器（会自动启动 MQTT 服务）
-        clusterManager.start();
+        return mqttServer;
+    }
 
+    public MqttServer start() {
+        MqttServer mqttServer = this.build();
+        try {
+            if (clusterManager != null) {
+                clusterManager.start();
+            } else {
+                mqttServer.start();
+            }
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to start MqttClusterBroker", e);
+        }
         return mqttServer;
     }
 
+    public MqttServerCreator getServerCreator() {
+        return serverCreator;
+    }
+
     public MqttClusterManager getClusterManager() {
         return clusterManager;
     }
diff --git a/mica-mqtt-broker/src/main/java/org/dromara/mica/mqtt/broker/cluster/MqttClusterConfig.java b/mica-mqtt-broker/src/main/java/org/dromara/mica/mqtt/broker/cluster/MqttClusterConfig.java
@@ -81,4 +81,39 @@ public long getNodeTimeout() {
     public void setNodeTimeout(long nodeTimeout) {
         this.nodeTimeout = nodeTimeout;
     }
+
+    public MqttClusterConfig enabled(boolean enabled) {
+        this.enabled = enabled;
+        return this;
+    }
+
+    public MqttClusterConfig clusterHost(String clusterHost) {
+        this.clusterHost = clusterHost;
+        return this;
+    }
+
+    public MqttClusterConfig clusterPort(int clusterPort) {
+        this.clusterPort = clusterPort;
+        return this;
+    }
+
+    public MqttClusterConfig seedMembers(List<String> seedMembers) {
+        this.seedMembers = seedMembers;
+        return this;
+    }
+
+    public MqttClusterConfig clusterName(String clusterName) {
+        this.clusterName = clusterName;
+        return this;
+    }
+
+    public MqttClusterConfig heartbeatInterval(long heartbeatInterval) {
+        this.heartbeatInterval = heartbeatInterval;
+        return this;
+    }
+
+    public MqttClusterConfig nodeTimeout(long nodeTimeout) {
+        this.nodeTimeout = nodeTimeout;
+        return this;
+    }
 }
diff --git a/mica-mqtt-broker/src/main/java/org/dromara/mica/mqtt/broker/cluster/MqttClusterManager.java b/mica-mqtt-broker/src/main/java/org/dromara/mica/mqtt/broker/cluster/MqttClusterManager.java
@@ -1,8 +1,7 @@
 package org.dromara.mica.mqtt.broker.cluster;
 
-import org.dromara.mica.mqtt.broker.cluster.message.ClusterMessage;
-import org.dromara.mica.mqtt.broker.cluster.message.GenericClusterMessage;
-import org.dromara.mica.mqtt.broker.cluster.message.StateSyncResponseMessage;
+import org.dromara.mica.mqtt.broker.cluster.message.*;
+import org.dromara.mica.mqtt.codec.MqttQoS;
 import org.dromara.mica.mqtt.core.server.MqttServer;
 import org.dromara.mica.mqtt.core.server.model.Subscribe;
 import org.slf4j.Logger;
@@ -12,11 +11,9 @@
 import org.tio.server.cluster.core.ClusterConfig;
 import org.tio.server.cluster.core.ClusterImpl;
 import org.tio.server.cluster.message.ClusterDataMessage;
+
 import java.io.*;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 public class MqttClusterManager {
     private static final Logger logger = LoggerFactory.getLogger(MqttClusterManager.class);
@@ -78,7 +75,7 @@ private void requestStateSync() {
             if (!nodeToString(seed).equals(localNodeId)) {
                 try {
                     GenericClusterMessage syncRequest = new GenericClusterMessage();
-                    syncRequest.setType(org.dromara.mica.mqtt.broker.cluster.message.MessageType.STATE_SYNC_REQUEST);
+                    syncRequest.setType(MessageType.STATE_SYNC_REQUEST);
                     syncRequest.setSourceNode(localNodeId);
                     cluster.send(seed, serialize(syncRequest));
                     logger.info("Sent state sync request to seed node: {}", seed);
@@ -109,33 +106,28 @@ private void handleClusterMessage(ClusterDataMessage message) {
     private void handleClusterMessageInternal(ClusterMessage clusterMsg, ClusterDataMessage rawMessage) {
         ClusterMqttSessionManager sessionManager = (ClusterMqttSessionManager) mqttServer.getServerCreator().getSessionManager();
         
-        if (clusterMsg.getType() == org.dromara.mica.mqtt.broker.cluster.message.MessageType.PUBLISH_FORWARD) {
-            org.dromara.mica.mqtt.broker.cluster.message.PublishForwardMessage pfm = 
-                (org.dromara.mica.mqtt.broker.cluster.message.PublishForwardMessage) clusterMsg;
-            mqttServer.publishAll(pfm.getTopic(), pfm.getPayload(), org.dromara.mica.mqtt.codec.MqttQoS.valueOf(pfm.getQos()), pfm.isRetain());
-        } else if (clusterMsg.getType() == org.dromara.mica.mqtt.broker.cluster.message.MessageType.SUBSCRIBE_NOTIFY) {
-            org.dromara.mica.mqtt.broker.cluster.message.SubscribeNotifyMessage snm = 
-                (org.dromara.mica.mqtt.broker.cluster.message.SubscribeNotifyMessage) clusterMsg;
+        if (clusterMsg.getType() == MessageType.PUBLISH_FORWARD) {
+            PublishForwardMessage pfm = (PublishForwardMessage) clusterMsg;
+            mqttServer.publishAll(pfm.getTopic(), pfm.getPayload(), MqttQoS.valueOf(pfm.getQos()), pfm.isRetain());
+        } else if (clusterMsg.getType() == MessageType.SUBSCRIBE_NOTIFY) {
+            SubscribeNotifyMessage snm = (SubscribeNotifyMessage) clusterMsg;
             sessionManager.syncRemoteSubscriptions(snm.getClientId(), snm.getNodeId(), snm.getSubscriptions());
-        } else if (clusterMsg.getType() == org.dromara.mica.mqtt.broker.cluster.message.MessageType.UNSUBSCRIBE_NOTIFY) {
-            org.dromara.mica.mqtt.broker.cluster.message.UnsubscribeNotifyMessage unm = 
-                (org.dromara.mica.mqtt.broker.cluster.message.UnsubscribeNotifyMessage) clusterMsg;
+        } else if (clusterMsg.getType() == MessageType.UNSUBSCRIBE_NOTIFY) {
+            UnsubscribeNotifyMessage unm = (UnsubscribeNotifyMessage) clusterMsg;
             sessionManager.removeRemoteSubscriptions(unm.getClientId(), unm.getTopics());
-        } else if (clusterMsg.getType() == org.dromara.mica.mqtt.broker.cluster.message.MessageType.CLIENT_CONNECT) {
-            org.dromara.mica.mqtt.broker.cluster.message.ClientConnectMessage ccm = 
-                (org.dromara.mica.mqtt.broker.cluster.message.ClientConnectMessage) clusterMsg;
+        } else if (clusterMsg.getType() == MessageType.CLIENT_CONNECT) {
+            ClientConnectMessage ccm = (ClientConnectMessage) clusterMsg;
             sessionManager.registerRemoteClient(ccm.getClientId(), ccm.getSourceNode());
-        } else if (clusterMsg.getType() == org.dromara.mica.mqtt.broker.cluster.message.MessageType.CLIENT_DISCONNECT) {
-            org.dromara.mica.mqtt.broker.cluster.message.ClientDisconnectMessage cdm = 
-                (org.dromara.mica.mqtt.broker.cluster.message.ClientDisconnectMessage) clusterMsg;
+        } else if (clusterMsg.getType() == MessageType.CLIENT_DISCONNECT) {
+            ClientDisconnectMessage cdm = (ClientDisconnectMessage) clusterMsg;
             sessionManager.removeRemoteClient(cdm.getClientId());
-        } else if (clusterMsg.getType() == org.dromara.mica.mqtt.broker.cluster.message.MessageType.STATE_SYNC_REQUEST) {
+        } else if (clusterMsg.getType() == MessageType.STATE_SYNC_REQUEST) {
             handleStateSyncRequest(clusterMsg.getSourceNode());
-        } else if (clusterMsg.getType() == org.dromara.mica.mqtt.broker.cluster.message.MessageType.STATE_SYNC_RESPONSE) {
+        } else if (clusterMsg.getType() == MessageType.STATE_SYNC_RESPONSE) {
             StateSyncResponseMessage ssm = (StateSyncResponseMessage) clusterMsg;
             sessionManager.syncFullState(ssm.getClientNodeMap(), ssm.getSubscriptionMap());
             logger.info("State sync completed, received {} client mappings", ssm.getClientNodeMap().size());
-        } else if (clusterMsg.getType() == org.dromara.mica.mqtt.broker.cluster.message.MessageType.NODE_LEAVE) {
+        } else if (clusterMsg.getType() == MessageType.NODE_LEAVE) {
             // 节点离开消息，需要清理该节点的所有订阅
             String leavingNodeId = clusterMsg.getSourceNode();
             sessionManager.clearNodeClientsAndSubscriptions(leavingNodeId);
@@ -150,7 +142,7 @@ private void handleStateSyncRequest(String requestNodeId) {
         ClusterMqttSessionManager sessionManager = (ClusterMqttSessionManager) mqttServer.getServerCreator().getSessionManager();
         
         StateSyncResponseMessage response = new StateSyncResponseMessage();
-        response.setType(org.dromara.mica.mqtt.broker.cluster.message.MessageType.STATE_SYNC_RESPONSE);
+        response.setType(MessageType.STATE_SYNC_RESPONSE);
         response.setSourceNode(localNodeId);
         
         // 获取远程客户端映射
@@ -214,7 +206,7 @@ public void stop() {
         if (cluster != null) {
             // 广播节点离开消息
             GenericClusterMessage leaveMsg = new GenericClusterMessage();
-            leaveMsg.setType(org.dromara.mica.mqtt.broker.cluster.message.MessageType.NODE_LEAVE);
+            leaveMsg.setType(MessageType.NODE_LEAVE);
             leaveMsg.setSourceNode(localNodeId);
             broadcast(leaveMsg);
             
@@ -229,9 +221,55 @@ public void stop() {
         }
     }
 
+    /**
+     * 集群级别的下发消息：发布消息到集群中的所有匹配订阅者
+     *
+     * @param topic   主题
+     * @param payload 消息体
+     * @param qos     QoS
+     * @param retain  是否保留消息
+     */
+    public void publish(String topic, byte[] payload, int qos, boolean retain) {
+        if (mqttServer == null) {
+            return;
+        }
+        
+        // 1. 先在本地节点下发
+        mqttServer.publishAll(topic, payload, MqttQoS.valueOf(qos), retain);
+        
+        // 2. 查找是否有其他节点存在该 topic 的订阅者，按需转发以节省网络开销
+        if (config.isEnabled() && cluster != null) {
+            ClusterMqttSessionManager sessionManager = (ClusterMqttSessionManager) mqttServer.getServerCreator().getSessionManager();
+            List<Subscribe> allSubs = sessionManager.searchAllSubscribe(topic);
+            
+            if (allSubs != null && !allSubs.isEmpty()) {
+                Set<String> targetNodes = new HashSet<>();
+                for (Subscribe sub : allSubs) {
+                    String node = sessionManager.getClientNode(sub.getClientId());
+                    if (node != null && !node.equals(localNodeId)) {
+                        targetNodes.add(node);
+                    }
+                }
+                
+                if (!targetNodes.isEmpty()) {
+                    PublishForwardMessage clusterMsg = new PublishForwardMessage();
+                    clusterMsg.setType(MessageType.PUBLISH_FORWARD);
+                    clusterMsg.setTopic(topic);
+                    clusterMsg.setPayload(payload);
+                    clusterMsg.setQos(qos);
+                    clusterMsg.setRetain(retain);
+                    
+                    for (String node : targetNodes) {
+                        sendToNode(node, clusterMsg);
+                    }
+                }
+            }
+        }
+    }
+
     public Collection<Node> getRemoteMembers() {
         if (cluster == null) {
-            return java.util.Collections.emptyList();
+            return Collections.emptyList();
         }
         return cluster.getRemoteMembers();
     }
diff --git a/mica-mqtt-broker/src/test/java/org/dromara/mica/mqtt/broker/cluster/ClusterTestNode1.java b/mica-mqtt-broker/src/test/java/org/dromara/mica/mqtt/broker/cluster/ClusterTestNode1.java
@@ -16,6 +16,7 @@
 
 package org.dromara.mica.mqtt.broker.cluster;
 
+import org.dromara.mica.mqtt.broker.MqttBroker;
 import org.dromara.mica.mqtt.core.server.MqttServer;
 import org.dromara.mica.mqtt.core.server.MqttServerCreator;
 
@@ -36,11 +37,11 @@ public static void main(String[] args) throws Exception {
         System.out.println("========================================");
 
         // 1. 集群配置
-        MqttClusterConfig clusterConfig = new MqttClusterConfig();
-        clusterConfig.setEnabled(true);
-        clusterConfig.setClusterHost("127.0.0.1");
-        clusterConfig.setClusterPort(9001);
-        clusterConfig.setSeedMembers(Arrays.asList("127.0.0.1:9001", "127.0.0.1:9002", "127.0.0.1:9003"));
+        MqttClusterConfig clusterConfig = new MqttClusterConfig()
+            .enabled(true)
+            .clusterHost("127.0.0.1")
+            .clusterPort(9001)
+            .seedMembers(Arrays.asList("127.0.0.1:9001", "127.0.0.1:9002", "127.0.0.1:9003"));
 
         // 2. 创建 MQTT Server
         MqttServerCreator creator = MqttServer.create()
@@ -49,14 +50,35 @@ public static void main(String[] args) throws Exception {
             .enableMqtt(1883);
 
         // 3. 使用集群创建器构建并启动
-        MqttServer mqttServer = new MqttClusterBrokerCreator(creator)
-            .clusterConfig(clusterConfig)
-            .build();
+        MqttClusterBrokerCreator brokerCreator = MqttBroker.create(creator)
+            .clusterConfig(clusterConfig);
+        MqttServer mqttServer = brokerCreator.start();
+        MqttClusterManager clusterManager = brokerCreator.getClusterManager();
 
         System.out.println("Node 1 started successfully!");
         System.out.println("MQTT Server listening on port 1883");
 
-        // 4. 等待关闭
+        // 4. 定时下发测试消息（每隔5秒下发一次集群广播消息）
+        new Thread(() -> {
+            int count = 1;
+            while (true) {
+                try {
+                    Thread.sleep(5000);
+                    
+                    // 集群广播下发测试（所有订阅了 /test/cluster/topic 的设备都会收到）
+                    String broadcastMsg = "Broadcast message from Node 1, count: " + count;
+                    clusterManager.publish("/test/cluster/topic", broadcastMsg.getBytes(), 0, false);
+                    System.out.println("[Node 1] Published broadcast: " + broadcastMsg);
+
+                    count++;
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    break;
+                }
+            }
+        }).start();
+
+        // 5. 等待关闭
         Thread.sleep(Long.MAX_VALUE);
     }
 }
diff --git a/mica-mqtt-broker/src/test/java/org/dromara/mica/mqtt/broker/cluster/ClusterTestNode2.java b/mica-mqtt-broker/src/test/java/org/dromara/mica/mqtt/broker/cluster/ClusterTestNode2.java
@@ -36,11 +36,11 @@ public static void main(String[] args) throws Exception {
         System.out.println("========================================");
 
         // 1. 集群配置
-        MqttClusterConfig clusterConfig = new MqttClusterConfig();
-        clusterConfig.setEnabled(true);
-        clusterConfig.setClusterHost("127.0.0.1");
-        clusterConfig.setClusterPort(9002);
-        clusterConfig.setSeedMembers(Arrays.asList("127.0.0.1:9001", "127.0.0.1:9002", "127.0.0.1:9003"));
+        MqttClusterConfig clusterConfig = new MqttClusterConfig()
+            .enabled(true)
+            .clusterHost("127.0.0.1")
+            .clusterPort(9002)
+            .seedMembers(Arrays.asList("127.0.0.1:9001", "127.0.0.1:9002", "127.0.0.1:9003"));
 
         // 2. 创建 MQTT Server
         MqttServerCreator creator = MqttServer.create()
@@ -49,9 +49,9 @@ public static void main(String[] args) throws Exception {
             .enableMqtt(1884);
 
         // 3. 使用集群创建器构建并启动
-        MqttServer mqttServer = new MqttClusterBrokerCreator(creator)
+        MqttServer mqttServer = org.dromara.mica.mqtt.broker.MqttBroker.create(creator)
             .clusterConfig(clusterConfig)
-            .build();
+            .start();
 
         System.out.println("Node 2 started successfully!");
         System.out.println("MQTT Server listening on port 1884");
diff --git a/mica-mqtt-broker/src/test/java/org/dromara/mica/mqtt/broker/cluster/ClusterTestNode3.java b/mica-mqtt-broker/src/test/java/org/dromara/mica/mqtt/broker/cluster/ClusterTestNode3.java
diff --git a/mica-mqtt-broker/src/test/java/org/dromara/mica/mqtt/broker/cluster/MqttClusterIntegrationTest.java b/mica-mqtt-broker/src/test/java/org/dromara/mica/mqtt/broker/cluster/MqttClusterIntegrationTest.java
