package com.zksy.environment.config; import com.zksy.api.domain.WarningThreshold; import com.zksy.api.service.WarningThresholdService; import com.zksy.api.utils.SmsUtil; import com.zksy.environment.domain.ERealTimeData; import com.zksy.environment.mapper.ERealTimeDataMapper; import com.zksy.environment.utils.AlarmUtil; import com.zksy.utils.DevicePhoneFetchUtil; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import rk.netDevice.sdk.p2.*; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import java.math.BigDecimal; import java.text.SimpleDateFormat; import java.time.LocalDateTime; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @Component @Slf4j public class RSServerService { private RSServer rsServer; private Future serverFuture; // 专用线程池:隔离RSServer任务 private final ExecutorService serverExecutor = Executors.newSingleThreadExecutor(r -> { Thread thread = new Thread(r, "RSServer-Worker"); thread.setDaemon(true); return thread; }); @Autowired private ERealTimeDataMapper realTimeDataMapper; @Autowired private WarningThresholdService warningThresholdService; @Autowired private SmsUtil smsUtil; @Autowired private DevicePhoneFetchUtil devicePhoneFetchUtil; @Autowired private AlarmUtil alarmUtil; @PostConstruct public void init() { log.info("RSServer服务开始初始化,监听端口:9801"); // 提交初始化任务 + 启动监控 serverFuture = serverExecutor.submit(this::initializeServer); new Thread(this::monitorServerInitialization, "RSServer-Monitor").start(); } /** * 监控线程 */ private void monitorServerInitialization() { if (serverFuture == null) { log.error("RSServer初始化任务未提交,监控终止"); return; } try { long timeoutSeconds = 20; serverFuture.get(timeoutSeconds, TimeUnit.SECONDS); log.info("RSServer初始化任务正常完成"); } catch (java.util.concurrent.TimeoutException e) { log.warn("RSServer初始化超时(>20秒),服务器可能仍在启动(正常慢启动),可延长timeoutSeconds阈值"); } catch (Exception e) { log.error("RSServer初始化发生异常(非超时)", e); if (!serverFuture.isDone()) { serverFuture.cancel(true); log.warn("已强制取消异常初始化任务"); } } } private void initializeServer() { try { long initiateStartTime = System.currentTimeMillis(); rsServer = RSServer.Initiate(9801); log.info("RSServer实例初始化完成,耗时:{}ms", System.currentTimeMillis() - initiateStartTime); rsServer.addDataListener(new IDataListener() { @Override public void receiveTimmingAck(TimmingAck data) { System.out.println("校时应答->设备编号:" + data.getDeviceId() + "\t执行结果:" + data.getStatus()); } @Override public void receiveTelecontrolAck(TelecontrolAck data) { System.out.println("遥控应答->设备编号:" + data.getDeviceId() + "\t继电器编号:" + data.getRelayId() + "\t执行结果:" + data.getStatus()); } @Override public void receiveStoreData(StoreData data) { for (NodeData nd : data.getNodeList()) { SimpleDateFormat sdf = new SimpleDateFormat( "yy-MM-dd HH:mm:ss"); String str = sdf.format(nd.getRecordTime()); System.out.println("存储数据->设备地址:" + data.getDeviceId() + "\t节点:" + nd.getNodeId() + "\t温度:" + nd.getTem() + "\t湿度:" + nd.getHum() + "\t存储时间:" + str); } } @Override public void receiveRealtimeData(RealTimeData data) { for (NodeData nd : data.getNodeList()) { try { // 获取当前节点ID Integer nodeId = nd.getNodeId(); // 定义最终要存储的数值 float finalValue; // 字段赋值标记:true=存floatValue(悬浮物),false=存hum(湿度) boolean saveAsFloatValue = false; // ====================== 按节点处理数值 ====================== if (nodeId == 1) { // 节点1:悬浮物,系数10 finalValue = nd.getFloatValue() * 10; saveAsFloatValue = true; } else if (nodeId == 2) { // 节点2:湿度,系数1 finalValue = nd.getHum() * 1; } else if (nodeId == 3) { // 节点3:湿度,系数0.1 finalValue = nd.getHum() * 0.1f; } else if (nodeId == 4) { // 节点4:湿度,系数10 finalValue = nd.getHum() * 10; } else if (nodeId == 5) { // 节点5:湿度,系数0.1 finalValue = nd.getHum() * 0.1f; } else { // 未知节点,跳过处理 log.warn("未知节点ID:{},不进行数据处理", nodeId); continue; } // 保留两位小数(核心处理) finalValue = Math.round(finalValue * 100) / 100.0f; // ====================== 封装实体 ====================== ERealTimeData realTimeData = new ERealTimeData(); realTimeData.setDeviceId(data.getDeviceId()); realTimeData.setNodeId(nodeId); realTimeData.setTem(nd.getTem()); realTimeData.setLng(nd.getLng()); realTimeData.setLat(nd.getLat()); realTimeData.setCoordinateType(String.valueOf(data.getCoordinateType())); realTimeData.setRelayStatus(String.valueOf(data.getRelayStatus())); realTimeData.setCreateTime(LocalDateTime.now()); // 按节点类型赋值对应字段 if (saveAsFloatValue) { // 节点1:存入悬浮物 realTimeData.setFloatValue(String.valueOf(finalValue)); // 节点1不需要湿度,清空避免脏数据 realTimeData.setHum(0.0f); } else { // 节点2-5:存入湿度 realTimeData.setHum(finalValue); // 非节点1清空浮点值 realTimeData.setFloatValue("0.00"); } // 数据库插入 realTimeDataMapper.insert(realTimeData); // ====================== 报警处理 ====================== String deviceId = String.valueOf(data.getDeviceId()); // 悬浮物判断 if (!"".equals(realTimeData.getFloatValue()) && nd.getNodeId() == 1) { String temWarningCode = "WARN-SUSPENDED-SOLIDS"; WarningThreshold temThreshold = checkThreshold(deviceId, temWarningCode); Double temMinValue = temThreshold != null ? temThreshold.getMinValue() : null; Double temMaxValue = temThreshold != null ? temThreshold.getMaxValue() : null; String temWarningType = temThreshold != null ? temThreshold.getWarningType() : "悬浮物预警"; String temRemark = temThreshold != null ? temThreshold.getRemark() : "环境悬浮物报警"; boolean temShouldAlarm = false; BigDecimal temValue = BigDecimal.valueOf(nd.getTem()); if (temMinValue != null && nd.getTem() <= temMinValue) { temShouldAlarm = true; } if (temMaxValue != null && nd.getTem() >= temMaxValue) { temShouldAlarm = true; } if (temShouldAlarm) { alarmUtil.saveAlarm(deviceId, temWarningType, temWarningCode, temMinValue != null ? BigDecimal.valueOf(temMinValue) : null, temMaxValue != null ? BigDecimal.valueOf(temMaxValue) : null, temValue, temRemark); } } // 只有非节点1才判断湿度报警(节点1无湿度数据) if (nodeId == 2 && !Float.isNaN(nd.getHum())) { String humWarningCode = "WARN-HUMIDITY"; WarningThreshold humThreshold = checkThreshold(deviceId, humWarningCode); Double humMinValue = humThreshold != null ? humThreshold.getMinValue() : null; Double humMaxValue = humThreshold != null ? humThreshold.getMaxValue() : 90.0; String humWarningType = humThreshold != null ? humThreshold.getWarningType() : "湿度预警"; String humRemark = humThreshold != null ? humThreshold.getRemark() : "环境湿度报警"; boolean humShouldAlarm = false; BigDecimal humValue = BigDecimal.valueOf(finalValue); if (humMinValue != null && finalValue <= humMinValue) { humShouldAlarm = true; } if (humMaxValue != null && finalValue >= humMaxValue) { humShouldAlarm = true; } if (humShouldAlarm) { alarmUtil.saveAlarm(deviceId, humWarningType, humWarningCode, humMinValue != null ? BigDecimal.valueOf(humMinValue) : null, humMaxValue != null ? BigDecimal.valueOf(humMaxValue) : null, humValue, humRemark); } } if (nodeId == 3 && !Float.isNaN(nd.getHum())) { String humWarningCode = "WARN-AMMONIA-NITROGEN"; WarningThreshold humThreshold = checkThreshold(deviceId, humWarningCode); Double humMinValue = humThreshold != null ? humThreshold.getMinValue() : null; Double humMaxValue = humThreshold != null ? humThreshold.getMaxValue() : null; String humWarningType = humThreshold != null ? humThreshold.getWarningType() : "氨氮预警"; String humRemark = humThreshold != null ? humThreshold.getRemark() : "环境氨氮报警"; boolean humShouldAlarm = false; BigDecimal humValue = BigDecimal.valueOf(finalValue); if (humMinValue != null && finalValue <= humMinValue) { humShouldAlarm = true; } if (humMaxValue != null && finalValue >= humMaxValue) { humShouldAlarm = true; } if (humShouldAlarm) { alarmUtil.saveAlarm(deviceId, humWarningType, humWarningCode, humMinValue != null ? BigDecimal.valueOf(humMinValue) : null, humMaxValue != null ? BigDecimal.valueOf(humMaxValue) : null, humValue, humRemark); } } if (nodeId == 4 && !Float.isNaN(nd.getHum())) { String humWarningCode = "WARN-CONDUCTIVITY"; WarningThreshold humThreshold = checkThreshold(deviceId, humWarningCode); Double humMinValue = humThreshold != null ? humThreshold.getMinValue() : null; Double humMaxValue = humThreshold != null ? humThreshold.getMaxValue() : null; String humWarningType = humThreshold != null ? humThreshold.getWarningType() : "电导率预警"; String humRemark = humThreshold != null ? humThreshold.getRemark() : "环境电导率报警"; boolean humShouldAlarm = false; BigDecimal humValue = BigDecimal.valueOf(finalValue); if (humMinValue != null && finalValue <= humMinValue) { humShouldAlarm = true; } if (humMaxValue != null && finalValue >= humMaxValue) { humShouldAlarm = true; } if (humShouldAlarm) { alarmUtil.saveAlarm(deviceId, humWarningType, humWarningCode, humMinValue != null ? BigDecimal.valueOf(humMinValue) : null, humMaxValue != null ? BigDecimal.valueOf(humMaxValue) : null, humValue, humRemark); } } if (nodeId == 5 && !Float.isNaN(nd.getHum())) { String humWarningCode = "WARN-PH"; WarningThreshold humThreshold = checkThreshold(deviceId, humWarningCode); Double humMinValue = humThreshold != null ? humThreshold.getMinValue() : null; Double humMaxValue = humThreshold != null ? humThreshold.getMaxValue() : null; String humWarningType = humThreshold != null ? humThreshold.getWarningType() : "PH预警"; String humRemark = humThreshold != null ? humThreshold.getRemark() : "环境PH报警"; boolean humShouldAlarm = false; BigDecimal humValue = BigDecimal.valueOf(finalValue); if (humMinValue != null && finalValue <= humMinValue) { humShouldAlarm = true; } if (humMaxValue != null && finalValue >= humMaxValue) { humShouldAlarm = true; } if (humShouldAlarm) { alarmUtil.saveAlarm(deviceId, humWarningType, humWarningCode, humMinValue != null ? BigDecimal.valueOf(humMinValue) : null, humMaxValue != null ? BigDecimal.valueOf(humMaxValue) : null, humValue, humRemark); } } // 获取手机号并发送短信 List devicePhoneList = devicePhoneFetchUtil.getPhoneListByDeviceId(deviceId); smsUtil.checkDeviceAlarmAndSend(data, nd, devicePhoneList); } catch (Exception e) { log.error("实时数据入库失败:设备ID={}, 节点ID={}", data.getDeviceId(), nd.getNodeId(), e); } } } @Override public void receiveLoginData(LoginData data) { System.out.println("登录->设备地址:" + data.getDeviceId()); } @Override public void receiveParamIds(ParamIdsData data) { String str = "设备参数编号列表->设备编号:" + data.getDeviceId() + "\t参数总数量:" + data.getTotalCount() + "\t本帧参数数量:" + data.getCount() + "\r\n"; for (int paramId : data.getPararmIdList()) { str += paramId + ","; } System.out.println(str); } @Override public void receiveParam(ParamData data) { String str = "设备参数->设备编号:" + data.getDeviceId() + "\r\n"; for (ParamItem pararm : data.getParameterList()) { str += "参数编号:" + pararm.getParamId() + "\t参数描述:" + pararm.getDescription() + "\t参数值:" + (pararm.getValueDescription() == null ? pararm .getValue() : pararm.getValueDescription() .get(pararm.getValue())) + "\r\n"; } System.out.println(str); } @Override public void receiveWriteParamAck(WriteParamAck data) { String str = "下载设备参数->设备编号:" + data.getDeviceId() + "\t参数数量:" + data.getCount() + "\t" + (data.isSuccess() ? "下载成功" : "下载失败"); System.out.println(str); } @Override public void receiveTransDataAck(TransDataAck data) { String str = "数据透传->设备编号:" + data.getDeviceId() + "\t响应结果:" + data.getData() + "\r\n字节数:" + data.getTransDataLen(); System.out.println(str); } @Override public void receiveHeartbeatData(HeartbeatData heartbeatData) { } }); log.info("开始启动RSServer(阻塞式监听)"); long startTime = System.currentTimeMillis(); rsServer.start(); // 正常运行时不返回,停止时执行后续日志 log.info("RSServer已启动并监听端口:9801,启动耗时:{}ms", System.currentTimeMillis() - startTime); } catch (Exception e) { log.error("RSServer初始化/启动异常", e); throw new RuntimeException("RSServer启动失败", e); } } /** * 关闭 */ @PreDestroy public void destroy() { log.info("开始关闭RSServer资源"); // 1. 停止服务器 if (rsServer != null) { try { rsServer.stop(); log.info("RSServer服务器已停止"); } catch (Exception e) { log.error("停止RSServer时发生异常", e); } } // 2. 关闭线程池 serverExecutor.shutdown(); try { if (!serverExecutor.awaitTermination(5, TimeUnit.SECONDS)) { serverExecutor.shutdownNow(); log.warn("RSServer线程池优雅关闭超时,已强制关闭"); } else { log.info("RSServer线程池已正常关闭"); } } catch (InterruptedException e) { serverExecutor.shutdownNow(); log.warn("关闭线程池时被中断", e); Thread.currentThread().interrupt(); } log.info("RSServer资源关闭完成"); } private WarningThreshold checkThreshold(String deviceCode, String warningCode) { try { return warningThresholdService.getWarningThresholdByDeviceAndCode(deviceCode, warningCode); } catch (Exception e) { log.error("查询预警阈值失败:deviceCode={}, warningCode={}", deviceCode, warningCode, e); return null; } } }