fix 修复bug

xiaoyi

6 天以前 05ab6e3f69a5e73624b2573db043f5923e8f7d8b

 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 是设备名、value 是属性字典，按设备逐台落实时数据；
     * 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);