|
|
@@ -7,29 +7,35 @@ 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.MqttGateway;
|
|
|
import com.usky.cdi.service.config.mqtt.MqttOutConfig;
|
|
|
import com.usky.cdi.service.enums.EnvMonitorMqttTopic;
|
|
|
+import com.usky.cdi.service.mqtt.MqttConnectionTool;
|
|
|
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.boot.autoconfigure.condition.ConditionalOnProperty;
|
|
|
+import org.springframework.beans.factory.support.DefaultListableBeanFactory;
|
|
|
+import org.springframework.context.ApplicationContext;
|
|
|
+import org.springframework.context.ConfigurableApplicationContext;
|
|
|
+import org.springframework.context.support.GenericApplicationContext;
|
|
|
+import org.springframework.integration.dsl.IntegrationFlow;
|
|
|
+import org.springframework.integration.dsl.IntegrationFlows;
|
|
|
+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 javax.annotation.PostConstruct;
|
|
|
|
|
|
-import javax.annotation.Resource;
|
|
|
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.*;
|
|
|
+import java.util.concurrent.ConcurrentHashMap;
|
|
|
import java.util.stream.Collectors;
|
|
|
|
|
|
/**
|
|
|
@@ -39,12 +45,27 @@ import java.util.stream.Collectors;
|
|
|
*/
|
|
|
@Slf4j
|
|
|
@Service
|
|
|
-@ConditionalOnProperty(prefix = "mqtt", value = {"enabled"}, havingValue = "true")
|
|
|
public class IotDataTransferService {
|
|
|
|
|
|
- @Resource
|
|
|
private MqttOutConfig.MqttGateway mqttGateway;
|
|
|
|
|
|
+ @Autowired
|
|
|
+ private MqttConnectionTool mqttConnectionTool;
|
|
|
+
|
|
|
+ // 注入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;
|
|
|
+
|
|
|
+ // 存储每个任务的MQTT客户端工厂和网关
|
|
|
+ private final Map<String, MqttConnectionTool.MqttGateway> mqttGatewayMap = new ConcurrentHashMap<>();
|
|
|
+ private final Map<String, DefaultMqttPahoClientFactory> mqttClientFactoryMap = new ConcurrentHashMap<>();
|
|
|
+
|
|
|
private SnowflakeIdGenerator idGenerator;
|
|
|
|
|
|
@Autowired
|
|
|
@@ -96,12 +117,13 @@ public class IotDataTransferService {
|
|
|
result.put("successCount", 0);
|
|
|
result.put("failureCount", 0);
|
|
|
|
|
|
- if (!validateMqttGateway()) {
|
|
|
+ if (!validateMqttGateway(transferVO.getUsername())) {
|
|
|
return result;
|
|
|
}
|
|
|
|
|
|
try {
|
|
|
List<JSONObject> deviceData = deviceDataQuery.getDeviceData(transferVO);
|
|
|
+ log.warn("获取到的数据:{}", deviceData);
|
|
|
Integer deviceType = transferVO.getDeviceType();
|
|
|
Integer totalDevices = transferVO.getDevices().size();
|
|
|
|
|
|
@@ -139,7 +161,7 @@ public class IotDataTransferService {
|
|
|
vo.setDataEndTime(dataEndTime);
|
|
|
|
|
|
try {
|
|
|
- sendMqttMessage(EnvMonitorMqttTopic.WATER_LEAK, vo, "水浸状态信息");
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.WATER_LEAK, vo, "水浸状态信息", transferVO.getUsername());
|
|
|
result.put("successCount", result.get("successCount") + 1);
|
|
|
} catch (Exception e) {
|
|
|
log.warn("设备{}的水浸状态数据推送失败:{}", deviceId, e.getMessage());
|
|
|
@@ -169,7 +191,7 @@ public class IotDataTransferService {
|
|
|
result.put("successCount", 0);
|
|
|
result.put("failureCount", 0);
|
|
|
|
|
|
- if (!validateMqttGateway()) {
|
|
|
+ if (!validateMqttGateway(transferVO.getUsername())) {
|
|
|
return result;
|
|
|
}
|
|
|
|
|
|
@@ -178,8 +200,20 @@ public class IotDataTransferService {
|
|
|
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);
|
|
|
@@ -202,19 +236,19 @@ public class IotDataTransferService {
|
|
|
try {
|
|
|
switch (deviceType) {
|
|
|
case 707:
|
|
|
- sendTempData(deviceId, dataEndTime, deviceDataItem, engineeringId);
|
|
|
+ sendTempData(deviceId, dataEndTime, deviceDataItem, engineeringId, transferVO.getUsername());
|
|
|
break;
|
|
|
case 708:
|
|
|
- sendHumidityData(deviceId, dataEndTime, deviceDataItem, engineeringId);
|
|
|
+ sendHumidityData(deviceId, dataEndTime, deviceDataItem, engineeringId, transferVO.getUsername());
|
|
|
break;
|
|
|
case 709:
|
|
|
- sendOxygenData(deviceId, dataEndTime, deviceDataItem, engineeringId);
|
|
|
+ sendOxygenData(deviceId, dataEndTime, deviceDataItem, engineeringId, transferVO.getUsername());
|
|
|
break;
|
|
|
case 710:
|
|
|
- sendCo2Data(deviceId, dataEndTime, deviceDataItem, engineeringId);
|
|
|
+ sendCo2Data(deviceId, dataEndTime, deviceDataItem, engineeringId, transferVO.getUsername());
|
|
|
break;
|
|
|
case 711:
|
|
|
- sendCoData(deviceId, dataEndTime, deviceDataItem, engineeringId);
|
|
|
+ sendCoData(deviceId, dataEndTime, deviceDataItem, engineeringId, transferVO.getUsername());
|
|
|
break;
|
|
|
}
|
|
|
} catch (Exception e) {
|
|
|
@@ -251,7 +285,7 @@ public class IotDataTransferService {
|
|
|
result.put("successCount", 0);
|
|
|
result.put("failureCount", 0);
|
|
|
|
|
|
- if (!validateMqttGateway()) {
|
|
|
+ if (!validateMqttGateway(transferVO.getUsername())) {
|
|
|
return result;
|
|
|
}
|
|
|
|
|
|
@@ -289,7 +323,7 @@ public class IotDataTransferService {
|
|
|
vo.setSensorValue(0);
|
|
|
|
|
|
try {
|
|
|
- sendMqttMessage(EnvMonitorMqttTopic.PERSON_PRESENCE, vo, "人员闯入情况");
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.PERSON_PRESENCE, vo, "人员闯入情况", transferVO.getUsername());
|
|
|
result.put("successCount", result.get("successCount") + 1);
|
|
|
} catch (Exception e) {
|
|
|
log.warn("设备{}的人员闯入情况数据推送失败:{}", deviceId, e.getMessage());
|
|
|
@@ -318,7 +352,7 @@ public class IotDataTransferService {
|
|
|
result.put("successCount", 0);
|
|
|
result.put("failureCount", 0);
|
|
|
|
|
|
- if (!validateMqttGateway()) {
|
|
|
+ if (!validateMqttGateway(transferVO.getUsername())) {
|
|
|
return result;
|
|
|
}
|
|
|
|
|
|
@@ -368,7 +402,7 @@ public class IotDataTransferService {
|
|
|
deviceDataItem.getFloat("active_power"));
|
|
|
|
|
|
try {
|
|
|
- sendMqttMessage(EnvMonitorMqttTopic.ELECTRICITY_LOAD, vo, "人防用电负荷情况");
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.ELECTRICITY_LOAD, vo, "人防用电负荷情况", transferVO.getUsername());
|
|
|
result.put("successCount", result.get("successCount") + 1);
|
|
|
} catch (Exception e) {
|
|
|
log.warn("设备{}的人防用电负荷情况数据推送失败:{}", deviceId, e.getMessage());
|
|
|
@@ -387,8 +421,16 @@ public class IotDataTransferService {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- // 提取公共发送方法,减少代码冗余
|
|
|
- private void sendTempData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID) {
|
|
|
+ /**
|
|
|
+ * 推送温度信息(701)
|
|
|
+ *
|
|
|
+ * @param deviceDataItem 设备数据
|
|
|
+ * @param deviceId 设备ID
|
|
|
+ * @param dataEndTime 数据结束时间
|
|
|
+ * @param engineeringID 工程ID
|
|
|
+ * @param username 用户名
|
|
|
+ **/
|
|
|
+ private void sendTempData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID, String username) {
|
|
|
Float value = deviceDataItem.getFloat("wd");
|
|
|
if (value == null) {
|
|
|
log.warn("设备{}的温度数据为空", deviceId);
|
|
|
@@ -401,10 +443,19 @@ public class IotDataTransferService {
|
|
|
tempVO.setPublishTime(getCurrentTime());
|
|
|
tempVO.setSensorValue(value);
|
|
|
tempVO.setDataEndTime(dataEndTime);
|
|
|
- sendMqttMessage(EnvMonitorMqttTopic.TEMP, tempVO, "温度信息");
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.TEMP, tempVO, "温度信息", username);
|
|
|
}
|
|
|
|
|
|
- private void sendHumidityData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID) {
|
|
|
+ /**
|
|
|
+ * 推送湿度信息(702)
|
|
|
+ *
|
|
|
+ * @param deviceDataItem 设备数据
|
|
|
+ * @param deviceId 设备ID
|
|
|
+ * @param dataEndTime 数据结束时间
|
|
|
+ * @param engineeringID 工程ID
|
|
|
+ * @param username 用户名
|
|
|
+ **/
|
|
|
+ private void sendHumidityData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID, String username) {
|
|
|
Float value = deviceDataItem.getFloat("sd");
|
|
|
if (value == null) {
|
|
|
log.warn("设备{}的湿度数据为空", deviceId);
|
|
|
@@ -417,10 +468,19 @@ public class IotDataTransferService {
|
|
|
humidityVO.setPublishTime(getCurrentTime());
|
|
|
humidityVO.setSensorValue(value);
|
|
|
humidityVO.setDataEndTime(dataEndTime);
|
|
|
- sendMqttMessage(EnvMonitorMqttTopic.HUMIDITY, humidityVO, "湿度信息");
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.HUMIDITY, humidityVO, "湿度信息", username);
|
|
|
}
|
|
|
|
|
|
- private void sendOxygenData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID) {
|
|
|
+ /**
|
|
|
+ * 推送氧气浓度信息(705)
|
|
|
+ *
|
|
|
+ * @param deviceDataItem 设备数据
|
|
|
+ * @param deviceId 设备ID
|
|
|
+ * @param dataEndTime 数据结束时间
|
|
|
+ * @param engineeringID 工程ID
|
|
|
+ * @param username 用户名
|
|
|
+ **/
|
|
|
+ private void sendOxygenData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID, String username) {
|
|
|
Float value = deviceDataItem.getFloat("o2");
|
|
|
if (value == null) {
|
|
|
log.warn("设备{}的氧气浓度数据为空", deviceId);
|
|
|
@@ -433,10 +493,19 @@ public class IotDataTransferService {
|
|
|
oxygenVO.setPublishTime(getCurrentTime());
|
|
|
oxygenVO.setSensorValue(value);
|
|
|
oxygenVO.setDataEndTime(dataEndTime);
|
|
|
- sendMqttMessage(EnvMonitorMqttTopic.OXYGEN, oxygenVO, "氧气浓度信息");
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.OXYGEN, oxygenVO, "氧气浓度信息", username);
|
|
|
}
|
|
|
|
|
|
- private void sendCoData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID) {
|
|
|
+ /**
|
|
|
+ * 推送一氧化碳浓度信息(706)
|
|
|
+ *
|
|
|
+ * @param deviceDataItem 设备数据
|
|
|
+ * @param deviceId 设备ID
|
|
|
+ * @param dataEndTime 数据结束时间
|
|
|
+ * @param engineeringID 工程ID
|
|
|
+ * @param username 用户名
|
|
|
+ **/
|
|
|
+ private void sendCoData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID, String username) {
|
|
|
Float value = deviceDataItem.getFloat("co");
|
|
|
if (value == null) {
|
|
|
log.warn("设备{}的一氧化碳浓度数据为空", deviceId);
|
|
|
@@ -449,10 +518,19 @@ public class IotDataTransferService {
|
|
|
coVO.setPublishTime(getCurrentTime());
|
|
|
coVO.setSensorValue(value);
|
|
|
coVO.setDataEndTime(dataEndTime);
|
|
|
- sendMqttMessage(EnvMonitorMqttTopic.CO, coVO, "一氧化碳浓度信息");
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.CO, coVO, "一氧化碳浓度信息", username);
|
|
|
}
|
|
|
|
|
|
- private void sendCo2Data(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID) {
|
|
|
+ /**
|
|
|
+ * 推送二氧化碳浓度信息(707)
|
|
|
+ *
|
|
|
+ * @param deviceDataItem 设备数据
|
|
|
+ * @param deviceId 设备ID
|
|
|
+ * @param dataEndTime 数据结束时间
|
|
|
+ * @param engineeringID 工程ID
|
|
|
+ * @param username 用户名
|
|
|
+ **/
|
|
|
+ private void sendCo2Data(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID, String username) {
|
|
|
Float value = deviceDataItem.getFloat("co2");
|
|
|
if (value == null) {
|
|
|
log.warn("设备{}的二氧化碳浓度数据为空", deviceId);
|
|
|
@@ -465,18 +543,237 @@ public class IotDataTransferService {
|
|
|
co2VO.setPublishTime(getCurrentTime());
|
|
|
co2VO.setSensorValue(value);
|
|
|
co2VO.setDataEndTime(dataEndTime);
|
|
|
- sendMqttMessage(EnvMonitorMqttTopic.CO2, co2VO, "二氧化碳浓度信息");
|
|
|
+ sendMqttMessage(EnvMonitorMqttTopic.CO2, co2VO, "二氧化碳浓度信息", username);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 发送倾斜数据(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(transferVO.getUsername())) {
|
|
|
+ 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, "位移信息", transferVO.getUsername());
|
|
|
+ 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) {
|
|
|
+ public void synchronizeDeviceData(Integer tenantId, Long engineeringId, String username, String password) {
|
|
|
+ log.info("用户名:{},密码:{}", username, password);
|
|
|
// 参数校验
|
|
|
- if (tenantId == null || engineeringId == null) {
|
|
|
- log.error("租户ID或工程ID不能为空");
|
|
|
+ if (engineeringId == null || username == null || password == null) {
|
|
|
+ log.error("工程ID、MQTT用户名或密码不能为空");
|
|
|
return;
|
|
|
}
|
|
|
|
|
|
@@ -488,7 +785,7 @@ public class IotDataTransferService {
|
|
|
}
|
|
|
|
|
|
// 查询设备列表
|
|
|
- List<DmpDevice> deviceList = getDeviceListByType(deviceTypeList, tenantId);
|
|
|
+ List<DmpDevice> deviceList = getDeviceListByType(deviceTypeList);
|
|
|
if (deviceList.isEmpty()) {
|
|
|
log.warn("租户{}不存在任何设备", tenantId);
|
|
|
return;
|
|
|
@@ -505,6 +802,7 @@ public class IotDataTransferService {
|
|
|
transferVO.setDeviceType(deviceType);
|
|
|
transferVO.setDevices(devices);
|
|
|
transferVO.setEngineeringId(engineeringId);
|
|
|
+ transferVO.setUsername(username); // 保存当前任务的用户名
|
|
|
transferList.add(transferVO);
|
|
|
});
|
|
|
|
|
|
@@ -514,6 +812,9 @@ public class IotDataTransferService {
|
|
|
.collect(Collectors.groupingBy(DmpDevice::getProductCode,
|
|
|
Collectors.mapping(DmpDevice::getDeviceUuid, Collectors.toList())));
|
|
|
|
|
|
+ // 创建MQTT连接
|
|
|
+ createMqttConnection(username, password);
|
|
|
+
|
|
|
// 任务开始日志
|
|
|
Integer totalDevices = deviceList.size();
|
|
|
Integer totalProductTypes = codeDeviceUuidsMap.size();
|
|
|
@@ -552,6 +853,16 @@ public class IotDataTransferService {
|
|
|
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;
|
|
|
@@ -577,7 +888,7 @@ public class IotDataTransferService {
|
|
|
private List<String> getDeviceTypeListByTenant(Integer tenantId) {
|
|
|
LambdaQueryWrapper<DmpProduct> productQueryWrapper = new LambdaQueryWrapper<>();
|
|
|
productQueryWrapper.select(DmpProduct::getProductCode)
|
|
|
- .eq(DmpProduct::getTenantId, tenantId)
|
|
|
+ .eq(tenantId != null && tenantId > 0, DmpProduct::getTenantId, tenantId)
|
|
|
.eq(DmpProduct::getDeleteFlag, 0);
|
|
|
List<DmpProduct> productList = dmpProductMapper.selectList(productQueryWrapper);
|
|
|
return productList.stream()
|
|
|
@@ -591,7 +902,7 @@ public class IotDataTransferService {
|
|
|
* @param productCodeList 设备类型列表
|
|
|
* @return 设备列表
|
|
|
*/
|
|
|
- private List<DmpDevice> getDeviceListByType(List<String> productCodeList, Integer tenantId) {
|
|
|
+ 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)
|
|
|
@@ -601,13 +912,45 @@ public class IotDataTransferService {
|
|
|
return dmpDeviceMapper.selectList(queryWrapper);
|
|
|
}
|
|
|
|
|
|
+ /**
|
|
|
+ * 手动创建/刷新 MQTT 连接(含动态 clientId)
|
|
|
+ * @param username MQTT用户名
|
|
|
+ * @param password MQTT密码
|
|
|
+ */
|
|
|
+ public synchronized void createMqttConnection(String username, String password) {
|
|
|
+ log.info("手动创建/刷新 MQTT 连接(含动态 clientId),用户名:{},密码:{}", username, password);
|
|
|
+ try {
|
|
|
+ // 检查MqttConnectionTool是否已注入
|
|
|
+ if (this.mqttConnectionTool == null) {
|
|
|
+ throw new IllegalStateException("MqttConnectionTool未注入,无法获取MQTT Gateway");
|
|
|
+ }
|
|
|
+
|
|
|
+ // 使用MqttConnectionTool创建或刷新MQTT连接
|
|
|
+ MqttConnectionTool.MqttGateway gateway = mqttConnectionTool.connectOrRefresh(username, password);
|
|
|
+
|
|
|
+ // 存储到映射中
|
|
|
+ mqttGatewayMap.put(username, gateway);
|
|
|
+ log.info("MQTT连接创建/刷新成功,用户名:{}", username);
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("初始化MQTT连接失败: {}", e.getMessage(), e);
|
|
|
+ throw new RuntimeException("初始化MQTT连接失败", e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
/**
|
|
|
* 验证MQTT网关是否初始化
|
|
|
+ * @param username 用户名
|
|
|
* @return 是否初始化
|
|
|
*/
|
|
|
- private boolean validateMqttGateway() {
|
|
|
- if (mqttGateway == null) {
|
|
|
- log.warn("MQTT Gateway未初始化,无法发送消息");
|
|
|
+ private boolean validateMqttGateway(String username) {
|
|
|
+ if (username == null) {
|
|
|
+ log.warn("MQTT Gateway未初始化,无法发送消息,用户名:null");
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ // 一次性获取网关实例,避免竞态条件
|
|
|
+ MqttConnectionTool.MqttGateway gateway = mqttGatewayMap.get(username);
|
|
|
+ if (gateway == null) {
|
|
|
+ log.warn("MQTT Gateway未初始化,无法发送消息,用户名:{}", username);
|
|
|
return false;
|
|
|
}
|
|
|
return true;
|
|
|
@@ -619,7 +962,8 @@ public class IotDataTransferService {
|
|
|
* @return 解析后的时间,如果解析失败返回null
|
|
|
*/
|
|
|
private LocalDateTime parseDataTime(JSONObject deviceDataItem) {
|
|
|
- Long dataTime = deviceDataItem.getLong("time");
|
|
|
+ log.warn("解析的json{}", deviceDataItem.toString());
|
|
|
+ Long dataTime = deviceDataItem.getLong("realtime");
|
|
|
if (dataTime == null) {
|
|
|
log.warn("设备{}的time为空", deviceDataItem.getString("device_id"));
|
|
|
return null;
|
|
|
@@ -632,11 +976,22 @@ public class IotDataTransferService {
|
|
|
* @param topicEnum 主题枚举
|
|
|
* @param vo 消息对象
|
|
|
* @param messageType 消息类型描述
|
|
|
+ * @param username 用户名
|
|
|
*/
|
|
|
- private void sendMqttMessage(EnvMonitorMqttTopic topicEnum, Object vo, String messageType) {
|
|
|
+ private void sendMqttMessage(EnvMonitorMqttTopic topicEnum, Object vo, String messageType, String username) {
|
|
|
String json = JSON.toJSONString(vo);
|
|
|
String topic = topicEnum.getTopic();
|
|
|
// 不再记录每条数据的详情,只记录发送操作
|
|
|
- mqttGateway.sendToMqtt(topic, json);
|
|
|
+ MqttConnectionTool.MqttGateway gateway = mqttGatewayMap.get(username);
|
|
|
+ if (gateway != null) {
|
|
|
+ gateway.sendToMqtt(topic, json);
|
|
|
+ } else {
|
|
|
+ log.warn("MQTT Gateway未找到,无法发送消息,用户名:{}", username);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ public void allData(Long engineeringId, String username, String password) {
|
|
|
+ Integer tenantId = 0;
|
|
|
+ synchronizeDeviceData(tenantId, engineeringId, username, password);
|
|
|
}
|
|
|
}
|