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/service/collect/MesCollectMessageDispatcher.java | 75 +++++++++++++++++++++++++++++++++++++
1 files changed, 75 insertions(+), 0 deletions(-)
diff --git a/yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/service/collect/MesCollectMessageDispatcher.java b/yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/service/collect/MesCollectMessageDispatcher.java
index 5b1a468..5d58a8e 100644
--- a/yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/service/collect/MesCollectMessageDispatcher.java
+++ b/yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/service/collect/MesCollectMessageDispatcher.java
@@ -13,6 +13,7 @@
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
+import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
@@ -26,6 +27,11 @@
@Component
@Slf4j
public class MesCollectMessageDispatcher {
+
+ /**
+ * 缃戝叧鏍煎紡閲屾壙杞借澶囧睘鎬у揩鐓х殑涓婚鏈
+ */
+ private static final String ATTR_TOPIC_SUFFIX = "attributes";
@Resource
private MesCollectProperties properties;
@@ -61,6 +67,11 @@
}
private void process(String topic, String payload) {
+ // 缃戝叧鏍煎紡鎶ユ枃缁撴瀯瀹屽叏涓嶅悓锛堜富棰樺浐瀹氥�佽澶囦互鎶ユ枃棣栧眰 key 鍖哄垎锛夛紝鍗曠嫭澶勭悊
+ if (properties.getMqtt().isGatewayFormat()) {
+ processGateway(topic, payload);
+ return;
+ }
JsonNode root;
try {
root = parser.parsePayload(payload);
@@ -111,6 +122,70 @@
}
}
+ /**
+ * 缃戝叧鏍煎紡鎶ユ枃澶勭悊銆�
+ *
+ * <p>attributes锛氭姤鏂囬灞� key 鏄澶囧悕銆乿alue 鏄睘鎬у瓧鍏革紝鎸夎澶囬�愬彴钀藉疄鏃舵暟鎹紱
+ * connect / disconnect锛氳繛鎺ヤ簨浠讹紝鍙暀鐥曚笉钀藉疄鏃舵暟鎹紝渚夸簬鎺掓煡璁惧绂荤嚎銆�</p>
+ */
+ private void processGateway(String topic, String payload) {
+ JsonNode root;
+ try {
+ root = parser.parsePayload(payload);
+ } catch (Exception e) {
+ log.warn("[processGateway][涓婚({}) 鎶ユ枃涓嶆槸鍚堟硶 JSON锛歿}]", topic, StrUtil.maxLength(payload, 200));
+ dataService.markRawMessageFailed(
+ dataService.saveRawMessage(null, null, topic, MesCollectMsgTypeEnum.UNKNOWN,
+ LocalDateTime.now(), payload),
+ "鎶ユ枃涓嶆槸鍚堟硶 JSON锛�" + e.getMessage());
+ return;
+ }
+
+ // 杩炴帴浜嬩欢锛歿"device":"澶ф媺"}锛屾病鏈夊睘鎬у瓧鍏�
+ if (!ATTR_TOPIC_SUFFIX.equalsIgnoreCase(parser.extractTopicSuffix(topic))) {
+ String deviceCode = StrUtil.trim(parser.getText(root, "device"));
+ MesDvMachineryDO machinery = StrUtil.isBlank(deviceCode) ? null
+ : machineryMapper.selectByCollectDeviceCode(deviceCode);
+ dataService.saveRawMessage(deviceCode, machinery == null ? null : machinery.getId(), topic,
+ MesCollectMsgTypeEnum.getByType(parser.extractTopicSuffix(topic)),
+ LocalDateTime.now(), payload);
+ return;
+ }
+
+ Map<String, JsonNode> devices = parser.splitGatewayDevices(root);
+ if (devices.isEmpty()) {
+ log.debug("[processGateway][涓婚({}) 鎶ユ枃閲屾病鏈夎澶囧睘鎬э細{}]", topic, StrUtil.maxLength(payload, 200));
+ dataService.saveRawMessage(null, null, topic, MesCollectMsgTypeEnum.UNKNOWN, LocalDateTime.now(), payload);
+ return;
+ }
+ // 缃戝叧鎶ユ枃涓嶅甫鏃堕棿鎴筹紝缁熶竴鎸夋帴鏀舵椂闂磋惤搴�
+ LocalDateTime eventTime = parser.parseEventTime(root);
+ devices.forEach((deviceCode, data) -> {
+ MesDvMachineryDO machinery = machineryMapper.selectByCollectDeviceCode(deviceCode);
+ if (machinery == null) {
+ warnUnboundDevice(deviceCode);
+ }
+ Long machineryId = machinery == null ? null : machinery.getId();
+ Long taskId = null;
+ Long workOrderId = null;
+ if (machinery != null && machinery.getWorkstationId() != null) {
+ MesProTaskDO task = taskResolver.resolve(machinery.getWorkstationId());
+ if (task != null) {
+ taskId = task.getId();
+ workOrderId = task.getWorkOrderId();
+ }
+ }
+ Long rawId = dataService.saveRawMessage(deviceCode, machineryId, topic,
+ MesCollectMsgTypeEnum.REALTIME, eventTime, payload);
+ try {
+ dataService.saveRealtime(deviceCode, machineryId, taskId, workOrderId, eventTime, data);
+ } catch (Exception e) {
+ dataService.markRawMessageFailed(rawId, e.getMessage());
+ log.error("[processGateway][涓婚({}) 璁惧({}) 澶勭悊澶辫触]", topic, deviceCode, e);
+ }
+ });
+ }
+
private void warnUnboundDevice(String deviceCode) {
long now = System.currentTimeMillis();
Long last = unboundWarnCache.get(deviceCode);
--
Gitblit v1.9.3