pipe-ner-server/environment-service/src/main/java/com/zksy/environment/config/RSServerService.java

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<String> 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;
        }
    }
}