package cn.iocoder.yudao.module.mes.service.collect;
|
|
import cn.hutool.core.util.StrUtil;
|
import cn.iocoder.yudao.module.mes.dal.dataobject.dv.machinery.MesDvMachineryDO;
|
import cn.iocoder.yudao.module.mes.dal.dataobject.pro.task.MesProTaskDO;
|
import cn.iocoder.yudao.module.mes.dal.mysql.dv.machinery.MesDvMachineryMapper;
|
import cn.iocoder.yudao.module.mes.enums.collect.MesCollectMsgTypeEnum;
|
import cn.iocoder.yudao.module.mes.framework.collect.config.MesCollectProperties;
|
import tools.jackson.databind.JsonNode;
|
import jakarta.annotation.PreDestroy;
|
import jakarta.annotation.Resource;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.stereotype.Component;
|
|
import java.time.LocalDateTime;
|
import java.util.Map;
|
import java.util.concurrent.*;
|
import java.util.concurrent.atomic.AtomicInteger;
|
|
/**
|
* MES 数采报文分发器
|
*
|
* <p>Paho 回调线程只负责入队，解析与落库在单线程中串行执行，保证同一设备报警等状态按顺序处理。</p>
|
*
|
* @author 超级管理员
|
*/
|
@Component
|
@Slf4j
|
public class MesCollectMessageDispatcher {
|
|
/**
|
* 网关格式里承载设备属性快照的主题末段
|
*/
|
private static final String ATTR_TOPIC_SUFFIX = "attributes";
|
|
@Resource
|
private MesCollectProperties properties;
|
@Resource
|
private MesCollectDataParser parser;
|
@Resource
|
private MesCollectDataService dataService;
|
@Resource
|
private MesDvMachineryMapper machineryMapper;
|
@Resource
|
private MesCollectTaskResolver taskResolver;
|
|
private final ExecutorService executor = new ThreadPoolExecutor(1, 1,
|
0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(10_000),
|
new ThreadFactory() {
|
private final AtomicInteger index = new AtomicInteger();
|
|
@Override
|
public Thread newThread(Runnable runnable) {
|
Thread thread = new Thread(runnable, "mes-collect-dispatch-" + index.incrementAndGet());
|
thread.setDaemon(true);
|
return thread;
|
}
|
}, new ThreadPoolExecutor.CallerRunsPolicy());
|
|
/**
|
* 设备未绑定台账时的告警缓存，避免每条报文重复刷日志
|
*/
|
private final ConcurrentHashMap<String, Long> unboundWarnCache = new ConcurrentHashMap<>();
|
|
public void dispatch(String topic, String payload) {
|
executor.execute(() -> process(topic, payload));
|
}
|
|
private void process(String topic, String payload) {
|
// 网关格式报文结构完全不同（主题固定、设备以报文首层 key 区分），单独处理
|
if (properties.getMqtt().isGatewayFormat()) {
|
processGateway(topic, payload);
|
return;
|
}
|
JsonNode root;
|
try {
|
root = parser.parsePayload(payload);
|
} catch (Exception e) {
|
log.warn("[process][主题({}) 报文不是合法 JSON：{}]", topic, StrUtil.maxLength(payload, 200));
|
dataService.markRawMessageFailed(
|
dataService.saveRawMessage(null, null, topic, MesCollectMsgTypeEnum.UNKNOWN,
|
LocalDateTime.now(), payload),
|
"报文不是合法 JSON：" + e.getMessage());
|
return;
|
}
|
|
JsonNode data = parser.parseData(root);
|
MesCollectMsgTypeEnum msgType = parser.resolveMsgType(parser.getText(root, "topic"), topic, data,
|
properties.getMqtt().getAlarmTopicSuffix());
|
String deviceCode = parser.resolveDeviceId(root, topic);
|
LocalDateTime eventTime = parser.parseEventTime(root);
|
|
MesDvMachineryDO machinery = StrUtil.isBlank(deviceCode) ? null
|
: machineryMapper.selectByCollectDeviceCode(deviceCode);
|
if (machinery == null && StrUtil.isNotBlank(deviceCode)) {
|
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, msgType, eventTime, payload);
|
try {
|
switch (msgType) {
|
case REALTIME -> dataService.saveRealtime(deviceCode, machineryId, taskId, workOrderId, eventTime, data);
|
case SPC -> dataService.saveCycle(deviceCode, machineryId, taskId, workOrderId, eventTime, data);
|
case TECH -> dataService.saveTech(deviceCode, machineryId, taskId, workOrderId, eventTime, data);
|
case OP_LOG -> dataService.saveOperateLog(deviceCode, machineryId, taskId, workOrderId, eventTime, data);
|
case ALARM -> dataService.saveAlarm(deviceCode, machineryId, taskId, workOrderId, eventTime, data);
|
default -> log.debug("[process][忽略未知类型报文，主题({})]", topic);
|
}
|
} catch (Exception e) {
|
dataService.markRawMessageFailed(rawId, e.getMessage());
|
log.error("[process][主题({}) 设备({}) 处理失败]", topic, deviceCode, e);
|
}
|
}
|
|
/**
|
* 网关格式报文处理。
|
*
|
* <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);
|
if (last != null && now - last < 60_000L) {
|
return;
|
}
|
unboundWarnCache.put(deviceCode, now);
|
log.warn("[process][数采设备({}) 未绑定任何设备台账，数据将按设备编号落库，machineryId 为空]", deviceCode);
|
}
|
|
@PreDestroy
|
public void destroy() {
|
executor.shutdownNow();
|
}
|
|
}
|
