|
|
@@ -0,0 +1,988 @@
|
|
|
+package com.usky.cdi.service.impl;
|
|
|
+
|
|
|
+import com.alibaba.fastjson.JSON;
|
|
|
+import com.alibaba.fastjson.JSONObject;
|
|
|
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
|
|
+import com.usky.cdi.domain.DmpDevice;
|
|
|
+import com.usky.cdi.domain.DmpProduct;
|
|
|
+import com.usky.cdi.mapper.DmpDeviceMapper;
|
|
|
+import com.usky.cdi.mapper.DmpProductMapper;
|
|
|
+import com.usky.cdi.service.config.mqtt.MqttBaseConfig;
|
|
|
+import com.usky.cdi.service.config.mqtt.MqttOutConfig;
|
|
|
+import com.usky.cdi.service.enums.EnvMonitorMqttTopic;
|
|
|
+import com.usky.cdi.service.util.DeviceDataQuery;
|
|
|
+import com.usky.cdi.service.util.SnowflakeIdGenerator;
|
|
|
+import com.usky.cdi.service.vo.IotDataTransferVO;
|
|
|
+import com.usky.cdi.service.vo.base.*;
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
|
|
|
+import org.springframework.beans.factory.annotation.Autowired;
|
|
|
+import org.springframework.beans.factory.annotation.Value;
|
|
|
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
|
|
|
+import org.springframework.context.ApplicationContext;
|
|
|
+import org.springframework.context.ConfigurableApplicationContext;
|
|
|
+import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
|
|
|
+import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
|
|
|
+import org.springframework.messaging.MessageChannel;
|
|
|
+import org.springframework.stereotype.Service;
|
|
|
+import org.springframework.web.context.ContextLoader;
|
|
|
+
|
|
|
+import javax.annotation.PostConstruct;
|
|
|
+
|
|
|
+import java.time.Instant;
|
|
|
+import java.time.LocalDateTime;
|
|
|
+import java.time.ZoneId;
|
|
|
+import java.util.ArrayList;
|
|
|
+import java.util.HashMap;
|
|
|
+import java.util.List;
|
|
|
+import java.util.Map;
|
|
|
+import java.util.stream.Collectors;
|
|
|
+
|
|
|
+/**
|
|
|
+ * @author fyc
|
|
|
+ * @email yuchuan.fu@chinausky.com
|
|
|
+ * @date 2025/11/20
|
|
|
+ */
|
|
|
+@Slf4j
|
|
|
+@Service
|
|
|
+public class IotDataTransferService {
|
|
|
+
|
|
|
+ private MqttOutConfig.MqttGateway mqttGateway;
|
|
|
+
|
|
|
+ // 注入ApplicationContext,确保总是能获取到
|
|
|
+ @Autowired
|
|
|
+ private ApplicationContext context;
|
|
|
+
|
|
|
+ // MQTT连接相关配置
|
|
|
+ private static final String MQTT_URL = "ssl://114.80.201.143:8883";
|
|
|
+ private static final String MQTT_TOPIC = "iotInfo/+";
|
|
|
+ private static final int KEEP_ALIVE_INTERVAL = 60;
|
|
|
+ private static final int COMPLETION_TIMEOUT = 5000;
|
|
|
+
|
|
|
+ private SnowflakeIdGenerator idGenerator;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private DeviceDataQuery deviceDataQuery;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private DmpDeviceMapper dmpDeviceMapper;
|
|
|
+
|
|
|
+ @Value("${device.data.simulation}")
|
|
|
+ private boolean simulation;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private DmpProductMapper dmpProductMapper;
|
|
|
+
|
|
|
+ // 从配置文件读取Snowflake参数,默认值为2
|
|
|
+ @Value("${snowflake.worker-id:2}")
|
|
|
+ private long workerId;
|
|
|
+
|
|
|
+ @Value("${snowflake.data-center-id:2}")
|
|
|
+ private long dataCenterId;
|
|
|
+
|
|
|
+ @PostConstruct
|
|
|
+ public void init() {
|
|
|
+ this.idGenerator = new SnowflakeIdGenerator(workerId, dataCenterId);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 获取当前时间
|
|
|
+ */
|
|
|
+ private LocalDateTime getCurrentTime() {
|
|
|
+ return LocalDateTime.now();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 生成数据包ID
|
|
|
+ */
|
|
|
+ private Long generateDataPacketID() {
|
|
|
+ return idGenerator.nextPacketId();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 发送水浸状态(deviceType:702)
|
|
|
+ * Topic: iotInfo/flooded
|
|
|
+ *
|
|
|
+ * @return 推送结果,包含成功数和失败数
|
|
|
+ */
|
|
|
+ public Map<String, Integer> sendWaterLeak(IotDataTransferVO transferVO) {
|
|
|
+ Map<String, Integer> result = new HashMap<>();
|
|
|
+ result.put("successCount", 0);
|
|
|
+ result.put("failureCount", 0);
|
|
|
+
|
|
|
+ if (!validateMqttGateway()) {
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ try {
|
|
|
+ List<JSONObject> deviceData = deviceDataQuery.getDeviceData(transferVO);
|
|
|
+ log.warn("获取到的数据:{}", deviceData);
|
|
|
+ Integer deviceType = transferVO.getDeviceType();
|
|
|
+ Integer totalDevices = transferVO.getDevices().size();
|
|
|
+
|
|
|
+ log.info("开始推送水浸状态数据,设备类型:{},设备数量:{},获取到的数据条数:{}",
|
|
|
+ deviceType, totalDevices, deviceData.size());
|
|
|
+
|
|
|
+ if (deviceData.isEmpty()) {
|
|
|
+ log.warn("没有获取到水浸数据!设备类型:{}", deviceType);
|
|
|
+ result.put("failureCount", totalDevices);
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ Long engineeringId = transferVO.getEngineeringId();
|
|
|
+ for (JSONObject deviceDataItem : deviceData) {
|
|
|
+ LocalDateTime dataEndTime = parseDataTime(deviceDataItem);
|
|
|
+ if (dataEndTime == null) {
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ Integer deviceId = deviceDataItem.getIntValue("device_id");
|
|
|
+ Integer value = deviceDataItem.getInteger("leach_status");
|
|
|
+ if (value == null) {
|
|
|
+ log.warn("设备{}的水浸状态数据为空", deviceId);
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ WaterLeakVO vo = new WaterLeakVO();
|
|
|
+ vo.setDataPacketID(generateDataPacketID());
|
|
|
+ vo.setSensorID(deviceId);
|
|
|
+ vo.setEngineeringID(engineeringId);
|
|
|
+ vo.setPublishTime(getCurrentTime());
|
|
|
+ vo.setSensorValue(value);
|
|
|
+ vo.setDataEndTime(dataEndTime);
|
|
|
+
|
|
|
+ try {
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.WATER_LEAK, vo, "水浸状态信息");
|
|
|
+ result.put("successCount", result.get("successCount") + 1);
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.warn("设备{}的水浸状态数据推送失败:{}", deviceId, e.getMessage());
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ log.info("水浸状态数据推送完成,设备类型:{},成功:{},失败:{}",
|
|
|
+ deviceType, result.get("successCount"), result.get("failureCount"));
|
|
|
+
|
|
|
+ return result;
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("水浸状态数据推送发生异常", e);
|
|
|
+ result.put("failureCount", transferVO.getDevices().size());
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 发送温湿度及气体浓度数据(设备类型:701,707-711)
|
|
|
+ * 包含: wd(温度), sd(湿度), o2(氧气), co(一氧化碳), co2(二氧化碳)
|
|
|
+ *
|
|
|
+ * @return 推送结果,包含成功数和失败数
|
|
|
+ */
|
|
|
+ public Map<String, Integer> sendEnvData(IotDataTransferVO transferVO) {
|
|
|
+ Map<String, Integer> result = new HashMap<>();
|
|
|
+ result.put("successCount", 0);
|
|
|
+ result.put("failureCount", 0);
|
|
|
+
|
|
|
+ if (!validateMqttGateway()) {
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ try {
|
|
|
+ Integer deviceType = transferVO.getDeviceType();
|
|
|
+ List<JSONObject> deviceData = deviceDataQuery.getDeviceData(transferVO);
|
|
|
+ Integer totalDevices = transferVO.getDevices().size();
|
|
|
+
|
|
|
+ // 统计各类型设备数据数量
|
|
|
+ int tempCount = 0, humidityCount = 0, coCount = 0, o2Count = 0, co2Count = 0;
|
|
|
+ for (JSONObject dataItem : deviceData) {
|
|
|
+ if (dataItem.containsKey("wd") && dataItem.getFloat("wd") != null) tempCount++;
|
|
|
+ if (dataItem.containsKey("sd") && dataItem.getFloat("sd") != null) humidityCount++;
|
|
|
+ if (dataItem.containsKey("co") && dataItem.getFloat("co") != null) coCount++;
|
|
|
+ if (dataItem.containsKey("o2") && dataItem.getFloat("o2") != null) o2Count++;
|
|
|
+ if (dataItem.containsKey("co2") && dataItem.getFloat("co2") != null) co2Count++;
|
|
|
+ }
|
|
|
+
|
|
|
+ log.info("开始推送温湿度及气体浓度数据,设备类型:{},设备数量:{},获取到的数据条数:{}",
|
|
|
+ deviceType, totalDevices, deviceData.size());
|
|
|
+ log.info("各类型设备数据数量:温度{}条,湿度{}条,一氧化碳{}条,氧气{}条,二氧化碳{}条",
|
|
|
+ tempCount, humidityCount, coCount, o2Count, co2Count);
|
|
|
+
|
|
|
+ if (deviceData.isEmpty()) {
|
|
|
+ log.warn("没有获取到空气质量数据!设备类型:{}", deviceType);
|
|
|
+ result.put("failureCount", totalDevices);
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ Long engineeringId = transferVO.getEngineeringId();
|
|
|
+ for (JSONObject deviceDataItem : deviceData) {
|
|
|
+ LocalDateTime dataEndTime = parseDataTime(deviceDataItem);
|
|
|
+ if (dataEndTime == null) {
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ Integer deviceId = deviceDataItem.getIntValue("device_id");
|
|
|
+ boolean deviceSuccess = true;
|
|
|
+
|
|
|
+ // 根据设备类型发送对应的数据
|
|
|
+ try {
|
|
|
+ switch (deviceType) {
|
|
|
+ case 707:
|
|
|
+ sendTempData(deviceId, dataEndTime, deviceDataItem, engineeringId);
|
|
|
+ break;
|
|
|
+ case 708:
|
|
|
+ sendHumidityData(deviceId, dataEndTime, deviceDataItem, engineeringId);
|
|
|
+ break;
|
|
|
+ case 709:
|
|
|
+ sendOxygenData(deviceId, dataEndTime, deviceDataItem, engineeringId);
|
|
|
+ break;
|
|
|
+ case 710:
|
|
|
+ sendCo2Data(deviceId, dataEndTime, deviceDataItem, engineeringId);
|
|
|
+ break;
|
|
|
+ case 711:
|
|
|
+ sendCoData(deviceId, dataEndTime, deviceDataItem, engineeringId);
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.warn("设备{}的环境数据推送失败:{}", deviceId, e.getMessage());
|
|
|
+ deviceSuccess = false;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 统计设备推送结果
|
|
|
+ if (deviceSuccess) {
|
|
|
+ result.put("successCount", result.get("successCount") + 1);
|
|
|
+ } else {
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ log.info("温湿度及气体浓度数据推送完成,设备类型:{},成功:{},失败:{}",
|
|
|
+ deviceType, result.get("successCount"), result.get("failureCount"));
|
|
|
+
|
|
|
+ return result;
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("温湿度及气体浓度数据推送发生异常", e);
|
|
|
+ result.put("failureCount", transferVO.getDevices().size());
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 发送人员闯入情况(703)
|
|
|
+ *
|
|
|
+ * @return 推送结果,包含成功数和失败数
|
|
|
+ **/
|
|
|
+ public Map<String, Integer> sendPersonPresence(IotDataTransferVO transferVO) {
|
|
|
+ Map<String, Integer> result = new HashMap<>();
|
|
|
+ result.put("successCount", 0);
|
|
|
+ result.put("failureCount", 0);
|
|
|
+
|
|
|
+ if (!validateMqttGateway()) {
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ try {
|
|
|
+ List<JSONObject> deviceData = deviceDataQuery.getDeviceData(transferVO);
|
|
|
+ Integer deviceType = transferVO.getDeviceType();
|
|
|
+ Integer totalDevices = transferVO.getDevices().size();
|
|
|
+
|
|
|
+ log.info("开始推送人员闯入情况数据,设备类型:{},设备数量:{},获取到的数据条数:{}",
|
|
|
+ deviceType, totalDevices, deviceData.size());
|
|
|
+
|
|
|
+ if (deviceData.isEmpty()) {
|
|
|
+ log.warn("没有获取到人员闯入情况数据!设备类型:{}", deviceType);
|
|
|
+ result.put("failureCount", totalDevices);
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ Long engineeringId = transferVO.getEngineeringId();
|
|
|
+ for (JSONObject deviceDataItem : deviceData) {
|
|
|
+ LocalDateTime dataEndTime = parseDataTime(deviceDataItem);
|
|
|
+ if (dataEndTime == null) {
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ Integer deviceId = deviceDataItem.getIntValue("device_id");
|
|
|
+
|
|
|
+ PersonPresenceVO vo = new PersonPresenceVO();
|
|
|
+ vo.setDataPacketID(generateDataPacketID());
|
|
|
+ vo.setSensorID(deviceId);
|
|
|
+ vo.setDataEndTime(dataEndTime);
|
|
|
+ vo.setPublishTime(getCurrentTime());
|
|
|
+ vo.setEngineeringID(engineeringId);
|
|
|
+ // 传感器值固定为0(可能是默认值或占位符)
|
|
|
+ vo.setSensorValue(0);
|
|
|
+
|
|
|
+ try {
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.PERSON_PRESENCE, vo, "人员闯入情况");
|
|
|
+ result.put("successCount", result.get("successCount") + 1);
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.warn("设备{}的人员闯入情况数据推送失败:{}", deviceId, e.getMessage());
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ log.info("人员闯入情况数据推送完成,设备类型:{},成功:{},失败:{}",
|
|
|
+ deviceType, result.get("successCount"), result.get("failureCount"));
|
|
|
+
|
|
|
+ return result;
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("人员闯入情况数据推送发生异常", e);
|
|
|
+ result.put("failureCount", transferVO.getDevices().size());
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 发送人防用电负荷情况(704)
|
|
|
+ *
|
|
|
+ * @return 推送结果,包含成功数和失败数
|
|
|
+ **/
|
|
|
+ public Map<String, Integer> sendElectricityLoad(IotDataTransferVO transferVO) {
|
|
|
+ Map<String, Integer> result = new HashMap<>();
|
|
|
+ result.put("successCount", 0);
|
|
|
+ result.put("failureCount", 0);
|
|
|
+
|
|
|
+ if (!validateMqttGateway()) {
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ try {
|
|
|
+ List<JSONObject> deviceData = deviceDataQuery.getDeviceData(transferVO);
|
|
|
+ Integer deviceType = transferVO.getDeviceType();
|
|
|
+ Integer totalDevices = transferVO.getDevices().size();
|
|
|
+
|
|
|
+ log.info("开始推送人防用电负荷情况数据,设备类型:{},设备数量:{},获取到的数据条数:{}",
|
|
|
+ deviceType, totalDevices, deviceData.size());
|
|
|
+
|
|
|
+ if (deviceData.isEmpty()) {
|
|
|
+ log.warn("没有获取到人防用电负荷情况数据!设备类型:{}", deviceType);
|
|
|
+ result.put("failureCount", totalDevices);
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ Long engineeringId = transferVO.getEngineeringId();
|
|
|
+ for (JSONObject deviceDataItem : deviceData) {
|
|
|
+ LocalDateTime dataEndTime = parseDataTime(deviceDataItem);
|
|
|
+ if (dataEndTime == null) {
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ Integer deviceId = deviceDataItem.getIntValue("device_id");
|
|
|
+ ElectricityLoadVO vo = new ElectricityLoadVO();
|
|
|
+ vo.setDataPacketID(generateDataPacketID());
|
|
|
+ vo.setSensorID(deviceId);
|
|
|
+ vo.setEngineeringID(engineeringId);
|
|
|
+ vo.setPublishTime(getCurrentTime());
|
|
|
+ vo.setDataEndTime(dataEndTime);
|
|
|
+ vo.setAVoltage(deviceDataItem.getFloat("aVoltage"));
|
|
|
+ vo.setBVoltage(deviceDataItem.getFloat("bVoltage"));
|
|
|
+ vo.setCVoltage(deviceDataItem.getFloat("cVoltage"));
|
|
|
+ vo.setAElectricity(deviceDataItem.getFloat("aElectricity"));
|
|
|
+ vo.setBElectricity(deviceDataItem.getFloat("bElectricity"));
|
|
|
+ vo.setCElectricity(deviceDataItem.getFloat("cElectricity"));
|
|
|
+ vo.setLine1TEMP(deviceDataItem.getFloat("line1TEMP"));
|
|
|
+ vo.setLine2TEMP(deviceDataItem.getFloat("Line2TEMP"));
|
|
|
+ vo.setLine3TEMP(deviceDataItem.getFloat("Line3TEMP"));
|
|
|
+ vo.setLeakageCurrent(deviceDataItem.getFloat("leakageCurrent"));
|
|
|
+
|
|
|
+ // 根据模拟模式选择不同的功率字段
|
|
|
+ vo.setTotalPower(simulation ?
|
|
|
+ deviceDataItem.getFloat("totalPower") :
|
|
|
+ deviceDataItem.getFloat("active_power"));
|
|
|
+
|
|
|
+ try {
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.ELECTRICITY_LOAD, vo, "人防用电负荷情况");
|
|
|
+ result.put("successCount", result.get("successCount") + 1);
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.warn("设备{}的人防用电负荷情况数据推送失败:{}", deviceId, e.getMessage());
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ log.info("人防用电负荷情况数据推送完成,设备类型:{},成功:{},失败:{}",
|
|
|
+ deviceType, result.get("successCount"), result.get("failureCount"));
|
|
|
+
|
|
|
+ return result;
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("人防用电负荷情况数据推送发生异常", e);
|
|
|
+ result.put("failureCount", transferVO.getDevices().size());
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 推送温度信息(701)
|
|
|
+ *
|
|
|
+ * @param deviceDataItem 设备数据
|
|
|
+ * @param deviceId 设备ID
|
|
|
+ * @param dataEndTime 数据结束时间
|
|
|
+ * @param engineeringID 工程ID
|
|
|
+ **/
|
|
|
+ private void sendTempData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID) {
|
|
|
+ Float value = deviceDataItem.getFloat("wd");
|
|
|
+ if (value == null) {
|
|
|
+ log.warn("设备{}的温度数据为空", deviceId);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ TempVO tempVO = new TempVO();
|
|
|
+ tempVO.setDataPacketID(generateDataPacketID());
|
|
|
+ tempVO.setSensorID(deviceId);
|
|
|
+ tempVO.setEngineeringID(engineeringID);
|
|
|
+ tempVO.setPublishTime(getCurrentTime());
|
|
|
+ tempVO.setSensorValue(value);
|
|
|
+ tempVO.setDataEndTime(dataEndTime);
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.TEMP, tempVO, "温度信息");
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 推送湿度信息(702)
|
|
|
+ *
|
|
|
+ * @param deviceDataItem 设备数据
|
|
|
+ * @param deviceId 设备ID
|
|
|
+ * @param dataEndTime 数据结束时间
|
|
|
+ * @param engineeringID 工程ID
|
|
|
+ **/
|
|
|
+ private void sendHumidityData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID) {
|
|
|
+ Float value = deviceDataItem.getFloat("sd");
|
|
|
+ if (value == null) {
|
|
|
+ log.warn("设备{}的湿度数据为空", deviceId);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ HumidityVO humidityVO = new HumidityVO();
|
|
|
+ humidityVO.setDataPacketID(generateDataPacketID());
|
|
|
+ humidityVO.setSensorID(deviceId);
|
|
|
+ humidityVO.setEngineeringID(engineeringID);
|
|
|
+ humidityVO.setPublishTime(getCurrentTime());
|
|
|
+ humidityVO.setSensorValue(value);
|
|
|
+ humidityVO.setDataEndTime(dataEndTime);
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.HUMIDITY, humidityVO, "湿度信息");
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 推送氧气浓度信息(705)
|
|
|
+ *
|
|
|
+ * @param deviceDataItem 设备数据
|
|
|
+ * @param deviceId 设备ID
|
|
|
+ * @param dataEndTime 数据结束时间
|
|
|
+ * @param engineeringID 工程ID
|
|
|
+ **/
|
|
|
+ private void sendOxygenData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID) {
|
|
|
+ Float value = deviceDataItem.getFloat("o2");
|
|
|
+ if (value == null) {
|
|
|
+ log.warn("设备{}的氧气浓度数据为空", deviceId);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ OxygenVO oxygenVO = new OxygenVO();
|
|
|
+ oxygenVO.setDataPacketID(generateDataPacketID());
|
|
|
+ oxygenVO.setSensorID(deviceId);
|
|
|
+ oxygenVO.setEngineeringID(engineeringID);
|
|
|
+ oxygenVO.setPublishTime(getCurrentTime());
|
|
|
+ oxygenVO.setSensorValue(value);
|
|
|
+ oxygenVO.setDataEndTime(dataEndTime);
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.OXYGEN, oxygenVO, "氧气浓度信息");
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 推送一氧化碳浓度信息(706)
|
|
|
+ *
|
|
|
+ * @param deviceDataItem 设备数据
|
|
|
+ * @param deviceId 设备ID
|
|
|
+ * @param dataEndTime 数据结束时间
|
|
|
+ * @param engineeringID 工程ID
|
|
|
+ **/
|
|
|
+ private void sendCoData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID) {
|
|
|
+ Float value = deviceDataItem.getFloat("co");
|
|
|
+ if (value == null) {
|
|
|
+ log.warn("设备{}的一氧化碳浓度数据为空", deviceId);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ CoVO coVO = new CoVO();
|
|
|
+ coVO.setDataPacketID(generateDataPacketID());
|
|
|
+ coVO.setSensorID(deviceId);
|
|
|
+ coVO.setEngineeringID(engineeringID);
|
|
|
+ coVO.setPublishTime(getCurrentTime());
|
|
|
+ coVO.setSensorValue(value);
|
|
|
+ coVO.setDataEndTime(dataEndTime);
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.CO, coVO, "一氧化碳浓度信息");
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 推送二氧化碳浓度信息(707)
|
|
|
+ *
|
|
|
+ * @param deviceDataItem 设备数据
|
|
|
+ * @param deviceId 设备ID
|
|
|
+ * @param dataEndTime 数据结束时间
|
|
|
+ * @param engineeringID 工程ID
|
|
|
+ **/
|
|
|
+ private void sendCo2Data(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID) {
|
|
|
+ Float value = deviceDataItem.getFloat("co2");
|
|
|
+ if (value == null) {
|
|
|
+ log.warn("设备{}的二氧化碳浓度数据为空", deviceId);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ Co2VO co2VO = new Co2VO();
|
|
|
+ co2VO.setDataPacketID(generateDataPacketID());
|
|
|
+ co2VO.setSensorID(deviceId);
|
|
|
+ co2VO.setEngineeringID(engineeringID);
|
|
|
+ co2VO.setPublishTime(getCurrentTime());
|
|
|
+ co2VO.setSensorValue(value);
|
|
|
+ co2VO.setDataEndTime(dataEndTime);
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.CO2, co2VO, "二氧化碳浓度信息");
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 发送倾斜数据(712)
|
|
|
+ *
|
|
|
+ * @return 推送结果,包含成功数和失败数
|
|
|
+ **/
|
|
|
+ public Map<String, Integer> sendTiltData(IotDataTransferVO transferVO) {
|
|
|
+ Map<String, Integer> result = new HashMap<>();
|
|
|
+ result.put("successCount", 0);
|
|
|
+ result.put("failureCount", 0);
|
|
|
+
|
|
|
+ if (!validateMqttGateway()) {
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ try {
|
|
|
+ List<JSONObject> deviceData = deviceDataQuery.getDeviceData(transferVO);
|
|
|
+ Integer deviceType = transferVO.getDeviceType();
|
|
|
+ Integer totalDevices = transferVO.getDevices().size();
|
|
|
+
|
|
|
+ log.info("开始推送倾斜数据,设备类型:{},设备数量:{},获取到的数据条数:{}",
|
|
|
+ deviceType, totalDevices, deviceData.size());
|
|
|
+
|
|
|
+ if (deviceData.isEmpty()) {
|
|
|
+ log.warn("没有获取到倾斜数据!设备类型:{}", deviceType);
|
|
|
+ result.put("failureCount", totalDevices);
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ Long engineeringId = transferVO.getEngineeringId();
|
|
|
+ for (JSONObject deviceDataItem : deviceData) {
|
|
|
+ LocalDateTime dataEndTime = parseDataTime(deviceDataItem);
|
|
|
+ if (dataEndTime == null) {
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ Integer deviceId = deviceDataItem.getIntValue("device_id");
|
|
|
+ Integer value = deviceDataItem.getInteger("qx");
|
|
|
+ if (value == null) {
|
|
|
+ log.warn("设备{}的倾斜数据为空", deviceId);
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ TiltVO vo = new TiltVO();
|
|
|
+ vo.setDataPacketID(generateDataPacketID());
|
|
|
+ vo.setSensorID(deviceId);
|
|
|
+ vo.setEngineeringID(engineeringId);
|
|
|
+ vo.setPublishTime(getCurrentTime());
|
|
|
+ vo.setDataEndTime(dataEndTime);
|
|
|
+ vo.setSensorValue(value == 0 ? 0 : 1);
|
|
|
+
|
|
|
+ try {
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.TILT, vo, "倾斜信息");
|
|
|
+ result.put("successCount", result.get("successCount") + 1);
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.warn("设备{}的倾斜数据推送失败:{}", deviceId, e.getMessage());
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ log.info("倾斜数据推送完成,设备类型:{},成功:{},失败:{}",
|
|
|
+ deviceType, result.get("successCount"), result.get("failureCount"));
|
|
|
+
|
|
|
+ return result;
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("倾斜数据推送发生异常", e);
|
|
|
+ result.put("failureCount", transferVO.getDevices().size());
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 发送裂缝数据(713)
|
|
|
+ *
|
|
|
+ * @return 推送结果,包含成功数和失败数
|
|
|
+ **/
|
|
|
+ public Map<String, Integer> sendCrackData(IotDataTransferVO transferVO) {
|
|
|
+ Map<String, Integer> result = new HashMap<>();
|
|
|
+ result.put("successCount", 0);
|
|
|
+ result.put("failureCount", 0);
|
|
|
+
|
|
|
+ if (!validateMqttGateway()) {
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ try {
|
|
|
+ List<JSONObject> deviceData = deviceDataQuery.getDeviceData(transferVO);
|
|
|
+ Integer deviceType = transferVO.getDeviceType();
|
|
|
+ Integer totalDevices = transferVO.getDevices().size();
|
|
|
+
|
|
|
+ log.info("开始推送裂缝数据,设备类型:{},设备数量:{},获取到的数据条数:{}",
|
|
|
+ deviceType, totalDevices, deviceData.size());
|
|
|
+
|
|
|
+ if (deviceData.isEmpty()) {
|
|
|
+ log.warn("没有获取到裂缝数据!设备类型:{}", deviceType);
|
|
|
+ result.put("failureCount", totalDevices);
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ Long engineeringId = transferVO.getEngineeringId();
|
|
|
+ for (JSONObject deviceDataItem : deviceData) {
|
|
|
+ LocalDateTime dataEndTime = parseDataTime(deviceDataItem);
|
|
|
+ if (dataEndTime == null) {
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ Integer deviceId = deviceDataItem.getIntValue("device_id");
|
|
|
+ Integer value = deviceDataItem.getInteger("cd");
|
|
|
+ if (value == null) {
|
|
|
+ log.warn("设备{}的裂缝数据为空", deviceId);
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ CrackVO vo = new CrackVO();
|
|
|
+ vo.setDataPacketID(generateDataPacketID());
|
|
|
+ vo.setSensorID(deviceId);
|
|
|
+ vo.setEngineeringID(engineeringId);
|
|
|
+ vo.setPublishTime(getCurrentTime());
|
|
|
+ vo.setDataEndTime(dataEndTime);
|
|
|
+ vo.setSensorValue(value == 0 ? 0 : 1);
|
|
|
+
|
|
|
+ try {
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.CRACK, vo, "裂缝信息");
|
|
|
+ result.put("successCount", result.get("successCount") + 1);
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.warn("设备{}的裂缝数据推送失败:{}", deviceId, e.getMessage());
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ log.info("裂缝数据推送完成,设备类型:{},成功:{},失败:{}",
|
|
|
+ deviceType, result.get("successCount"), result.get("failureCount"));
|
|
|
+
|
|
|
+ return result;
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("裂缝数据推送发生异常", e);
|
|
|
+ result.put("failureCount", transferVO.getDevices().size());
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 发送位移数据(714)
|
|
|
+ *
|
|
|
+ * @return 推送结果,包含成功数和失败数
|
|
|
+ **/
|
|
|
+ public Map<String, Integer> sendDeviationData(IotDataTransferVO transferVO) {
|
|
|
+ Map<String, Integer> result = new HashMap<>();
|
|
|
+ result.put("successCount", 0);
|
|
|
+ result.put("failureCount", 0);
|
|
|
+
|
|
|
+ if (!validateMqttGateway()) {
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ try {
|
|
|
+ List<JSONObject> deviceData = deviceDataQuery.getDeviceData(transferVO);
|
|
|
+ Integer deviceType = transferVO.getDeviceType();
|
|
|
+ Integer totalDevices = transferVO.getDevices().size();
|
|
|
+
|
|
|
+ log.info("开始推送位移数据,设备类型:{},设备数量:{},获取到的数据条数:{}",
|
|
|
+ deviceType, totalDevices, deviceData.size());
|
|
|
+
|
|
|
+ if (deviceData.isEmpty()) {
|
|
|
+ log.warn("没有获取到位移数据!设备类型:{}", deviceType);
|
|
|
+ result.put("failureCount", totalDevices);
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ Long engineeringId = transferVO.getEngineeringId();
|
|
|
+ for (JSONObject deviceDataItem : deviceData) {
|
|
|
+ LocalDateTime dataEndTime = parseDataTime(deviceDataItem);
|
|
|
+ if (dataEndTime == null) {
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ Integer deviceId = deviceDataItem.getIntValue("device_id");
|
|
|
+ Integer value = deviceDataItem.getInteger("wy");
|
|
|
+ if (value == null) {
|
|
|
+ log.warn("设备{}的位移数据为空", deviceId);
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ DeviationVO vo = new DeviationVO();
|
|
|
+ vo.setDataPacketID(generateDataPacketID());
|
|
|
+ vo.setSensorID(deviceId);
|
|
|
+ vo.setEngineeringID(engineeringId);
|
|
|
+ vo.setPublishTime(getCurrentTime());
|
|
|
+ vo.setDataEndTime(dataEndTime);
|
|
|
+ vo.setSensorValue(value == 0 ? 0 : 1);
|
|
|
+
|
|
|
+ try {
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.DEVIATION, vo, "位移信息");
|
|
|
+ result.put("successCount", result.get("successCount") + 1);
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.warn("设备{}的位移数据推送失败:{}", deviceId, e.getMessage());
|
|
|
+ result.put("failureCount", result.get("failureCount") + 1);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ log.info("位移数据推送完成,设备类型:{},成功:{},失败:{}",
|
|
|
+ deviceType, result.get("successCount"), result.get("failureCount"));
|
|
|
+
|
|
|
+ return result;
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("位移数据推送发生异常", e);
|
|
|
+ result.put("failureCount", transferVO.getDevices().size());
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 同步设备数据
|
|
|
+ * @param tenantId 租户ID
|
|
|
+ * @param engineeringId 工程ID
|
|
|
+ * @param username MQTT用户名
|
|
|
+ * @param password MQTT密码
|
|
|
+ */
|
|
|
+ public void synchronizeDeviceData(Integer tenantId, Long engineeringId, String username, String password) {
|
|
|
+ // 参数校验
|
|
|
+ if (engineeringId == null || username == null || password == null) {
|
|
|
+ log.error("工程ID、MQTT用户名或密码不能为空");
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 查询租户下的所有产品类型
|
|
|
+ List<String> deviceTypeList = getDeviceTypeListByTenant(tenantId);
|
|
|
+ if (deviceTypeList.isEmpty()) {
|
|
|
+ log.warn("租户{}不存在任何产品", tenantId);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 查询设备列表
|
|
|
+ List<DmpDevice> deviceList = getDeviceListByType(deviceTypeList);
|
|
|
+ if (deviceList.isEmpty()) {
|
|
|
+ log.warn("租户{}不存在任何设备", tenantId);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 按设备类型分组
|
|
|
+ Map<Integer, List<DmpDevice>> deviceTypeMap = deviceList.stream()
|
|
|
+ .collect(Collectors.groupingBy(DmpDevice::getDeviceType));
|
|
|
+
|
|
|
+ // 构建数据传输对象列表
|
|
|
+ List<IotDataTransferVO> transferList = new ArrayList<>();
|
|
|
+ deviceTypeMap.forEach((deviceType, devices) -> {
|
|
|
+ IotDataTransferVO transferVO = new IotDataTransferVO();
|
|
|
+ transferVO.setDeviceType(deviceType);
|
|
|
+ transferVO.setDevices(devices);
|
|
|
+ transferVO.setEngineeringId(engineeringId);
|
|
|
+ transferList.add(transferVO);
|
|
|
+ });
|
|
|
+
|
|
|
+ // 按产品代码分组,构建ProductCode到uuid列表的映射,过滤掉产品代码为null的设备
|
|
|
+ Map<String, List<String>> codeDeviceUuidsMap = deviceList.stream()
|
|
|
+ .filter(device -> device.getProductCode() != null)
|
|
|
+ .collect(Collectors.groupingBy(DmpDevice::getProductCode,
|
|
|
+ Collectors.mapping(DmpDevice::getDeviceUuid, Collectors.toList())));
|
|
|
+
|
|
|
+ // 创建MQTT连接
|
|
|
+ createMqttConnection(username, password);
|
|
|
+
|
|
|
+ // 任务开始日志
|
|
|
+ Integer totalDevices = deviceList.size();
|
|
|
+ Integer totalProductTypes = codeDeviceUuidsMap.size();
|
|
|
+ log.info("设备数据同步任务开始,租户ID:{},工程ID:{}", tenantId, engineeringId);
|
|
|
+ log.info("总共涉及产品类型数:{}个,产品代码为:{}", totalProductTypes, codeDeviceUuidsMap.keySet());
|
|
|
+ log.info("总共需要推送设备数量:{}个,涉及设备类型数:{}个,设备类型为:{}",
|
|
|
+ totalDevices, deviceTypeMap.size(), deviceTypeMap.keySet());
|
|
|
+
|
|
|
+ // 记录每种设备类型的设备数量
|
|
|
+ deviceTypeMap.forEach((deviceType, devices) -> {
|
|
|
+ log.info("设备类型:{},设备数量:{}", deviceType, devices.size());
|
|
|
+ });
|
|
|
+
|
|
|
+ // 按设备类型处理数据同步
|
|
|
+ int totalSuccessCount = 0;
|
|
|
+ int totalFailureCount = 0;
|
|
|
+
|
|
|
+ for (IotDataTransferVO transferVO : transferList) {
|
|
|
+ Integer deviceType = transferVO.getDeviceType();
|
|
|
+ Map<String, Integer> result = new HashMap<>();
|
|
|
+
|
|
|
+ switch (deviceType) {
|
|
|
+ case 707:
|
|
|
+ case 708:
|
|
|
+ case 709:
|
|
|
+ case 710:
|
|
|
+ case 711:
|
|
|
+ result = sendEnvData(transferVO);
|
|
|
+ break;
|
|
|
+ case 702:
|
|
|
+ result = sendWaterLeak(transferVO);
|
|
|
+ break;
|
|
|
+ case 703:
|
|
|
+ result = sendPersonPresence(transferVO);
|
|
|
+ break;
|
|
|
+ case 704:
|
|
|
+ result = sendElectricityLoad(transferVO);
|
|
|
+ break;
|
|
|
+ case 712:
|
|
|
+ result = sendTiltData(transferVO);
|
|
|
+ break;
|
|
|
+ case 713:
|
|
|
+ result = sendCrackData(transferVO);
|
|
|
+ break;
|
|
|
+ case 714:
|
|
|
+ result = sendDeviationData(transferVO);
|
|
|
+ break;
|
|
|
+
|
|
|
+ default:
|
|
|
+ log.debug("不支持的设备类型:{}", deviceType);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 累加成功数和失败数
|
|
|
+ totalSuccessCount += result.get("successCount");
|
|
|
+ totalFailureCount += result.get("failureCount");
|
|
|
+ }
|
|
|
+
|
|
|
+ // 任务完成总结
|
|
|
+ log.info("设备数据同步任务完成,租户ID:{},工程ID:{}", tenantId, engineeringId);
|
|
|
+ log.info("总共涉及产品类型数:{}个,产品代码为:{}", totalProductTypes, codeDeviceUuidsMap.keySet());
|
|
|
+ log.info("总共推送设备数量:{}个,成功:{}个,失败:{}个",
|
|
|
+ totalDevices, totalSuccessCount, totalFailureCount);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 查询租户下的设备类型列表
|
|
|
+ * @param tenantId 租户ID
|
|
|
+ * @return 设备类型列表
|
|
|
+ */
|
|
|
+ private List<String> getDeviceTypeListByTenant(Integer tenantId) {
|
|
|
+ LambdaQueryWrapper<DmpProduct> productQueryWrapper = new LambdaQueryWrapper<>();
|
|
|
+ productQueryWrapper.select(DmpProduct::getProductCode)
|
|
|
+ .eq(tenantId != null && tenantId > 0, DmpProduct::getTenantId, tenantId)
|
|
|
+ .eq(DmpProduct::getDeleteFlag, 0);
|
|
|
+ List<DmpProduct> productList = dmpProductMapper.selectList(productQueryWrapper);
|
|
|
+ return productList.stream()
|
|
|
+ .map(DmpProduct::getProductCode)
|
|
|
+ .distinct() // 去重
|
|
|
+ .collect(Collectors.toList());
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 根据设备类型列表查询设备
|
|
|
+ * @param productCodeList 设备类型列表
|
|
|
+ * @return 设备列表
|
|
|
+ */
|
|
|
+ private List<DmpDevice> getDeviceListByType(List<String> productCodeList) {
|
|
|
+ LambdaQueryWrapper<DmpDevice> queryWrapper = new LambdaQueryWrapper<>();
|
|
|
+ queryWrapper.select(DmpDevice::getDeviceUuid, DmpDevice::getDeviceType, DmpDevice::getDeviceId, DmpDevice::getProductCode)
|
|
|
+ .in(DmpDevice::getProductCode, productCodeList)
|
|
|
+ .eq(DmpDevice::getDeleteFlag, 0)
|
|
|
+ .notIn(DmpDevice::getServiceStatus, 3)
|
|
|
+ .orderByAsc(DmpDevice::getProductCode);
|
|
|
+ return dmpDeviceMapper.selectList(queryWrapper);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 手动创建MQTT连接
|
|
|
+ * @param username MQTT用户名
|
|
|
+ * @param password MQTT密码
|
|
|
+ */
|
|
|
+ public void createMqttConnection(String username, String password) {
|
|
|
+ try {
|
|
|
+ // 使用注入的ApplicationContext获取已有的mqttGateway实例
|
|
|
+ // 因为我们保留了@MessagingGateway注解,Spring会自动创建这个实例
|
|
|
+ if (this.context == null) {
|
|
|
+ throw new IllegalStateException("ApplicationContext未注入,无法获取MQTT Gateway");
|
|
|
+ }
|
|
|
+
|
|
|
+ // 1. 获取mqttGateway实例
|
|
|
+ this.mqttGateway = this.context.getBean(MqttOutConfig.MqttGateway.class);
|
|
|
+ if (this.mqttGateway == null) {
|
|
|
+ throw new IllegalStateException("MQTT Gateway未找到,无法发送消息");
|
|
|
+ }
|
|
|
+
|
|
|
+ // 2. 获取现有的mqttClientFactory实例
|
|
|
+ DefaultMqttPahoClientFactory mqttClientFactory = this.context.getBean(DefaultMqttPahoClientFactory.class);
|
|
|
+ if (mqttClientFactory == null) {
|
|
|
+ throw new IllegalStateException("MQTT Client Factory未找到,无法创建MQTT连接");
|
|
|
+ }
|
|
|
+
|
|
|
+ // 3. 创建并配置MqttConnectOptions
|
|
|
+ MqttConnectOptions options = new MqttConnectOptions();
|
|
|
+ options.setServerURIs(new String[]{MQTT_URL});
|
|
|
+ options.setUserName(username);
|
|
|
+ options.setPassword(password.toCharArray());
|
|
|
+ options.setKeepAliveInterval(KEEP_ALIVE_INTERVAL);
|
|
|
+
|
|
|
+ // 4. 更新mqttClientFactory的连接选项
|
|
|
+ mqttClientFactory.setConnectionOptions(options);
|
|
|
+
|
|
|
+ log.info("MQTT Gateway初始化成功,用户名:{}", username);
|
|
|
+ log.info("MQTT连接配置完成,服务器地址:{},客户端ID:mqttx-{}", MQTT_URL, username);
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("初始化MQTT连接失败: {}", e.getMessage(), e);
|
|
|
+ throw new RuntimeException("初始化MQTT连接失败", e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 验证MQTT网关是否初始化
|
|
|
+ * @return 是否初始化
|
|
|
+ */
|
|
|
+ private boolean validateMqttGateway() {
|
|
|
+ if (mqttGateway == null) {
|
|
|
+ log.warn("MQTT Gateway未初始化,无法发送消息");
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 解析数据时间
|
|
|
+ * @param deviceDataItem 设备数据
|
|
|
+ * @return 解析后的时间,如果解析失败返回null
|
|
|
+ */
|
|
|
+ private LocalDateTime parseDataTime(JSONObject deviceDataItem) {
|
|
|
+ log.warn("解析的json{}", deviceDataItem.toString());
|
|
|
+ Long dataTime = deviceDataItem.getLong("realtime");
|
|
|
+ if (dataTime == null) {
|
|
|
+ log.warn("设备{}的time为空", deviceDataItem.getString("device_id"));
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ return LocalDateTime.ofInstant(Instant.ofEpochMilli(dataTime), ZoneId.systemDefault());
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 发送MQTT消息
|
|
|
+ * @param topicEnum 主题枚举
|
|
|
+ * @param vo 消息对象
|
|
|
+ * @param messageType 消息类型描述
|
|
|
+ */
|
|
|
+ private void sendMqttMessage(EnvMonitorMqttTopic topicEnum, Object vo, String messageType) {
|
|
|
+ String json = JSON.toJSONString(vo);
|
|
|
+ String topic = topicEnum.getTopic();
|
|
|
+ // 不再记录每条数据的详情,只记录发送操作
|
|
|
+ mqttGateway.sendToMqtt(topic, json);
|
|
|
+ }
|
|
|
+
|
|
|
+ public void allData(Long engineeringId, String username, String password) {
|
|
|
+ Integer tenantId = 0;
|
|
|
+ synchronizeDeviceData(tenantId, engineeringId, username, password);
|
|
|
+ }
|
|
|
+}
|