20260726:mqtt对接

.claude/settings.local.json

@@ -27,7 +27,8 @@
       "Bash(nvm use:*)",
       "Bash(npm run:*)",
       "mcp__filesystem__read_text_file",
-      "Bash(sips:*)"
+      "Bash(sips:*)",
+      "Bash(mysql -h localhost -u root -pfastbee fastbee -e \"SELECT id, name, config_json FROM bridge WHERE id = 9;\")"
     ]
   }
 }

bin/script/send_modbus_and_register.py

@@ -12,6 +12,9 @@ MODBUS_DATA_HEX = (
    "01030400110018AA3C"
 )
 
+# 重要-四川远德MQTT推送
+#  "7e803042593632517e",
+#  "01033041011b7b0000000041eaafb742d19bd74513d545450dd9a4450ec53d00000000420ccf64420c74a9420c74a9c2d200009d99"
 # 深部位移计
 # "7e80DD33D43DFC5B7e",
 #   "0103204135317841ab8e39bf27bccd3f1fb19ac2b231bbc0b6f82040ae3207c3f9f80efa23",

docs/scripts/data_bridge_script.groovy

@@ -0,0 +1,138 @@
+// 数据桥接脚本 - FastBee物模型格式转第三方平台格式
+// 输入格式：[{"message":[{"id":"物模型标识","value":"值"...}],"sources":"..."}]
+// 输出格式：{"设备编号":{"节点标识":{"字段":值}}}
+
+import cn.hutool.json.JSONArray;
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
+import cn.hutool.core.util.NumberUtil;
+
+// 在Groovy脚本中，msgContext对象直接可用
+// 获取上下文信息
+String topic = msgContext.getTopic();
+String payload = msgContext.getPayload();
+String serialNumber = msgContext.getSerialNumber();
+
+msgContext.logger.info("数据桥接 - 接收到FastBee数据: 设备={}, payload={}", serialNumber, payload);
+
+// 数据转换处理
+try {
+    // 直接解析为JSONArray（实际格式是对象数组）
+    JSONArray inputArray = JSONUtil.parseArray(payload);
+
+    if (inputArray == null || inputArray.isEmpty()) {
+        msgContext.logger.warn("数据桥接 - 输入数据为空");
+        return;
+    }
+
+    msgContext.logger.info("数据桥接 - 解析到 {} 个物模型项", inputArray.size());
+
+    // 准备输出数据结构
+    JSONObject outputData = new JSONObject();
+    JSONObject deviceData = new JSONObject();
+
+    // 初始化两个节点容器
+    JSONObject L3_MS_1 = new JSONObject();  // 频率、温度、压力
+    JSONObject S1_ZT_1 = new JSONObject();  // 经纬度、电压
+
+    // 遍历物模型数据，进行字段映射
+    for (int i = 0; i < inputArray.size(); i++) {
+        JSONObject item = inputArray.getJSONObject(i);
+        String identifier = item.getStr("id");
+        Object value = item.get("value");
+
+        if (identifier == null || value == null) {
+            msgContext.logger.debug("跳过空项: {}", item);
+            continue;
+        }
+
+        msgContext.logger.debug("处理物模型项: id={}, value={}", identifier, value);
+
+        switch (identifier) {
+            // 字段映射 - 频率通道 -> L3_MS_1
+            case "F_CH1":
+                L3_MS_1.set("hz1", convertValue(value))
+                break
+            case "F_CH2":
+                L3_MS_1.set("hz2", convertValue(value))
+                break
+            case "F_CH3":
+                L3_MS_1.set("hz3", convertValue(value))
+                break
+            // 字段映射 - 温度通道 -> L3_MS_1
+            case "T_CH1":
+                L3_MS_1.set("wd1", convertValue(value))
+                break
+            case "T_CH2":
+                L3_MS_1.set("wd2", convertValue(value))
+                break
+            case "T_CH3":
+                L3_MS_1.set("wd3", convertValue(value))
+                break
+            // 字段映射 - 压力 -> L3_MS_1
+            case "KN":
+                L3_MS_1.set("value", convertValue(value))
+                break
+            // 字段映射 - 经纬度、电压 -> S1_ZT_1
+            case "Lont":
+                S1_ZT_1.set("lon", convertValue(value))
+                break
+            case "Lat":
+                S1_ZT_1.set("lat", convertValue(value))
+                break
+            case "Vol":
+                S1_ZT_1.set("ext_power_volt", convertValue(value))
+                break
+            default:
+                msgContext.logger.debug("未映射的物模型标识: {}", identifier)
+                break;
+        }
+    }
+
+    // 组装输出数据结构
+    // 只有非空节点才添加到输出
+    if (!L3_MS_1.isEmpty()) {
+        deviceData.set("L3_MS_1", L3_MS_1);
+        msgContext.logger.info("L3_MS_1 节点字段数: {}", L3_MS_1.size());
+    }
+    if (!S1_ZT_1.isEmpty()) {
+        deviceData.set("S1_ZT_1", S1_ZT_1);
+        msgContext.logger.info("S1_ZT_1 节点字段数: {}", S1_ZT_1.size());
+    }
+
+    // 使用上下文中的设备编号
+    if (serialNumber == null) {
+        throw new IllegalArgumentException("数据桥接失败：serialNumber 不能为空");
+    }
+    outputData.set(serialNumber, deviceData);
+
+    // 构建新的payload
+    String newPayload = outputData.toString();
+
+    msgContext.logger.info("数据桥接 - 转换完成: 设备={}, 输出payload={}", serialNumber, newPayload);
+
+    // 返回新的数据
+    msgContext.setTopic(topic);
+    msgContext.setPayload(newPayload);
+
+} catch (Exception e) {
+    msgContext.logger.error("数据桥接 - 转换异常: ", e);
+    // 转换失败时保持原数据
+    msgContext.setTopic(topic);
+    msgContext.setPayload(payload);
+}
+// 执行Action动作参数(脚本由系统自动生成)
+msgContext.setData("mqttBridgeID", 3);
+
+// 辅助方法：转换值为数字类型（保持精度）
+Object convertValue(Object value) {
+    if (value == null) {
+        return null;
+    }
+    try {
+        String strValue = String.valueOf(value);
+        return Double.parseDouble(strValue);
+    } catch (Exception e) {
+        return value;
+    }
+}

fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttBridge.java

@@ -9,6 +9,7 @@ import com.fastbee.ruleEngine.context.MsgContext;
 import com.yomahub.liteflow.annotation.LiteflowComponent;
 import com.yomahub.liteflow.core.NodeComponent;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
@@ -35,16 +36,28 @@ public class MqttBridge extends NodeComponent {
         if (bridge != null) {
             MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class);
             MqttAsyncClient client = MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config), null);
+
+            if (client == null) {
+                log.error("=>MQTT客户端创建失败: bridgeId={}", id);
+                return;
+            }
+
+            // 优先使用bridge.route，如果为空再使用context.topic
+            String publishTopic = StringUtils.isNotBlank(bridge.getRoute())
+                    ? bridge.getRoute()
+                    : cxt.getTopic();
+
             MqttMessage message = new MqttMessage();
             message.setPayload(cxt.getPayload().getBytes());
+
             try {
-                client.publish(cxt.getTopic(), message);
+                client.publish(publishTopic, message);
-                log.info("=>桥接发布主题成功 topic={},message={}", cxt.getTopic(), message);
+                log.info("=>桥接发布主题成功: bridgeId={}, topic={}, payload={}", id, publishTopic, cxt.getPayload());
             } catch (MqttException e) {
-                log.error("msg:" + e.getMessage());
+                log.error("msg:{}", e.getMessage());
-                log.error("cause:" + e.getCause());
+                log.error("cause:{}", String.valueOf(e.getCause()));
-                log.error("reason:" + e.getReasonCode());
+                log.error("reason:{}", e.getReasonCode());
-                log.error("=>桥接发布主题时发生错误 topic={},message={}", cxt.getTopic(), e.getMessage());
+                log.error("=>桥接发布主题失败: bridgeId={}, topic={}, error={}", id, publishTopic, e.getMessage());
             }
         }
     }

fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttClientFactory.java

