From 8eb62ee0efe77e264cd48b3f1fd835e9b8430318 Mon Sep 17 00:00:00 2001
From: xiaoyi <908147524@qq.com>
Date: 星期三, 28 十月 2026 15:02:12 +0800
Subject: [PATCH] fix 设备数采
---
yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/framework/collect/mqtt/MesCollectMqttClient.java | 34 ++++++++++++++++++++++++++++------
1 files changed, 28 insertions(+), 6 deletions(-)
diff --git a/yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/framework/collect/mqtt/MesCollectMqttClient.java b/yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/framework/collect/mqtt/MesCollectMqttClient.java
index 6e20934..6cb6818 100644
--- a/yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/framework/collect/mqtt/MesCollectMqttClient.java
+++ b/yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/framework/collect/mqtt/MesCollectMqttClient.java
@@ -12,6 +12,9 @@
import java.net.InetAddress;
import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -114,15 +117,34 @@
private void subscribe() {
MesCollectProperties.Mqtt mqtt = properties.getMqtt();
- String topic = normalizeTopicPrefix(mqtt.getTopicPrefix()) + "+/+";
- try {
- client.subscribe(topic, mqtt.getQos());
- log.info("[subscribe][鏁伴噰 MQTT 璁㈤槄鎴愬姛锛歿}]", topic);
- } catch (Exception e) {
- log.error("[subscribe][鏁伴噰 MQTT 璁㈤槄澶辫触锛歿}]", topic, e);
+ for (String topic : resolveSubscribeTopics(mqtt)) {
+ try {
+ client.subscribe(topic, mqtt.getQos());
+ log.info("[subscribe][鏁伴噰 MQTT 璁㈤槄鎴愬姛锛歿}]", topic);
+ } catch (Exception e) {
+ log.error("[subscribe][鏁伴噰 MQTT 璁㈤槄澶辫触锛歿}]", topic, e);
+ }
}
}
+ /**
+ * 瑙ｆ瀽闇�瑕佽闃呯殑涓婚銆�
+ *
+ * <p>鏄惧紡閰嶇疆 topics 浼樺厛锛涚綉鍏虫牸寮忚闃呮暣涓墠缂�锛坓ateway/#锛夛紝
+ * 鍥犱负缃戝叧闄� attributes 澶栬繕浼氬彂 connect / disconnect锛屽悗缁篃鍙兘鏂板涓婚锛�
+ * 鍏朵綑鏍煎紡娌跨敤 /YLCY/IMM/+/+ 涓ょ骇閫氶厤銆�</p>
+ */
+ private List<String> resolveSubscribeTopics(MesCollectProperties.Mqtt mqtt) {
+ if (StrUtil.isNotBlank(mqtt.getTopics())) {
+ return Arrays.stream(mqtt.getTopics().split(","))
+ .map(String::trim)
+ .filter(StrUtil::isNotBlank)
+ .toList();
+ }
+ String prefix = normalizeTopicPrefix(mqtt.getTopicPrefix());
+ return Collections.singletonList(mqtt.isGatewayFormat() ? prefix + "#" : prefix + "+/+");
+ }
+
private String normalizeTopicPrefix(String prefix) {
if (StrUtil.isBlank(prefix)) {
return "";
--
Gitblit v1.9.3