From ced3257a7999eb6cdf41cb8129e488607da6f64a Mon Sep 17 00:00:00 2001 From: zhugongping Date: Sun, 26 Jul 2026 09:38:34 +0800 Subject: [PATCH] =?UTF-8?q?20260726:mqtt=E5=AF=B9=E6=8E=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .claude/settings.local.json | 3 +- bin/script/send_modbus_and_register.py | 3 + docs/scripts/data_bridge_script.groovy | 138 ++++++++++++++++++ .../fastbee/iot/ruleEngine/MqttBridge.java | 25 +++- .../iot/ruleEngine/MqttClientFactory.java | 122 ++++++++++++---- .../iot/service/impl/BridgeServiceImpl.java | 19 ++- 6 files changed, 268 insertions(+), 42 deletions(-) create mode 100644 docs/scripts/data_bridge_script.groovy diff --git a/.claude/settings.local.json b/.claude/settings.local.json index ee3a2356..fa977bd3 100644 --- a/.claude/settings.local.json +++ b/.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;\")" ] } } diff --git a/bin/script/send_modbus_and_register.py b/bin/script/send_modbus_and_register.py index b9d00023..138d2c46 100644 --- a/bin/script/send_modbus_and_register.py +++ b/bin/script/send_modbus_and_register.py @@ -12,6 +12,9 @@ MODBUS_DATA_HEX = ( "01030400110018AA3C" ) +# 重要-四川远德MQTT推送 +# "7e803042593632517e", +# "01033041011b7b0000000041eaafb742d19bd74513d545450dd9a4450ec53d00000000420ccf64420c74a9420c74a9c2d200009d99" # 深部位移计 # "7e80DD33D43DFC5B7e", # "0103204135317841ab8e39bf27bccd3f1fb19ac2b231bbc0b6f82040ae3207c3f9f80efa23", diff --git a/docs/scripts/data_bridge_script.groovy b/docs/scripts/data_bridge_script.groovy new file mode 100644 index 00000000..b056fa80 --- /dev/null +++ b/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; + } +} \ No newline at end of file diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttBridge.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttBridge.java index 588a19be..214c66b4 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttBridge.java +++ b/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); - log.info("=>桥接发布主题成功 topic={},message={}", cxt.getTopic(), message); + client.publish(publishTopic, message); + log.info("=>桥接发布主题成功: bridgeId={}, topic={}, payload={}", id, publishTopic, cxt.getPayload()); } catch (MqttException e) { - log.error("msg:" + e.getMessage()); - log.error("cause:" + e.getCause()); - log.error("reason:" + e.getReasonCode()); - log.error("=>桥接发布主题时发生错误 topic={},message={}", cxt.getTopic(), e.getMessage()); + log.error("msg:{}", e.getMessage()); + log.error("cause:{}", String.valueOf(e.getCause())); + log.error("reason:{}", e.getReasonCode()); + log.error("=>桥接发布主题失败: bridgeId={}, topic={}, error={}", id, publishTopic, e.getMessage()); } } } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttClientFactory.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttClientFactory.java index d6bd83b9..f159920a 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttClientFactory.java +++ b/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 CLIENT_CACHE = new ConcurrentHashMap<>(); private static final Map CLIENT_OPTIONS_CACHE = new ConcurrentHashMap<>(); + // 每个key独立的锁对象，避免同一bridge并发创建/重连时产生连接泄漏 + private static final Map KEY_LOCKS = new ConcurrentHashMap<>(); public static MqttAsyncClient instance(MqttClientConfig config, MqttCallbackonSuccess callback) throws MqttException { if (config.getKey() == null) { + log.error("=>MQTT配置key为空"); return null; } - MqttAsyncClient client = CLIENT_CACHE.get(config.getKey()); - if (client == null) { - log.debug("=>mqtt客户端：{} 不存在", config.getKey()); - client = instanceNew(config, callback); - 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(); + + Object lock = KEY_LOCKS.computeIfAbsent(config.getKey(), k -> new Object()); + synchronized (lock) { + MqttAsyncClient client = CLIENT_CACHE.get(config.getKey()); + if (client == null) { + 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()); - client = instanceNew(config, callback); - CLIENT_CACHE.put(config.getKey(), client); + 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; } - 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; - log.debug("=>mqtt客户端config：{}", config); + MqttAsyncClient client; + 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()); - options.setPassword(config.getPassword().toCharArray()); - options.setCleanSession(false); - options.setAutomaticReconnect(true); + if (config.getPassword() != null) { + options.setPassword(config.getPassword().toCharArray()); + } + + // 关键配置：避免会话冲突和超时 + 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); } } } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/BridgeServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/BridgeServiceImpl.java index a714fc56..0544a443 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/BridgeServiceImpl.java +++ b/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 impleme MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class); if (config != null) { try { - MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config), - new MqttCallbackonSuccess(bridge)); + MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config), new MqttCallbackonSuccess(bridge)); } catch (MqttException e) { throw new RuntimeException(e); } @@ -146,8 +146,7 @@ public class BridgeServiceImpl extends ServiceImpl impleme MqttClient config = JSON.parseObject(add.getConfigJson(), MqttClient.class); if (config != null) { try { - MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config), - new MqttCallbackonSuccess(add)); + MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config), new MqttCallbackonSuccess(add)); } catch (MqttException e) { throw new RuntimeException(e); } @@ -217,9 +216,9 @@ public class BridgeServiceImpl extends ServiceImpl 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), - new MqttCallbackonSuccess(bridge)); + MqttAsyncClient client = MqttClientFactory.instance(clientConfig, new MqttCallbackonSuccess(bridge)); if (client != null && client.isConnected()) { bridge.setStatus("1"); result.set(1); @@ -227,11 +226,11 @@ public class BridgeServiceImpl extends ServiceImpl 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) {