@@ -14,50 +14,119 @@ public class MqttClientFactory {
 
     private static final Map<String, MqttAsyncClient> CLIENT_CACHE = new ConcurrentHashMap<>();
     private static final Map<String, MqttConnectOptions> CLIENT_OPTIONS_CACHE = new ConcurrentHashMap<>();
+    // 每个key独立的锁对象，避免同一bridge并发创建/重连时产生连接泄漏
+    private static final Map<String, Object> KEY_LOCKS = new ConcurrentHashMap<>();
 
     public static MqttAsyncClient instance(MqttClientConfig config, MqttCallbackonSuccess callback) throws MqttException {
         if (config.getKey() == null) {
+            log.error("=>MQTT配置key为空");
             return null;
         }
+
+        Object lock = KEY_LOCKS.computeIfAbsent(config.getKey(), k -> new Object());
+        synchronized (lock) {
             MqttAsyncClient client = CLIENT_CACHE.get(config.getKey());
             if (client == null) {
-            log.debug("=>mqtt客户端：{} 不存在", config.getKey());
+                log.info("=>MQTT客户端不存在，创建新客户端: {}", config.getKey());
                 client = instanceNew(config, callback);
+                if (client != null && client.isConnected()) {
                     CLIENT_CACHE.put(config.getKey(), client);
-        } else {
-            log.debug("=>mqtt客户端：{} 存在", config.getKey());
-            MqttConnectOptions options = CLIENT_OPTIONS_CACHE.get(config.getKey());
-            if (options != null && Objects.equals(options.getUserName(), config.getUsername()) && new String(options.getPassword()).equals(config.getPassword())) {
-                if (!client.isConnected()) {
-                    client.reconnect();
                 }
             } else {
-                log.debug("=>mqtt客户端：{} 配置不一致，重建", config.getKey());
+                log.debug("=>MQTT客户端已存在: {}", config.getKey());
+
+                // 检查配置是否一致
+                MqttConnectOptions options = CLIENT_OPTIONS_CACHE.get(config.getKey());
+                boolean configChanged = options == null
+                        || !Objects.equals(options.getUserName(), config.getUsername())
+                        || !passwordEquals(options.getPassword(), config.getPassword());
+
+                if (configChanged) {
+                    log.warn("=>MQTT配置已变化，重新创建客户端: {}", config.getKey());
+                    // 关闭旧连接
+                    try {
+                        if (client.isConnected()) {
+                            client.disconnect();
+                        }
+                        client.close();
+                    } catch (MqttException e) {
+                        log.warn("=>关闭旧客户端失败: {}", e.getMessage());
+                    }
+                    // 创建新客户端
                     client = instanceNew(config, callback);
+                    if (client != null && client.isConnected()) {
                         CLIENT_CACHE.put(config.getKey(), client);
                     }
+                } else if (!client.isConnected()) {
+                    log.warn("=>MQTT客户端未连接，尝试重连: {}", config.getKey());
+                    try {
+                        client.reconnect();
+                        // 等待重连完成（最多10秒）
+                        int waitCount = 0;
+                        while (!client.isConnected() && waitCount < 100) {
+                            Thread.sleep(100);
+                            waitCount++;
+                        }
+                        if (client.isConnected()) {
+                            log.info("=>MQTT客户端重连成功: {}", config.getKey());
+                        } else {
+                            log.error("=>MQTT客户端重连超时: {}", config.getKey());
+                            return null;
+                        }
+                    } catch (Exception e) {
+                        log.error("=>MQTT客户端重连失败: {}", e.getMessage());
+                        return null;
+                    }
+                }
             }
             return client;
         }
+    }
+
+    private static boolean passwordEquals(char[] cached, String configured) {
+        String cachedStr = cached == null ? null : new String(cached);
+        return Objects.equals(cachedStr, configured);
+    }
 
     public static MqttAsyncClient instanceNew(MqttClientConfig config, MqttCallbackonSuccess callback) {
-        MqttAsyncClient client = null;
+        MqttAsyncClient client;
-        log.debug("=>mqtt客户端config：{}", config);
+        log.info("=>创建MQTT客户端: hostUrl={}, clientId={}, username={}", config.getHostUrl(), config.getClientId(), config.getUsername());
         try {
             client = new MqttAsyncClient(config.getHostUrl(), config.getClientId(), new MemoryPersistence());
+
             MqttConnectOptions options = new MqttConnectOptions();
             options.setUserName(config.getUsername());
+            if (config.getPassword() != null) {
                 options.setPassword(config.getPassword().toCharArray());
-            options.setCleanSession(false);
+            }
-            options.setAutomaticReconnect(true);
+
+            // 关键配置：避免会话冲突和超时
+            options.setCleanSession(true);              // 清除会话，避免ClientId冲突
+            options.setConnectionTimeout(30);           // 连接超时30秒
+            options.setKeepAliveInterval(60);           // 保活间隔60秒
+            options.setAutomaticReconnect(true);        // 自动重连
+            options.setMaxReconnectDelay(60000);        // 最大重连延迟60秒
+
             if (callback != null) {
                 callback.setClient(client);
             }
+
+            log.debug("=>开始连接MQTT Broker...");
             IMqttToken token = client.connect(options, null, callback);
-            token.waitForCompletion();
+
+            // 等待连接完成，设置30秒超时
+            token.waitForCompletion(30000);
+            log.info("=>MQTT客户端连接成功: {}", config.getKey());
+
             CLIENT_OPTIONS_CACHE.put(config.getKey(), options);
+
         } catch (MqttException e) {
-            log.error("=>mqtt客户端创建错误，原因：{}", e.getMessage());
+            log.error("=>MQTT客户端创建错误: reason={}, msg={}", e.getReasonCode(), e.getMessage());
+            // 连接失败时返回null，不要返回未连接的客户端
+            return null;
+        } catch (Exception e) {
+            log.error("=>MQTT客户端创建异常: {}", e.getMessage(), e);
+            return null;
         }
         return client;
     }
@@ -70,7 +139,7 @@ public class MqttClientFactory {
                 client.subscribe(topic, 0, new MessageArrivedHandler(topic));
             }
         } catch (MqttException e) {
-            e.printStackTrace();
+            log.error("=>MQTT订阅失败: key={}, topic={}, error={}", key, topic, e.getMessage());
         }
     }
 
@@ -78,7 +147,7 @@ public class MqttClientFactory {
         try {
             client.subscribe(topic, 0, new MessageArrivedHandler(topic));
         } catch (MqttException e) {
-            e.printStackTrace();
+            log.error("=>MQTT订阅失败: topic={}, error={}", topic, e.getMessage());
         }
     }
 
@@ -89,7 +158,7 @@ public class MqttClientFactory {
                 client.unsubscribe(topic);
             }
         } catch (MqttException e) {
-            e.printStackTrace();
+            log.error("=>MQTT取消订阅失败: key={}, topic={}, error={}", key, topic, e.getMessage());
         }
     }
 
