feat(mqtt): 增加设备实时数据处理

bsw215583320

2025-02-17 ff34638b445619d83740223514aa4de4a8e9a65f

feat(mqtt): 增加设备实时数据处理

- 在 MqttConstant 中添加新的 MQTT 主题常量
- 在 MqttSampleCallback 中增加对新主题的消息处理逻辑
- 引入 RealTimeDataParentVo 类用于解析设备实时数据
- 通过 realTimeDataService 处理实时数据

 jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/MqttConstant.java

@@ -77,6 +77,7 @@

  String TENANT_UP_PREFIX = "tenant/up";
  String TENANT_UP_PREFIX_REALTIME_DATA = TENANT_UP_PREFIX + "/realTime/data";
  String TENANT_UP_PREFIX_REALTIME_DATA_EQP = TENANT_UP_PREFIX + "/realTime/data/eqp";
  String TENANT_UP_PREFIX_FAULT_DATA = TENANT_UP_PREFIX + "/fault/data";
  String TENANT_UP_PREFIX_REAL_FAULT_DATA = TENANT_UP_PREFIX + "/real/fault/data";


 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/mqtt/MqttSampleCallback.java

@@ -24,6 +24,7 @@
import org.jeecg.modules.dry.service.*;
import org.jeecg.modules.dry.vo.DryEquipmentVo;
import org.jeecg.modules.dry.vo.DryFaultRecordVo;
import org.jeecg.modules.dry.vo.RealTimeDataParentVo;
import org.jeecg.modules.dry.vo.RealTimeDataVo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
@@ -215,6 +216,17 @@
                });

                break;
            // 接收设备实时数据-机台
            case MqttConstant.TENANT_UP_PREFIX_REALTIME_DATA_EQP:
                ThreadUtil.execute(() -> {
                    try {
                        RealTimeDataParentVo vo = JSON.parseObject(message, RealTimeDataParentVo.class);
                        realTimeDataService.realTimeDataHandle(vo);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });
                break;
            //各租户上传的实时报警数据
            case MqttConstant.TENANT_UP_PREFIX_REAL_FAULT_DATA: {

@@ -243,8 +255,6 @@
                //发送广播
                System.err.println("广播给：" + recTopic);
                sendMqttMessage(MqttConstant.SERVICE_BROADCAST_TENANT_REAL_FAULT, mqMessage, 1);


            }
            break;
            //移动端主动请求设备实时故障数据（用于页面刚打开时拉取一次数据）
@@ -260,8 +270,6 @@
                                entry -> entry.getKey().toString(),
                                entry -> (DryFaultRecordVo) entry.getValue()
                        ));


                if (dryFaultMap.isEmpty()) {
                    return;
                }