package com.zksy.radar.utils; import static com.zksy.common.utils.CollectionSideEffects.*; import com.zksy.api.domain.AlarmData; import com.zksy.api.domain.WarningThreshold; import com.zksy.api.service.AlarmDataService; import com.zksy.api.service.EarlyWarningAutoService; import com.zksy.api.service.WarningThresholdService; import com.zksy.api.utils.AlarmLevelUtil; import com.zksy.api.utils.SmsUtil; import com.zksy.common.exception.InvalidMessageException; import com.zksy.radar.domain.RadarData; import com.zksy.radar.service.RadarDataService; import com.zksy.api.service.DeviceContactService; import io.netty.buffer.ByteBuf; import io.netty.util.ReferenceCountUtil; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.handler.timeout.ReadTimeoutException; import lombok.extern.slf4j.Slf4j; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.http.HttpEntity; import org.springframework.http.HttpHeaders; import org.springframework.http.MediaType; import org.springframework.stereotype.Component; import org.springframework.web.client.RestTemplate; import java.math.BigDecimal; import java.time.LocalDateTime; import java.util.HashMap; import java.util.List; import java.util.Map; @ChannelHandler.Sharable @Slf4j @Component public class MessageHandler extends ChannelInboundHandlerAdapter { private static Logger logger = LoggerFactory.getLogger(MessageHandler.class); private final RadarDataService service; private final WarningThresholdService warningThresholdService; private final AlarmDataService alarmDataService; private final EarlyWarningAutoService earlyWarningAutoService; private final SmsUtil smsUtil; private final DeviceContactService devicePhoneFetchUtil; private final RestTemplate restTemplate; @Autowired private DeviceOfflineCheckTask deviceOfflineCheckTask; @Value("${device.status.api.url:http://zk-api-service/equipmentStatus/updateAlarmStatus}") private String deviceStatusApiUrl; @Autowired public MessageHandler(RadarDataService RadarDataService, WarningThresholdService warningThresholdService, AlarmDataService alarmDataService, EarlyWarningAutoService earlyWarningAutoService, SmsUtil smsUtil, DeviceContactService devicePhoneFetchUtil, RestTemplate restTemplate) { this.service = RadarDataService; this.warningThresholdService = warningThresholdService; this.alarmDataService = alarmDataService; this.earlyWarningAutoService = earlyWarningAutoService; this.smsUtil = smsUtil; this.devicePhoneFetchUtil = devicePhoneFetchUtil; this.restTemplate = restTemplate; } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { super.channelActive(ctx); //sendDataToDevice(ctx); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { try { ByteBuf in = (ByteBuf) msg; byte[] msgBytes = new byte[in.readableBytes()]; in.readBytes(msgBytes); logger.debug("接收到的原始数据帧: {}", printHexBinary(msgBytes)); // 提取帧类型 byte frameType = msgBytes[6]; logger.debug("帧类型: 0x{}", String.format("%02X", frameType)); // 1. 完整协议校验 DataParser.validateMessage(msgBytes); logger.info("数据帧校验通过"); // 2. 数据解析 RadarData resultData = DataParser.parseMessage(msgBytes); if (resultData.getSystemIdentifier() == null) { resultData.setSystemIdentifier("123456"); } // 3. 根据帧类型处理 if (frameType == 0x31) { if (msgBytes.length > 34) { byte flagByte = msgBytes[34]; String binary = String.format("%8s", Integer.toBinaryString(flagByte & 0xFF)).replace(' ', '0'); boolean isEnd = (flagByte & 0x08) != 0; logger.debug("结束位检测: 字节=0x{}, 二进制={}, 第5位(结束标识)={}", String.format("%02X", flagByte), binary, isEnd ? "1(结束)" : "0(未结束)"); } else { logger.warn("数据长度不足,无法检测结束位"); } // 入库 requireSaved(service.save(resultData)); String deviceCode = resultData.getDeviceCode(); run(deviceCode, "接收时间更新", () -> DeviceOfflineCheckTask.deviceLastReceiveTimeMap.put(deviceCode, new java.util.Date())); run(deviceCode, "采集应答", () -> { byte[] response = ProtocolUtils.buildCustomReplyFrame(msgBytes); ctx.writeAndFlush(Unpooled.copiedBuffer(response)); if (Boolean.TRUE.equals(resultData.getIsLastPacket())) { byte[] endResponse = ProtocolUtils.buildEndReplyFrame(msgBytes); ctx.executor().schedule(() -> run(deviceCode, "结束应答", () -> { if (ctx.channel().isActive()) { ctx.writeAndFlush(Unpooled.copiedBuffer(endResponse)); } }), 1, java.util.concurrent.TimeUnit.SECONDS); } }); run(deviceCode, "在线状态同步", () -> deviceOfflineCheckTask.markDeviceOnline(deviceCode)); run(deviceCode, "报警预警评估", () -> handleRadarAlarm(resultData)); } else if (frameType == 0x34) { /*logger.info("收到结束通讯帧,准备发送回应包"); byte[] response = ProtocolUtils.buildShutdownAckPacket(msgBytes); ctx.writeAndFlush(Unpooled.copiedBuffer(response)); logger.debug("已发送结束通讯回应包");*/ } else { logger.warn("收到未知帧类型: 0x{}", String.format("%02X", frameType)); } } catch (InvalidMessageException e) { logger.error("数据校验失败,不入库: {}", e.getMessage()); sendErrorResponse(ctx, "数据校验失败"); } catch (Exception e) { logger.error("数据解析或入库异常", e); sendErrorResponse(ctx, "数据处理异常"); } finally { ReferenceCountUtil.release(msg); } } // 工具方法:发送错误响应 private void sendErrorResponse(ChannelHandlerContext ctx, String msg) { ctx.writeAndFlush(Unpooled.copiedBuffer(("数据处理失败: " + msg).getBytes())); } private String printHexBinary(byte[] bytes) { StringBuilder sb = new StringBuilder(); for (byte b : bytes) { sb.append(String.format("%02X ", b)); } System.out.println("Received raw data: " + sb.toString()); logger.info("Received raw data: " + sb.toString()); return sb.toString(); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { if (cause instanceof ReadTimeoutException) { logger.info("来自" + ctx.channel().remoteAddress() + "的连接超时断开"); } else { logger.error("设备连接异常", cause); logger.info("来自" + ctx.channel().remoteAddress() + "的连接异常断开"); ctx.close(); } } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } @Override public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { logger.info("来自" + ctx.channel().remoteAddress() + "的连接主动断开"); ctx.fireChannelUnregistered(); } // 主动向设备发送数据 public void sendDataToDevice(ChannelHandlerContext ctx) { try { // 将字符串形式的十六进制数据转换为字节数组 String hexData = "01 03 00 00 00 01 84 0A "; String[] hexArray = hexData.split(" "); byte[] dataBytes = new byte[hexArray.length]; for (int i = 0; i < hexArray.length; i++) { dataBytes[i] = (byte) Integer.parseInt(hexArray[i], 16); } ByteBuf byteBuf = Unpooled.copiedBuffer(dataBytes); ctx.writeAndFlush(byteBuf); logger.info("已向设备发送数据: {}", printHexBinary(dataBytes)); } catch (Exception e) { logger.error("发送数据失败: {}", e.getMessage()); } } /** * 处理雷达设备告警 */ private void handleRadarAlarm(RadarData resultData) { String deviceCode = resultData.getDeviceCode(); Integer status = alarmStatus( handleAlarm(deviceCode, "WARN_INSTANT_FLOW", "瞬时流量", resultData.getMeter1InstantFlow()), handleAlarm(deviceCode, "WARN_FLOW_SPEED", "流速", resultData.getFlowSpeed())); if (status != null) { updateDeviceAlarmStatus(deviceCode, status); } } /** * 处理单个告警 * @return 是否触发告警 */ private Boolean handleAlarm(String deviceCode, String warningCode, String defaultType, String valueStr) { if (valueStr == null) { return null; } return evaluate(deviceCode, warningCode, () -> { BigDecimal actual = new BigDecimal(valueStr); WarningThreshold threshold = warningThresholdService.getWarningThresholdByDeviceAndCode(deviceCode, warningCode); if (threshold == null || (threshold.getMinValue() == null && threshold.getMaxValue() == null)) { throw new IllegalStateException("阈值缺失,不能判断设备已恢复正常"); } String type = defaultType; BigDecimal min = threshold.getMinValue() == null ? null : BigDecimal.valueOf(threshold.getMinValue()); BigDecimal max = threshold.getMaxValue() == null ? null : BigDecimal.valueOf(threshold.getMaxValue()); boolean over = (min != null && actual.compareTo(min) <= 0) || (max != null && actual.compareTo(max) >= 0); if (over) { run(deviceCode, warningCode + ":报警入库及通知", () -> saveAlarmData(deviceCode, type, warningCode, min, max, actual, null)); run(deviceCode, warningCode + ":关闭预警", () -> earlyWarningAutoService.closeAutoWarning(deviceCode, warningCode, "已达报警阈值,转报警处理")); } else { run(deviceCode, warningCode + ":生成预警", () -> earlyWarningAutoService.checkAndGenerate(deviceCode, type, warningCode, min, max, actual, null)); } return over; }); } /** * 保存告警数据到 alarm_data 表 */ private void saveAlarmData(String deviceCode, String warningType, String warningCode, BigDecimal minValue, BigDecimal maxValue, BigDecimal actualValue, String remark) { try { AlarmData alarmData = new AlarmData(); alarmData.setDeviceCode(deviceCode); alarmData.setWarningType(warningType); alarmData.setWarningCode(warningCode); alarmData.setMinValue(minValue); alarmData.setMaxValue(maxValue); alarmData.setActualValue(actualValue); int alarmLevel = AlarmLevelUtil.calculateLevel(actualValue, minValue, maxValue); BigDecimal deviationRatio = AlarmLevelUtil.calculateDeviationRatio(actualValue, minValue, maxValue); alarmData.setAlarmLevel(alarmLevel); alarmData.setDeviationRatio(deviationRatio); alarmData.setAlarmStatus(0); alarmData.setAlarmTime(LocalDateTime.now()); alarmData.setRemark(remark); alarmData.setCreateTime(LocalDateTime.now()); boolean saved = run(deviceCode, warningCode + ":报警入库", () -> requireSaved(alarmDataService.saveAlarmData(alarmData))); if (saved) { logger.info("告警数据入库成功: deviceCode={}, warningCode={}, level={}级", deviceCode, warningCode, alarmLevel); } if (alarmLevel == 1) { logger.warn("【1级紧急告警】deviceCode={}, warningCode={}, deviationRatio={}%", deviceCode, warningCode, deviationRatio); sendAlarmSms(deviceCode, warningType, actualValue, minValue, maxValue); } } catch (Exception e) { logger.error("保存告警数据失败", e); } } private void sendAlarmSms(String deviceCode, String warningType, BigDecimal actualValue, BigDecimal minValue, BigDecimal maxValue) { try { List phoneList = devicePhoneFetchUtil.getPhoneListByDeviceCode(deviceCode); if (phoneList == null || phoneList.isEmpty()) { logger.warn("报警短信发送失败:未找到设备{}的维护人员手机号", deviceCode); return; } com.alibaba.fastjson.JSONObject params = new com.alibaba.fastjson.JSONObject(); params.put("deviceNo", deviceCode); params.put("alarmType", warningType); params.put("location", "当前值:" + actualValue + ",正常范围:" + minValue + "~" + maxValue); Map result = smsUtil.sendBatchSms(phoneList, params.toJSONString()); long successCount = result.values().stream().filter(Boolean::booleanValue).count(); logger.info("1级告警短信发送完成,成功{}条,失败{}条", successCount, phoneList.size() - successCount); } catch (Exception e) { logger.error("发送报警短信异常: {}", e.getMessage()); } } /** * 更新设备报警状态 * @param deviceCode 设备编码 * @param alarmStatus 报警状态:0-正常,1-报警 */ private void updateDeviceAlarmStatus(String deviceCode, int alarmStatus) { try { Map params = new HashMap<>(); params.put("deviceCode", deviceCode); params.put("alarmStatus", alarmStatus); HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.APPLICATION_JSON); HttpEntity> request = new HttpEntity<>(params, headers); Map response = restTemplate.postForObject(deviceStatusApiUrl, request, Map.class); if (response == null || !Boolean.TRUE.equals(response.get("success"))) { logger.warn("设备报警状态更新未成功: deviceCode={}, alarmStatus={}, response={}", deviceCode, alarmStatus, response); return; } logger.info("设备报警状态更新成功: deviceCode={}, alarmStatus={}", deviceCode, alarmStatus); } catch (Exception e) { logger.error("更新设备报警状态失败: deviceCode={}, alarmStatus={}", deviceCode, alarmStatus, e); } } }