@@ -104,6 +173,9 @@ public class MqttClientFactory {
                 CLIENT_CACHE.remove(key);
             } catch (MqttException e) {
                 throw new RuntimeException(e);
+            } finally {
+                CLIENT_OPTIONS_CACHE.remove(key);
+                KEY_LOCKS.remove(key);
             }
         }
     }

fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/BridgeServiceImpl.java

@@ -18,6 +18,7 @@ import com.fastbee.http.service.SuccessCondition;
 import com.fastbee.iot.domain.HttpClient;
 import com.fastbee.iot.domain.MqttClient;
 import com.fastbee.iot.domain.MultipleDataSource;
+import com.fastbee.iot.model.MqttClientConfig;
 import com.fastbee.iot.ruleEngine.DataSourceService;
 import com.fastbee.iot.ruleEngine.MqttCallbackonSuccess;
 import com.fastbee.iot.ruleEngine.MqttClientFactory;
@@ -65,8 +66,7 @@ public class BridgeServiceImpl extends ServiceImpl<BridgeMapper, Bridge> impleme
                 MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class);
                 if (config != null) {
                     try {
-                        MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config),
+                        MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config), new MqttCallbackonSuccess(bridge));
-                                new MqttCallbackonSuccess(bridge));
                     } catch (MqttException e) {
                         throw new RuntimeException(e);
                     }
@@ -146,8 +146,7 @@ public class BridgeServiceImpl extends ServiceImpl<BridgeMapper, Bridge> impleme
             MqttClient config = JSON.parseObject(add.getConfigJson(), MqttClient.class);
             if (config != null) {
                 try {
-                    MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config),
+                    MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config), new MqttCallbackonSuccess(add));
-                            new MqttCallbackonSuccess(add));
                 } catch (MqttException e) {
                     throw new RuntimeException(e);
                 }
@@ -217,9 +216,9 @@ public class BridgeServiceImpl extends ServiceImpl<BridgeMapper, Bridge> impleme
         } else if (bridge.getType() == 4) {
             MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class);
             if (config != null) {
+                MqttClientConfig clientConfig = mqttClientService.buildmqttclientconfig(config);
                 try {
-                    MqttAsyncClient client = MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config),
+                    MqttAsyncClient client = MqttClientFactory.instance(clientConfig, new MqttCallbackonSuccess(bridge));
-                            new MqttCallbackonSuccess(bridge));
                     if (client != null && client.isConnected()) {
                         bridge.setStatus("1");
                         result.set(1);
@@ -227,11 +226,11 @@ public class BridgeServiceImpl extends ServiceImpl<BridgeMapper, Bridge> impleme
                         bridge.setStatus("0");
                         result.set(0);
                     }
-                    if (client != null) {
-                        client.disconnect().waitForCompletion();
-                    }
                 } catch (MqttException e) {
-                    log.error("=>mqtt客户端创建错误");
+                    log.error("=>MQTT客户端创建错误");
+                } finally {
+                    // 测试连接仅用于校验可用性，不长期占用缓存，避免下次真实发布时命中一个已断开的客户端
+                    MqttClientFactory.remove(clientConfig.getKey());
                 }
             }
         } else if (bridge.getType() == 5) {