refactor(audio): 重构音频服务处理逻辑并优化设备离线检测

- 替换 DevicePhoneFetchUtil 为 DeviceContactService 并更新方法调用
- 优化音频 RMS 计算算法，提升性能和准确性
- 改进协议解析逻辑，精确识别设备编码和功能码
- 重构设备离线检查任务，增加并发安全控制
- 优化音频上传处理流程，实现防重传和异常处理
- 调整告警参数配置，降低报警阈值提高敏感性
- 统一设备状态管理，改善在线离线状态同步机制

audio-service/src/main/java/com/zksy/audio/config/NettyServerThread.java

@@ -1,6 +1,7 @@
 package com.zksy.audio.config;
 
 import com.zksy.audio.utils.MessageHandler;
+import com.zksy.audio.utils.AudioFrameDecoder;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelInitializer;
@@ -10,6 +11,8 @@ import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.timeout.ReadTimeoutHandler;
+import io.netty.util.concurrent.DefaultEventExecutorGroup;
+import io.netty.util.concurrent.EventExecutorGroup;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -22,6 +25,12 @@ import org.springframework.stereotype.Component;
 public class NettyServerThread {
 	@Value("${netty.port:8313}")
 	private int port;
+    @Value("${collection.worker-threads:16}")
+    private int businessThreads = 16;
+    @Value("${wlds.raw-pcm-bytes:0}")
+    private int rawPcmBytes;
+    @Value("${device.offline.timeout-minutes:1560}")
+    private int offlineTimeoutMinutes = 1560;
 	private static Logger logger = LoggerFactory.getLogger(NettyServerThread.class);
 	private final MessageHandler messageHandler;
 	@Autowired
@@ -32,6 +41,7 @@ public class NettyServerThread {
 		System.out.println("Netty服务启动端口号" + port);
 		EventLoopGroup acceptor = new NioEventLoopGroup();
 		EventLoopGroup worker = new NioEventLoopGroup();
+        EventExecutorGroup business = new DefaultEventExecutorGroup(businessThreads, null, 1024, io.netty.util.concurrent.RejectedExecutionHandlers.reject());
 		NettyServer.acceptor = acceptor;
 		NettyServer.worker = worker;
 		ServerBootstrap bootstrap = new ServerBootstrap();
@@ -52,8 +62,7 @@ public class NettyServerThread {
 			public void initChannel(SocketChannel ch) throws Exception {
 				logger.info("来自" + ch.remoteAddress() + "的新连接接入");
 				// 注册handler
-				ch.pipeline().addLast(new ReadTimeoutHandler(3600));// 超时时间,1小时内没有从通道（Channel）读取到任何数据
-				ch.pipeline().addLast(messageHandler);
+                configurePipeline(ch.pipeline(), business);
 			}
 		});
 
@@ -66,6 +75,7 @@ public class NettyServerThread {
 		} catch (InterruptedException e) {
 			e.printStackTrace();
 		} finally {
+            business.shutdownGracefully();
 			if (acceptor != null) {
 				acceptor.shutdownGracefully();
 			}
@@ -74,4 +84,11 @@ public class NettyServerThread {
 			}
 		}
 	}
-}
+
+    public void configurePipeline(io.netty.channel.ChannelPipeline pipeline, EventExecutorGroup business) {
+                // 每日上报的长连接不能在1小时后被关闭，读取超时与26小时离线配置一致。
+                pipeline.addLast(new ReadTimeoutHandler(offlineTimeoutMinutes, java.util.concurrent.TimeUnit.MINUTES));
+				pipeline.addLast(new AudioFrameDecoder(rawPcmBytes));
+				pipeline.addLast(business, messageHandler);
+    }
+}

audio-service/src/main/java/com/zksy/audio/service/NoiseLeakAlarmService.java

@@ -25,11 +25,12 @@ import java.util.concurrent.ConcurrentHashMap;
 /**
  * 漏失监测告警服务
  * <p>
- * 采用"时间持续性 + 漏水特征频段过滤"策略（漏失噪音是持续的，瞬时环境噪音不应触发）：
+ * 设备每24小时上传正常或异常录音，单次达到RMS阈值且满足漏损频段即一级告警。
+ * 同一录音重传由接收层去重；保留配置多次确认的能力，供其他采样周期使用：
  * <ul>
  *   <li>rms 越限阈值(max_value)、恢复阈值(min_value) 均从 {@code warning_threshold} 表按设备编码获取，与系统现有逻辑一致</li>
- *   <li>出现单次 rms≥max 且在漏水特征频段 → <b>预警</b>（疑似漏失）</li>
- *   <li>最近窗口内累计 N 次 rms≥max 且在漏水特征频段 → <b>告警</b>（确认漏失），并解除预警</li>
+ *   <li>alarm-count=1：每份独立录音 rms≥max 且在漏水特征频段 → <b>一级告警及短信</b></li>
+ *   <li>alarm-count>1：窗口内累计 N 次确认，次数不足时预警</li>
  *   <li>rms 回落≤min（或不在特征频段）→ 解除预警，重置确认态</li>
  * </ul>
  */
@@ -44,7 +45,7 @@ public class NoiseLeakAlarmService {
 
     private static final DateTimeFormatter TS_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
 
-    /** 已确认漏失的设备编码，用于避免同一漏失事件内重复告警 */
+    /** 已确认状态用于恢复与中间带保持；单次模式下每份独立异常录音均告警。 */
     private final Set<String> confirmedLeakEncode = ConcurrentHashMap.newKeySet();
 
     @Value("${wlds.leak.freq-min:200.0}")
@@ -56,8 +57,8 @@ public class NoiseLeakAlarmService {
     @Value("${wlds.leak.window-minutes:30}")
     private int windowMinutes;
 
-    @Value("${wlds.leak.alarm-count:3}")
-    private int alarmCount;
+    @Value("${wlds.leak.alarm-count:1}")
+    private int alarmCount = 1;
 
     @Autowired
     private NoiseInfoMapper noiseInfoMapper;
@@ -70,6 +71,8 @@ public class NoiseLeakAlarmService {
 
     @Autowired
     private WarningThresholdService warningThresholdService;
+    @Autowired
+    private com.zksy.api.utils.CollectionDeviceNotifier deviceNotifier;
 
     /**
      * 对一条新入库的噪音数据进行漏失预警/告警评估
@@ -107,43 +110,55 @@ public class NoiseLeakAlarmService {
             return;
         }
 
+        if (noise.getCenterFrequency() == null || !Double.isFinite(noise.getCenterFrequency())) {
+            throw new IllegalArgumentException("缺少有效音频频率，保留原报警状态");
+        }
         boolean freqInBand = noise.getCenterFrequency() != null
                 && noise.getCenterFrequency() >= lowerFreq && noise.getCenterFrequency() <= upperFreq;
 
         // 1) 特征频段消失：漏水特征不再成立，彻底解除
         if (!freqInBand) {
             confirmedLeakEncode.remove(encode);
-            earlyWarningAutoService.closeAutoWarning(encode, LEAK_WARNING_CODE, "漏失特征频段消失，解除预警");
+            run(encode, "解除预警", () -> earlyWarningAutoService.closeAutoWarning(encode, LEAK_WARNING_CODE, "漏失特征频段消失，解除预警"));
+            run(encode, "报警状态恢复", () -> deviceNotifier.updateAlarmStatus(encode, 0));
             return;
         }
 
         // 2) rms 回落至恢复阈值(≤min)：解除预警并重置确认态
         if (minValue != null && rms <= minValue) {
             confirmedLeakEncode.remove(encode);
-            earlyWarningAutoService.closeAutoWarning(encode, LEAK_WARNING_CODE, "rms回落至恢复阈值，解除预警");
+            run(encode, "解除预警", () -> earlyWarningAutoService.closeAutoWarning(encode, LEAK_WARNING_CODE, "rms回落至恢复阈值，解除预警"));
+            run(encode, "报警状态恢复", () -> deviceNotifier.updateAlarmStatus(encode, 0));
             return;
         }
 
         // 3) 中间带（min < rms < max）：保持现状，不做级别变更，避免边界抖动
         if (rms < maxValue) {
+            if (confirmedLeakEncode.contains(encode)) {
+                run(encode, "报警状态同步", () -> deviceNotifier.updateAlarmStatus(encode, 1));
+            }
             return;
         }
 
-        // 4) rms >= max：疑似→确认（时间持续性）
-        long count = countQualified(encode, maxValue, lowerFreq, upperFreq);
+        // 单次模式只依赖本次已保存的录音，不再查询历史计数。
+        long count = alarmCount == 1 ? 1 : countQualified(encode, maxValue, lowerFreq, upperFreq);
         if (count >= alarmCount) {
-            if (confirmedLeakEncode.add(encode)) {
+            boolean newlyConfirmed = confirmedLeakEncode.add(encode);
+            if (alarmCount == 1 || newlyConfirmed) {
+                run(encode, "一级漏失告警短信", () -> deviceNotifier.sendLevelOneSms(encode, "漏失报警", "RMS=" + rms));
                 if (saveLeakAlarm(encode, rms, count, lowerFreq, upperFreq)) {
-                    log.info("确认漏失告警: encode={}, 窗口累计{}次", encode, count);
+                    log.info("确认漏失告警: encode={}, 本次确认采用{}份独立采样", encode, count);
                 } else {
                     confirmedLeakEncode.remove(encode);
                 }
             }
-            earlyWarningAutoService.closeAutoWarning(encode, LEAK_WARNING_CODE, "已确认漏失，转告警处理");
+            run(encode, "解除预警", () -> earlyWarningAutoService.closeAutoWarning(encode, LEAK_WARNING_CODE, "已确认漏失，转告警处理"));
+            run(encode, "报警状态同步", () -> deviceNotifier.updateAlarmStatus(encode, 1));
         } else {
             confirmedLeakEncode.remove(encode);
-            earlyWarningAutoService.recordWarning(encode, LEAK_WARNING_TYPE, LEAK_WARNING_CODE, 3,
-                    buildWarningContent(encode, count, maxValue, lowerFreq, upperFreq));
+            final double lower = lowerFreq, upper = upperFreq;
+            run(encode, "疑似漏失预警", () -> earlyWarningAutoService.recordWarning(encode, LEAK_WARNING_TYPE, LEAK_WARNING_CODE, 3,
+                    buildWarningContent(encode, count, maxValue, lower, upper)));
         }
     }
 
@@ -179,8 +194,10 @@ public class NoiseLeakAlarmService {
         alarm.setAlarmStatus(0);
         alarm.setAlarmTime(LocalDateTime.now());
         alarm.setCreateTime(LocalDateTime.now());
-        alarm.setRemark(String.format("漏失监测确认漏失：最近%d分钟内%d次采样rms≥阈值且在漏失频段[%.0f,%.0f]Hz",
-                windowMinutes, count, lowerFreq, upperFreq));
+        alarm.setRemark(alarmCount == 1
+                ? String.format("漏失监测单次独立采样确认：rms≥阈值且在漏失频段[%.0f,%.0f]Hz", lowerFreq, upperFreq)
+                : String.format("漏失监测确认漏失：最近%d分钟内%d次采样rms≥阈值且在漏失频段[%.0f,%.0f]Hz",
+                    windowMinutes, count, lowerFreq, upperFreq));
         try {
             requireSaved(alarmDataService.saveAlarmData(alarm));
             log.info("漏失告警入库成功: encode={}, count={}", encode, count);
@@ -195,4 +212,4 @@ public class NoiseLeakAlarmService {
         return String.format("【系统自动预警】设备[%s]疑似漏失：最近%d分钟内%d次采样rms≥%s且中心频率落在漏失频段[%.0f,%.0f]Hz，待确认",
                 encode, windowMinutes, count, Objects.toString(maxValue, "-"), lowerFreq, upperFreq);
     }
-}
+}

audio-service/src/main/java/com/zksy/audio/utils/AudioFrameDecoder.java

@@ -0,0 +1,58 @@
+package com.zksy.audio.utils;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+/** 身份/参数按CRLF分行，WAV按RIFF长度组帧；PCM内容不能按文本分隔。 */
+public final class AudioFrameDecoder extends ByteToMessageDecoder {
+    private static final int MAX_WAV_BYTES = 8 * 1024 * 1024;
+    private final int rawPcmBytes;
+    public AudioFrameDecoder() { this(0); }
+    public AudioFrameDecoder(int rawPcmBytes) {
+        if (rawPcmBytes < 0 || rawPcmBytes > MAX_WAV_BYTES - 44 || rawPcmBytes % 2 != 0) {
+            throw new IllegalArgumentException("PCM采样长度必须为偶数且不超过8MiB");
+        }
+        this.rawPcmBytes = rawPcmBytes;
+    }
+    @Override
+    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
+        while (in.readableBytes() >= 4) {
+            int start = in.readerIndex();
+            String prefix = in.toString(start, 4, StandardCharsets.US_ASCII);
+            if (prefix.startsWith("$$$") || prefix.equals("VBAT") || prefix.equals("Gain") || prefix.startsWith("OK")) {
+                int end = -1;
+                for (int i = start; i + 1 < in.writerIndex(); i++) {
+                    if (in.getByte(i) == '\r' && in.getByte(i + 1) == '\n') { end = i + 2; break; }
+                }
+                if (end == -1) {
+                    if (in.readableBytes() > 512) in.skipBytes(in.readableBytes());
+                    return;
+                }
+                out.add(in.readRetainedSlice(end - start));
+            } else if (prefix.equals("RIFF")) {
+                if (in.readableBytes() < 12) return;
+                long length = in.getUnsignedIntLE(start + 4) + 8;
+                if (length < 44 || length > MAX_WAV_BYTES || in.getInt(start + 8) != 0x57415645) {
+                    in.skipBytes(4);
+                    continue;
+                }
+                if (in.readableBytes() < length) return;
+                out.add(in.readRetainedSlice((int) length));
+            } else if (rawPcmBytes > 0) {
+                if (in.readableBytes() < rawPcmBytes) return;
+                ByteBuf wav = Unpooled.buffer(44 + rawPcmBytes);
+                wav.writeInt(0x52494646).writeIntLE(36 + rawPcmBytes).writeInt(0x57415645);
+                wav.writeInt(0x666D7420).writeIntLE(16).writeShortLE(1).writeShortLE(1);
+                wav.writeIntLE(8192).writeIntLE(16384).writeShortLE(2).writeShortLE(16);
+                wav.writeInt(0x64617461).writeIntLE(rawPcmBytes).writeBytes(in, rawPcmBytes);
+                out.add(wav);
+            } else {
+                in.skipBytes(1);
+            }
+        }
+    }
+}

audio-service/src/main/java/com/zksy/audio/utils/AudioUploadDeduplicator.java

@@ -0,0 +1,49 @@
+package com.zksy.audio.utils;
+
+import java.security.MessageDigest;
+import java.util.*;
+import java.util.function.LongSupplier;
+
+/** 单实例内按设备和完整录音内容去重；连续重传刷新静默窗口，失败不占用成功记录。 */
+final class AudioUploadDeduplicator {
+    @FunctionalInterface interface Upload { void save() throws Exception; }
+    private final Object[] locks = new Object[128];
+    private final Map<String, Long> completed = new HashMap<>();
+    private final LongSupplier clock;
+    private static final int CAPACITY = 100_000;
+    private long nextCleanup;
+
+    AudioUploadDeduplicator() { this(System::nanoTime); }
+    AudioUploadDeduplicator(LongSupplier clock) {
+        this.clock = clock;
+        Arrays.setAll(locks, i -> new Object());
+    }
+
+    boolean receive(String device, byte[] wav, long windowSeconds, Upload upload) throws Exception {
+        if (windowSeconds <= 0 || windowSeconds > 86400) {
+            throw new IllegalStateException("音频重传静默窗口必须为1~86400秒");
+        }
+        String key = device + ':' + Base64.getEncoder().encodeToString(MessageDigest.getInstance("SHA-256").digest(wav));
+        synchronized (locks[(device.hashCode() & Integer.MAX_VALUE) % locks.length]) {
+            long now = clock.getAsLong();
+            synchronized (completed) {
+                if (now >= nextCleanup || completed.size() >= CAPACITY) {
+                    completed.values().removeIf(expiry -> expiry <= now);
+                    nextCleanup = now + 1_000_000_000L;
+                }
+                Long expiry = completed.get(key);
+                if (expiry != null && expiry > now) {
+                    // 设备未收到OK会一直上传；连续重传不能因距首次成功已超时而再次入库。
+                    completed.put(key, now + windowSeconds * 1_000_000_000L);
+                    return false;
+                }
+                if (completed.size() >= CAPACITY) throw new IllegalStateException("音频去重缓存已满，保留重试机会");
+            }
+            upload.save();
+            synchronized (completed) {
+                completed.put(key, clock.getAsLong() + windowSeconds * 1_000_000_000L);
+            }
+            return true;
+        }
+    }
+}

audio-service/src/main/java/com/zksy/audio/utils/DataParser.java

@@ -27,66 +27,29 @@ public class DataParser {
     }
 
     public static int calculateRMS(String filePath) {
-        try {
-            File audioFile = new File(filePath);
-            AudioInputStream audioStream = AudioSystem.getAudioInputStream(audioFile);
-            AudioFormat format = audioStream.getFormat();
-            if (format.getEncoding() != AudioFormat.Encoding.PCM_SIGNED && format.getSampleSizeInBits() != 16) {
-                throw new UnsupportedAudioFileException("Only 16-bit PCM signed WAV files are supported.");
+        try (AudioInputStream audio = AudioSystem.getAudioInputStream(new File(filePath))) {
+            AudioFormat f = audio.getFormat();
+            if (f.getEncoding() != AudioFormat.Encoding.PCM_SIGNED || f.getSampleSizeInBits() != 16
+                    || f.getChannels() != 1 || f.getSampleRate() != 8192) {
+                throw new IllegalArgumentException("需要单声道16位8192Hz PCM音频");
             }
-
-            int blockAlign = (format.getSampleSizeInBits() / 8) * format.getChannels();
-
-            int sampleCount = 0;
-            double sumOfSquares = 0;
-            double sum = 0;
-
-            byte[] rawBuffer = new byte[blockAlign];
-            int bytesRead;
-
-            while ((bytesRead = audioStream.read(rawBuffer, 0, rawBuffer.length)) > 0) {
-
-                for (int i = 0; i < bytesRead / blockAlign; i++) {
-
-                    int sample = toInt16(rawBuffer, format.isBigEndian());
-
-
-                    sum += sample;  // 累加样本值
-                    sampleCount++;
-                }
-
+            long count = 0;
+            double sum = 0, squares = 0;
+            byte[] sample = new byte[2];
+            int n;
+            while ((n = audio.read(sample)) != -1) {
+                if (n != 2) throw new IllegalArgumentException("音频包含不完整采样");
+                int value = toInt16(sample, f.isBigEndian());
+                sum += value;
+                squares += (double) value * value;
+                count++;
             }
-
-            int mean = (int) (sum / sampleCount);
-
-            audioStream.close();
-            audioStream = AudioSystem.getAudioInputStream(audioFile);
-
-            sampleCount = 0;
-            sumOfSquares = 0;
-
-            // 从样本中减去平均值，计算新的平方和
-            while ((bytesRead = audioStream.read(rawBuffer, 0, rawBuffer.length)) > 0) {
-                for (int i = 0; i < bytesRead / blockAlign; i++) {
-                    int sample = toInt16(rawBuffer, format.isBigEndian());
-
-                    sample -= mean;  // 去掉平均值
-                    sumOfSquares += sample * sample;  // 累加平方值
-                    sampleCount++;
-                }
-            }
-
-            // 计算去直流后的 RMS
-            int rms = (int) Math.sqrt(sumOfSquares / sampleCount);
-
-            audioStream.close();
-
-            return rms;
-
-        } catch (Exception e) {
-            log.error(e.getMessage(), e);
+            if (count == 0) throw new IllegalArgumentException("音频采样为空");
+            double mean = sum / count;
+            return (int) Math.sqrt(Math.max(0, squares / count - mean * mean));
+        } catch (IOException | UnsupportedAudioFileException e) {
+            throw new IllegalArgumentException("音频读取失败，不能用0替代实际值", e);
         }
-        return 0;
     }
 
     public static short toInt16(byte[] array, int index) {

audio-service/src/main/java/com/zksy/audio/utils/DeviceOfflineCheckTask.java

@@ -15,6 +15,8 @@ import java.util.Date;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.time.Clock;
+import java.util.Arrays;
 
 @Component
 public class DeviceOfflineCheckTask {
@@ -23,25 +25,36 @@ public class DeviceOfflineCheckTask {
     public static ConcurrentHashMap<String, Date> deviceLastReceiveTimeMap = new ConcurrentHashMap<>();
 
     private final Set<String> offlineDeviceSet = ConcurrentHashMap.newKeySet();
+    private final Set<String> onlineDeviceSet = ConcurrentHashMap.newKeySet();
+    private final Object[] deviceLocks = new Object[128];
+    private Clock clock = Clock.systemUTC();
+
+    public DeviceOfflineCheckTask() {
+        Arrays.setAll(deviceLocks, i -> new Object());
+    }
 
     @Autowired
     private RestTemplate restTemplate;
 
-    @Value("${device.offline.timeout-minutes:30}")
-    private int offlineTimeoutMinutes;
+    // 每24小时上传一次，用户确认额外允许2小时延迟。
+    @Value("${device.offline.timeout-minutes:1560}")
+    private int offlineTimeoutMinutes = 1560;
 
     @Scheduled(fixedRateString = "${device.offline.check-interval-ms:300000}")
     public void checkDeviceOffline() {
-        Date now = new Date();
+        Date now = new Date(clock.millis());
         long timeoutMs = (long) offlineTimeoutMinutes * 60 * 1000;
         for (Map.Entry<String, Date> entry : deviceLastReceiveTimeMap.entrySet()) {
-            long diff = now.getTime() - entry.getValue().getTime();
-            if (diff > timeoutMs) {
+            synchronized (deviceLock(entry.getKey())) {
+                // 与在线恢复串行，且重新读取时间，避免旧快照把刚收到上报的设备置离线。
+                Date latest = deviceLastReceiveTimeMap.get(entry.getKey());
+                if (latest == null || now.getTime() - latest.getTime() <= timeoutMs) continue;
                 if (offlineDeviceSet.add(entry.getKey())) {
                     if (!updateDeviceOnlineStatus(entry.getKey(), 0)) {
                         offlineDeviceSet.remove(entry.getKey());
                         continue;
                     }
+                    onlineDeviceSet.remove(entry.getKey());
                     logger.info("设备 {} 已离线（超过{}分钟未收到数据）", entry.getKey(), offlineTimeoutMinutes);
                 }
             }
@@ -49,13 +62,21 @@ public class DeviceOfflineCheckTask {
     }
 
     public void markDeviceOnline(String deviceCode) {
-        if (deviceCode != null && offlineDeviceSet.contains(deviceCode)
-                && updateDeviceOnlineStatus(deviceCode, 1)) {
-            offlineDeviceSet.remove(deviceCode);
-            logger.info("设备 {} 恢复在线", deviceCode);
+        if (deviceCode == null || deviceCode.trim().isEmpty()) return;
+        synchronized (deviceLock(deviceCode)) {
+            // 首次上报/服务重启后也需同步，不能只依赖本实例曾经标记过离线。
+            if (!onlineDeviceSet.contains(deviceCode) && updateDeviceOnlineStatus(deviceCode, 1)) {
+                onlineDeviceSet.add(deviceCode);
+                offlineDeviceSet.remove(deviceCode);
+                logger.info("设备 {} 已在线", deviceCode);
+            }
         }
     }
 
+    private Object deviceLock(String deviceCode) {
+        return deviceLocks[(deviceCode.hashCode() & Integer.MAX_VALUE) % deviceLocks.length];
+    }
+
     private boolean updateDeviceOnlineStatus(String deviceCode, int onlineStatus) {
         try {
             Map<String, Object> params = Map.of("deviceCode", deviceCode, "onlineStatus", onlineStatus);
@@ -73,4 +94,4 @@ public class DeviceOfflineCheckTask {
             return false;
         }
     }
-}
+}

audio-service/src/main/java/com/zksy/audio/utils/MessageHandler.java

@@ -1,379 +1,138 @@
 package com.zksy.audio.utils;
 
 import static com.zksy.common.utils.CollectionSideEffects.*;
-
-import cn.hutool.core.convert.Convert;
-import cn.hutool.core.date.DateField;
-import cn.hutool.core.date.DatePattern;
-import cn.hutool.core.date.DateUtil;
-import cn.hutool.core.lang.Validator;
-import cn.hutool.core.thread.ThreadUtil;
-import cn.hutool.core.util.StrUtil;
 import com.zksy.audio.domain.ChannelInfo;
 import com.zksy.audio.domain.NoiseInfo;
 import com.zksy.audio.service.NoiseInfoService;
 import com.zksy.audio.service.NoiseLeakAlarmService;
-import com.zksy.audio.utils.CrtHelper;
 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.ChannelId;
-import io.netty.channel.ChannelInboundHandlerAdapter;
-import lombok.RequiredArgsConstructor;
+import io.netty.channel.*;
+import io.netty.util.ReferenceCountUtil;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.io.FileUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
-
-import java.io.File;
-import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
+import java.nio.file.*;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.*;
+import java.util.concurrent.*;
 
-/**
- * 消息处理器
- * 负责处理各种请求和响应
- */
+/** 只处理已组好的文本行或完整WAV，一份音频只计算和保存一次。 */
 @Slf4j
 @Component
 @ChannelHandler.Sharable
 public class MessageHandler extends ChannelInboundHandlerAdapter {
+    public final ConcurrentMap<ChannelId, ChannelInfo> channelGroup = new ConcurrentHashMap<>();
+    public final ConcurrentMap<String, ChannelInfo> deviceGroup = new ConcurrentHashMap<>();
+    @Value("${wlds.fileUploadBaseDir}") private String fileUploadBaseDir;
+    // 正常/异常录音每24小时上报；连续重传按滑动静默窗口去重，下一份录音独立处理。
+    @Value("${wlds.retransmission-window-seconds:120}") private long retransmissionWindowSeconds = 120;
+    private final AudioUploadDeduplicator uploads = new AudioUploadDeduplicator();
+    private final NoiseInfoService service;
+    @Autowired private DeviceOfflineCheckTask deviceOfflineCheckTask;
+    @Autowired private NoiseLeakAlarmService noiseLeakAlarmService;
+    @Autowired public MessageHandler(NoiseInfoService service) { this.service = service; }
+
+    @javax.annotation.PostConstruct
+    public void validateAudioTiming() {
+        if (retransmissionWindowSeconds <= 0 || retransmissionWindowSeconds > 86400) {
+            throw new IllegalArgumentException("音频重传静默窗口必须为1~86400秒");
+        }
+    }
 
-	public ConcurrentMap<ChannelId, ChannelInfo> channelGroup = new ConcurrentHashMap<>();
-
-	public ConcurrentMap<String, ChannelInfo> deviceGroup = new ConcurrentHashMap<>();
-
-	@Value("${wlds.fileUploadBaseDir}")
-	private String fileUploadBaseDir;
-
-	private final static String head = "524946462440010057415645666D742010000000010001000020000000400000020010006461746100400100";
-	private final NoiseInfoService service;
-	@Autowired
-	private DeviceOfflineCheckTask deviceOfflineCheckTask;
-	@Autowired
-	private NoiseLeakAlarmService noiseLeakAlarmService;
-	@Autowired
-	public MessageHandler(NoiseInfoService noiseInfoService) {
-		this.service = noiseInfoService;
-	}
-	@Override
-	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-		//log.info("==收到客户端：{}|{}发送的消息:{}============:", ctx.channel().id().asLongText(), ctx.channel().remoteAddress(), msg);
-		try {
-			ByteBuf byteBuffer = (ByteBuf) msg;
-			if (byteBuffer == null) {
-				return;
-			}
-			int readableLength = byteBuffer.readableBytes();
-			byte[] arrayWithOutHead = new byte[readableLength];//去掉包头标识的数据
-			byteBuffer.getBytes(byteBuffer.readerIndex(), arrayWithOutHead);
-
-			String hex = CrtHelper.byteToHexStr1(arrayWithOutHead);
-			log.info("收到消息:{}", hex);
-			double power = 0;
-			String deviceNo;
-			double gain = 0;
-
-			byte[] bytes;
-			String asciiString;
-			String path;
-
-			ChannelInfo info;
-
-			ChannelInfo channelInfo = new ChannelInfo("puqi", ctx.channel());
-
-			if (!channelGroup.containsKey(ctx.channel().id())) {
-				// 加入客户端
-				channelGroup.put(ctx.channel().id(), channelInfo);
-			}
-
-			info = channelGroup.get(ctx.channel().id());
-
-			// 申报身份
-			if (hex.startsWith("242424")) {
-				String reativePath = "";
-				String[] hexArray = Arrays.stream(hex.split("0D0A")).filter(StrUtil::isNotEmpty).toArray(String[]::new);
-				log.info("hexArray.length:{}", hexArray.length);
-				if (hexArray.length >= 1) {
-					bytes = CrtHelper.strToToHexByte1(hexArray[0]);
-					asciiString = new String(bytes, StandardCharsets.US_ASCII);
-					deviceNo = asciiString.replaceAll("\\$\\$\\$", "").replaceAll("\r\n", "").trim();
-					path = getPathName(deviceNo, reativePath);
-
-					info.setDeviceNo(deviceNo);
-					info.setPath(path);
-					info.setFirst(true);
-					info.setFilePath(path);
-					info.setLength(0);
-
-					deviceGroup.put(deviceNo, info);
-					channelGroup.put(ctx.channel().id(), info);
-					info = channelGroup.get(ctx.channel().id());
-				}
-				if (hexArray.length >= 2) {
-					bytes = CrtHelper.strToToHexByte1(hexArray[1]);
-					asciiString = new String(bytes, StandardCharsets.US_ASCII);
-					asciiString = asciiString.replaceAll("VBAT:", "").replaceAll("V", "").replaceAll("\r\n", "").trim();
-					if (Validator.isNumber(asciiString)) {
-						power = Convert.toDouble(asciiString);
-					}
-				}
-				if (hexArray.length >= 3) {
-					String gainHex = hexArray[3];
-					bytes = CrtHelper.strToToHexByte1(gainHex);
-					asciiString = new String(bytes, StandardCharsets.US_ASCII);
-					asciiString = asciiString.replaceAll("Gain:", "").replaceAll("\r\n", "");
-					if (Validator.isNumber(asciiString)) {
-						gain = Convert.toDouble(asciiString);
-					}
-				}
-
-				channelGroup.get(ctx.channel().id()).setPower(power);
-				channelGroup.get(ctx.channel().id()).setGain(gain);
-
-				if (hexArray.length >= 4) {
-					String videoHex = hexArray[3];
-					info = channelGroup.get(ctx.channel().id());
-					path = info.getPath();
-					info.setLength(info.getLength() + (videoHex.length() / 2));
-					String finalPath1 = path;
-					ThreadUtil.execAsync(() -> {
-						saveHexStringToFile(finalPath1, videoHex);
-					});
-				}
-
-			}
-			// 申报电量
-			else if (hex.startsWith("56424154")) {
-				if (!channelGroup.containsKey(ctx.channel().id())) {
-					return;
-				}
-				if (StrUtil.isEmpty(channelGroup.get(ctx.channel().id()).getDeviceNo())) {
-					return;
-				}
-				String[] hexArray = Arrays.stream(hex.split("0D0A")).filter(StrUtil::isNotEmpty).toArray(String[]::new);
-				if (hexArray.length >= 1) {
-					bytes = CrtHelper.strToToHexByte1(hexArray[0]);
-					asciiString = new String(bytes, StandardCharsets.US_ASCII);
-					asciiString = asciiString.replaceAll("VBAT:", "").replaceAll("V", "").replaceAll("\r\n", "").trim();
-					if (Validator.isNumber(asciiString)) {
-						power = Double.parseDouble(asciiString);
-					}
-				}
-				if (hexArray.length >= 2) {
-					String gainHex = hexArray[1];
-					bytes = CrtHelper.strToToHexByte1(gainHex);
-					asciiString = new String(bytes, StandardCharsets.US_ASCII);
-					asciiString = asciiString.replaceAll("Gain:", "").replaceAll("\r\n", "");
-					if (Validator.isNumber(asciiString)) {
-						gain = Double.parseDouble(asciiString);
-					}
-				}
-				channelGroup.get(ctx.channel().id()).setGain(gain);
-				channelGroup.get(ctx.channel().id()).setPower(power);
-
-				if (hexArray.length >= 3) {
-					String videoHex = hexArray[2];
-					info = channelGroup.get(ctx.channel().id());
-					path = info.getPath();
-					info.setLength(info.getLength() + (videoHex.length() / 2));
-					String finalPath1 = path;
-					ThreadUtil.execAsync(() -> {
-						saveHexStringToFile(finalPath1, videoHex);
-					});
-				}
-			}
-			// 增益值
-			else if (hex.startsWith("4761696E")) {
-				if (!channelGroup.containsKey(ctx.channel().id())) {
-					return;
-				}
-				if (StrUtil.isEmpty(channelGroup.get(ctx.channel().id()).getDeviceNo())) {
-					return;
-				}
-				String[] hexArray = Arrays.stream(hex.split("0D0A")).filter(StrUtil::isNotEmpty).toArray(String[]::new);
-				if (hexArray.length >= 1) {
-					String gainHex = hexArray[0];
-					bytes = CrtHelper.strToToHexByte1(gainHex);
-					asciiString = new String(bytes, StandardCharsets.US_ASCII);
-					asciiString = asciiString.replaceAll("Gain:", "").replaceAll("\r\n", "");
-					if (Validator.isNumber(asciiString)) {
-						gain = Double.parseDouble(asciiString);
-					}
-				}
-				channelGroup.get(ctx.channel().id()).setGain(gain);
-				if (hexArray.length >= 2) {
-					String videoHex = hexArray[1];
-					info = channelGroup.get(ctx.channel().id());
-					path = info.getPath();
-					info.setLength(info.getLength() + (videoHex.length() / 2));
-					String finalPath1 = path;
-					ThreadUtil.execAsync(() -> {
-						saveHexStringToFile(finalPath1, videoHex);
-					});
-				}
-			} else if (hex.startsWith("4F4B")) {
-				bytes = CrtHelper.strToToHexByte1(hex);
-				asciiString = new String(bytes, StandardCharsets.US_ASCII);
-				log.info("==收到客户端：{}，GPS信息：{}", ctx.channel().id().asLongText(), asciiString);
-				info = channelGroup.get(ctx.channel().id());
-				String pattern = ",(\\d+\\.\\d+),\\d+\\.\\d+";
-				Pattern regex = Pattern.compile(pattern);
-				Matcher match = regex.matcher(asciiString);
-				if (match.matches()) {
-					// 提取经纬度，注意第一个捕获组是纬度，第二个捕获组（如果有的话）是经度
-					// 但由于我们只关心第一个逗号后的经纬度，所以可以直接拆分
-					String[] coords = match.group(0).replaceFirst(",", "").split(",");
-					if (coords.length == 2) {
-						double lat = Double.parseDouble(coords[0]);
-						double lon = Double.parseDouble(coords[1]);
-						log.info("经纬度：{},{}", lat, lon);
-						//-----------
-					}
-				} else {
-					log.info("==收到客户端：{}的设备调试信息：{}", ctx.channel().id().asLongText(), asciiString);
-
-				}
-			} else {
-				if (!channelGroup.containsKey(ctx.channel().id())) {
-					return;
-				}
-				if (StrUtil.isEmpty(channelGroup.get(ctx.channel().id()).getDeviceNo())) {
-					return;
-				}
-				info = channelGroup.get(ctx.channel().id());
-				path = channelGroup.get(ctx.channel().id()).getPath();
-				String finalPath1 = path;
-				String headHex;
-				byte[] length = new byte[4];
-				if (info.getFirst() && !hex.startsWith("52494646") ){
-					//用于没带头文件的音频数据
-					headHex = head + hex;
-					long videoMaxSize = 80 * 1024 + 44;
-					info.setVideoMaxSize(videoMaxSize);
-
-				} else {
-					headHex = hex;
-					if (hex.startsWith("52494646")){
-						//24001900  ====> 163840
-						length[0] = (byte) Integer.parseInt(hex.substring(14, 16), 16);
-						length[1] = (byte) Integer.parseInt(hex.substring(12, 14), 16);
-						length[2] = (byte) Integer.parseInt(hex.substring(10, 12), 16);
-						length[3] = (byte) Integer.parseInt(hex.substring(8, 10), 16);
-						long videoMaxSize = (Long.parseLong(CrtHelper.byteToHexStr1(length), 16) - 36) / 10 + 44;
-						info.setVideoMaxSize(videoMaxSize);
-					}
-				}
-				info.setFirst(false);
-				info.setLength(info.getLength() + headHex.length() / 2);
-				ThreadUtil.execAsync(() -> {
-					saveHexStringToFile(finalPath1, headHex);
-				});
-				log.info("======收到客户端总字节{}-----videoMaxSize:{}",info.getLength(),info.getVideoMaxSize());
-				// 如果接收音频文件字节大于头文件标识长度；
-				if (info.getLength() >= info.getVideoMaxSize()) {
-					if (info.getGatheTime() == null) {
-						info.setGatheTime(DateUtil.date());
-					}
-					// TODO: 2023/9/5 音频文件信息入库
-					int rms = DataParser.calculateRMS(fileUploadBaseDir+path);
-					double centerFrequency = DataParser.calcCenterFrequency(fileUploadBaseDir+path);
-					NoiseInfo noise = new NoiseInfo();
-					noise.setEncode(info.getDeviceNo());
-					noise.setPower(info.getPower());
-					noise.setGain(info.getGain());
-					noise.setRms(rms);
-					noise.setCenterFrequency(centerFrequency);
-					noise.setFilepath(info.getFilePath());
-					noise.setTs(DateUtil.format(DateUtil.offset(info.getGatheTime(), DateField.HOUR, -8),
-							DatePattern.NORM_DATETIME_PATTERN));
-					//log.info(JsonUtils.toJsonString(noise));
-					requireSaved(service.save(noise));
-                    run(noise.getEncode(), "采集应答", () ->
-                            ctx.writeAndFlush(Unpooled.copiedBuffer("OK\r\n".getBytes(StandardCharsets.US_ASCII))));
-                    run(noise.getEncode(), "接收时间更新", () ->
-                            DeviceOfflineCheckTask.deviceLastReceiveTimeMap.put(noise.getEncode(), new java.util.Date()));
-                    run(noise.getEncode(), "在线状态同步", () -> deviceOfflineCheckTask.markDeviceOnline(noise.getEncode()));
-                    run(noise.getEncode(), "漏失报警预警", () -> noiseLeakAlarmService.evaluate(noise));
-				}
-
-			}
-
-		} catch (Exception e) {
-			log.error(e.getMessage(), e);
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) {
+        try {
+            ByteBuf input = (ByteBuf) msg;
+            if (!input.isReadable()) return;
+            ChannelInfo info = channelGroup.computeIfAbsent(ctx.channel().id(), k -> new ChannelInfo("puqi", ctx.channel()));
+            if (input.readableBytes() < 4 || input.getInt(input.readerIndex()) != 0x52494646) {
+                String line = input.toString(StandardCharsets.US_ASCII).trim();
+                if (line.startsWith("$$$")) {
+                    String device = line.substring(3).trim();
+                    if (!device.matches("[A-Za-z0-9_-]{1,80}")) throw new IllegalArgumentException("音频设备编号无效");
+                    if (info.getDeviceNo() != null) deviceGroup.remove(info.getDeviceNo(), info);
+                    info.setDeviceNo(device);
+                    deviceGroup.put(device, info);
+                } else if (line.startsWith("VBAT:")) {
+                    info.setPower(parseNumber(line.substring(5).replace("V", "")));
+                } else if (line.startsWith("Gain:")) {
+                    info.setGain(parseNumber(line.substring(5)));
+                }
+                return;
+            }
+            if (info.getDeviceNo() == null) throw new IllegalStateException("音频上传前必须申报设备身份");
+            byte[] bytes = new byte[input.readableBytes()];
+            input.readBytes(bytes);
+            boolean fresh = uploads.receive(info.getDeviceNo(), bytes, retransmissionWindowSeconds,
+                    () -> saveAudio(ctx, info, bytes));
+            if (!fresh) {
+                // 已落库的同一份音频仅补发确认和更新在线状态，不重复漏失计数。
+                acknowledge(ctx, info.getDeviceNo());
+            }
+        } catch (Exception e) {
+            log.error("音频采集处理失败，继续接收后续采样", e);
         } finally {
             ReferenceCountUtil.release(msg);
         }
     }
 
-	private String getPathName(String deviceNo, String tempPath) {
-		String ext = ".wav";
-		tempPath = "/devices/" + deviceNo + "/" + DateUtil.format(DateUtil.date(), DatePattern.PURE_DATE_FORMAT);
-		File desc = new File(fileUploadBaseDir, tempPath);
-		if (!desc.exists()) {
-			if (!desc.getParentFile().exists()) {
-				desc.getParentFile().mkdirs();
-			}
-		}
-		String fileName = deviceNo + "_" + DateUtil.format(DateUtil.date(), DatePattern.PURE_DATETIME_FORMAT) + ext;
-		return tempPath + "/" + fileName;
-	}
-
-	private void saveHexStringToFile(String filePath, String hexString) {
-		if (StrUtil.isEmpty(hexString)) {
-			return;
-		}
-		log.info("音频字节:{}",hexString);
-		byte[] bytes = CrtHelper.strToToHexByte1(hexString);
-		try {
-			FileUtils.writeByteArrayToFile(new File(fileUploadBaseDir + filePath), bytes, true);
-		} catch (IOException e) {
-		}
-	}
-
-	@Override
-	public void channelActive(ChannelHandlerContext ctx) throws Exception {
-		super.channelActive(ctx);
-		log.info("==客户端：{}，连接成功", ctx.channel().remoteAddress());
-	}
-
-	@Override
-	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
-		ctx.flush();
-	}
-
-	@Override
-	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-		log.error("TCP异常：{}", cause.getMessage());
-		super.exceptionCaught(ctx, cause);
-	}
+    private void saveAudio(ChannelHandlerContext ctx, ChannelInfo info, byte[] bytes) throws Exception {
+        Path file = null;
+        boolean saved = false;
+        try {
+            LocalDateTime now = LocalDateTime.now();
+            String relative = "/devices/" + info.getDeviceNo() + "/" + now.toLocalDate().toString().replace("-", "")
+                    + "/" + UUID.randomUUID() + ".wav";
+            Path base = Paths.get(fileUploadBaseDir).toAbsolutePath().normalize();
+            file = base.resolve(relative.substring(1)).normalize();
+            if (!file.startsWith(base)) throw new IllegalArgumentException("音频路径无效");
+            Files.createDirectories(file.getParent());
+            // 完整写入后再计算，避免异步追加造成不存在/不完整的文件。
+            Files.write(file, bytes, StandardOpenOption.CREATE_NEW);
+            NoiseInfo noise = new NoiseInfo();
+            noise.setEncode(info.getDeviceNo());
+            noise.setPower(info.getPower());
+            noise.setGain(info.getGain());
+            noise.setRms(DataParser.calculateRMS(file.toString()));
+            noise.setCenterFrequency(DataParser.calcCenterFrequency(file.toString()));
+            noise.setFilepath(relative);
+            noise.setTs(now.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
+            requireSaved(service.save(noise));
+            saved = true;
+            info.setPath(relative);
+            info.setFilePath(relative);
+            acknowledge(ctx, noise.getEncode());
+            run(noise.getEncode(), "漏失报警预警", () -> noiseLeakAlarmService.evaluate(noise));
+        } finally {
+            if (!saved && file != null) {
+                try { Files.deleteIfExists(file); } catch (Exception e) { log.warn("清理未入库音频失败", e); }
+            }
+        }
+    }
 
-	@Override
-	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-		log.info("==客户端：{}，掉线了", ctx.channel().remoteAddress());
-		ChannelInfo channelInfo = channelGroup.remove(ctx.channel().id());
-		if (channelInfo != null && StrUtil.isNotEmpty(channelInfo.getDeviceNo())) {
-			deviceGroup.remove(channelInfo.getDeviceNo());
-		}
-		super.channelInactive(ctx);
-	}
+    private void acknowledge(ChannelHandlerContext ctx, String device) {
+        run(device, "采集应答", () -> ctx.writeAndFlush(Unpooled.copiedBuffer("OK\r\n", StandardCharsets.US_ASCII)));
+        run(device, "接收时间更新", () -> DeviceOfflineCheckTask.deviceLastReceiveTimeMap.put(device, new Date()));
+        run(device, "在线状态同步", () -> deviceOfflineCheckTask.markDeviceOnline(device));
+    }
 
-	@Override
-	public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
-		log.info("==客户端：{}，掉线了，ChannelUnregistered", ctx.channel().remoteAddress());
-		super.channelUnregistered(ctx);
-	}
+    private double parseNumber(String value) {
+        double result = Double.parseDouble(value.trim());
+        if (!Double.isFinite(result)) throw new IllegalArgumentException("音频参数不是有效数值");
+        return result;
+    }
 
-	@Override
-	public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
-		log.info("==服务端：{}，下线了", ctx.toString());
-		super.handlerRemoved(ctx);
-	}
-}
+    @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        ChannelInfo info = channelGroup.remove(ctx.channel().id());
+        if (info != null && info.getDeviceNo() != null) deviceGroup.remove(info.getDeviceNo(), info);
+        super.channelInactive(ctx);
+    }
+    @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+        log.error("音频连接异常", cause);
+        ctx.close();
+    }
+}

audio-service/src/main/resources/application-dev.yaml

@@ -15,10 +15,15 @@ spring:
         server-addr: 192.168.110.30:8848
       config:
         server-addr: 192.168.110.30:8848
+device:
+  offline:
+    timeout-minutes: 1560 # 每24小时上报，超过26小时未收到有效数据才判离线
+    check-interval-ms: 300000
 wlds:
+  retransmission-window-seconds: 120 # 每24小时上传正常/异常录音；同内容连续重传刷新2分钟静默窗口
   fileUploadBaseDir: D:\home\wlds\uploads
   leak:
     freq-min: 200.0     # 特征频段下限 Hz
     freq-max: 2000.0    # 特征频段上限 Hz
     window-minutes: 30  # 时间窗口（分钟）
-    alarm-count: 3      # 确认漏失所需累计次数
+    alarm-count: 1      # 单次独立录音满足RMS及频段条件即一级告警、短信

audio-service/src/main/resources/application-druid.yaml

@@ -15,10 +15,15 @@ spring:
         server-addr: 172.16.102.52:8848
       config:
         server-addr: 172.16.102.52:8848
+device:
+  offline:
+    timeout-minutes: 1560 # 每24小时上报，超过26小时未收到有效数据才判离线
+    check-interval-ms: 300000
 wlds:
+  retransmission-window-seconds: 120 # 每24小时上传正常/异常录音；同内容连续重传刷新2分钟静默窗口
   fileUploadBaseDir: \home\wlds\uploads
   leak:
     freq-min: 200.0     # 特征频段下限 Hz
     freq-max: 2000.0    # 特征频段上限 Hz
     window-minutes: 30  # 时间窗口（分钟）
-    alarm-count: 3      # 确认漏失所需累计次数
+    alarm-count: 1      # 单次独立录音满足RMS及频段条件即一级告警、短信

audio-service/src/main/resources/application-prod.yaml

@@ -9,10 +9,15 @@ spring:
   redis:
     host: 47.107.107.47
     port: 6379
+device:
+  offline:
+    timeout-minutes: 1560 # 每24小时上报，超过26小时未收到有效数据才判离线
+    check-interval-ms: 300000
 wlds:
+  retransmission-window-seconds: 120 # 每24小时上传正常/异常录音；同内容连续重传刷新2分钟静默窗口
   fileUploadBaseDir: \home\wlds\uploads
   leak:
     freq-min: 200.0     # 特征频段下限 Hz
     freq-max: 2000.0    # 特征频段上限 Hz
     window-minutes: 30  # 时间窗口（分钟）
-    alarm-count: 3      # 确认漏失所需累计次数
+    alarm-count: 1      # 单次独立录音满足RMS及频段条件即一级告警、短信

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

@@ -4,12 +4,10 @@ import static com.zksy.common.utils.CollectionSideEffects.*;
 
 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.environment.utils.DeviceOfflineCheckTask;
-import com.zksy.utils.DevicePhoneFetchUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
@@ -54,10 +52,6 @@ public class RSServerService {
     @Autowired
     private WarningThresholdService warningThresholdService;
     @Autowired
-    private SmsUtil smsUtil;
-    @Autowired
-    private DevicePhoneFetchUtil devicePhoneFetchUtil;
-    @Autowired
     private AlarmUtil alarmUtil;
     @Autowired
     private RestTemplate restTemplate;
@@ -340,7 +334,7 @@ public class RSServerService {
                     String temWarningType = "悬浮物预警";
                     String temRemark = "环境悬浮物报警";
                     if (alarmUtil.checkAndSaveAlarm(deviceId, temWarningType, temWarningCode,
-                            temMinValue, temMaxValue, BigDecimal.valueOf(nd.getTem()), temRemark)) {
+                            temMinValue, temMaxValue, BigDecimal.valueOf(finalValue), temRemark)) {
                         hasAlarm = true;
                     }
                 }
@@ -399,10 +393,7 @@ public class RSServerService {
                 evaluationCompleted = false;
                 log.error("环境报警评估失败: deviceCode={}, nodeId={}", deviceId, nd.getNodeId(), e);
             }
-            run(deviceId, "环境节点短信通知:" + nd.getNodeId(), () -> {
-                List<String> devicePhoneList = devicePhoneFetchUtil.getPhoneListByDeviceId(deviceId);
-                smsUtil.checkDeviceAlarmAndSend(data, nd, devicePhoneList);
-            });
+            // AlarmUtil已按换算后的实际值完成告警与一级短信，不再走旧节点重复报警入口。
         }
         // 同步设备报警状态到 equipment_status 表
         if (hasAlarm || evaluationCompleted) {

environment-service/src/main/java/com/zksy/environment/utils/AlarmUtil.java

@@ -8,7 +8,7 @@ import com.zksy.api.service.AlarmDataService;
 import com.zksy.api.service.EarlyWarningAutoService;
 import com.zksy.api.utils.AlarmLevelUtil;
 import com.zksy.api.utils.SmsUtil;
-import com.zksy.utils.DevicePhoneFetchUtil;
+import com.zksy.api.service.DeviceContactService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -31,7 +31,7 @@ public class AlarmUtil {
     @Autowired
     private SmsUtil smsUtil;
     @Autowired
-    private DevicePhoneFetchUtil devicePhoneFetchUtil;
+    private DeviceContactService devicePhoneFetchUtil;
 
     /**
      * 越限 → 报警；接近阈值未越限 → 自动预警
@@ -106,7 +106,7 @@ public class AlarmUtil {
     private void sendAlarmSms(String deviceCode, String warningType, BigDecimal actualValue,
                               BigDecimal minValue, BigDecimal maxValue) {
         try {
-            List<String> phoneList = devicePhoneFetchUtil.getPhoneListByDeviceId(deviceCode);
+            List<String> phoneList = devicePhoneFetchUtil.getPhoneListByDeviceCode(deviceCode);
             if (phoneList == null || phoneList.isEmpty()) {
                 log.warn("报警短信发送失败：未找到设备{}的联系人手机号", deviceCode);
                 return;

firefighting-pressure-service/src/main/java/com/zksy/pressure/config/NettyServerThread.java

@@ -1,6 +1,7 @@
 package com.zksy.pressure.config;
 
 import com.zksy.pressure.utils.MessageHandler;
+import com.zksy.common.netty.DeviceFrameDecoder;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelInitializer;
@@ -10,6 +11,8 @@ import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.timeout.ReadTimeoutHandler;
+import io.netty.util.concurrent.DefaultEventExecutorGroup;
+import io.netty.util.concurrent.EventExecutorGroup;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -22,6 +25,8 @@ import org.springframework.stereotype.Component;
 public class NettyServerThread {
 	@Value("${netty.port:8311}")
 	private int port;
+    @Value("${collection.worker-threads:16}")
+    private int businessThreads = 16;
 	private static Logger logger = LoggerFactory.getLogger(NettyServerThread.class);
 	private final MessageHandler messageHandler;
 	@Autowired
@@ -32,6 +37,7 @@ public class NettyServerThread {
 		System.out.println("Netty服务启动端口号" + port);
 		EventLoopGroup acceptor = new NioEventLoopGroup();
 		EventLoopGroup worker = new NioEventLoopGroup();
+        EventExecutorGroup business = new DefaultEventExecutorGroup(businessThreads, null, 1024, io.netty.util.concurrent.RejectedExecutionHandlers.reject());
 		NettyServer.acceptor = acceptor;
 		NettyServer.worker = worker;
 		ServerBootstrap bootstrap = new ServerBootstrap();
@@ -52,8 +58,7 @@ public class NettyServerThread {
 			public void initChannel(SocketChannel ch) throws Exception {
 				logger.info("来自" + ch.remoteAddress() + "的新连接接入");
 				// 注册handler
-				ch.pipeline().addLast(new ReadTimeoutHandler(3600));// 超时时间,1小时内没有从通道（Channel）读取到任何数据
-				ch.pipeline().addLast(messageHandler);
+                configurePipeline(ch.pipeline(), business);
 			}
 		});
 
@@ -66,6 +71,7 @@ public class NettyServerThread {
 		} catch (InterruptedException e) {
 			e.printStackTrace();
 		} finally {
+            business.shutdownGracefully();
 			if (acceptor != null) {
 				acceptor.shutdownGracefully();
 			}
@@ -74,4 +80,10 @@ public class NettyServerThread {
 			}
 		}
 	}
-}
+
+    public void configurePipeline(io.netty.channel.ChannelPipeline pipeline, EventExecutorGroup business) {
+				pipeline.addLast(new ReadTimeoutHandler(3600));// 超时时间,1小时内没有从通道（Channel）读取到任何数据
+				pipeline.addLast(new DeviceFrameDecoder(DeviceFrameDecoder.Protocol.FIRE_PRESSURE));
+				pipeline.addLast(business, messageHandler);
+    }
+}

firefighting-pressure-service/src/main/java/com/zksy/pressure/utils/MessageHandler.java

@@ -12,7 +12,7 @@ import com.zksy.api.utils.SmsUtil;
 import com.zksy.common.exception.InvalidMessageException;
 import com.zksy.pressure.domain.FirefightingPressure;
 import com.zksy.pressure.service.FirefightingPressureService;
-import com.zksy.utils.DevicePhoneFetchUtil;
+import com.zksy.api.service.DeviceContactService;
 import io.netty.buffer.ByteBuf;
 import io.netty.util.ReferenceCountUtil;
 import io.netty.buffer.Unpooled;
@@ -49,7 +49,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 	@Autowired
 	private SmsUtil smsUtil;
 	@Autowired
-    private DevicePhoneFetchUtil devicePhoneFetchUtil;
+    private DeviceContactService devicePhoneFetchUtil;
     @Autowired
     private RestTemplate restTemplate;
 
@@ -329,7 +329,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 		params.put("alarmType", alarmType);
 		// 经纬度处理（为执行信息处理）
 		params.put("location", String.format("经纬度：%.6f,%.8f", lng, lat));
-		List<String> phoneListByDeviceId = devicePhoneFetchUtil.getPhoneListByDeviceId(deviceNo);
+		List<String> phoneListByDeviceId = devicePhoneFetchUtil.getPhoneListByDeviceCode(deviceNo);
 			Map<String, Boolean> sendResults = smsUtil.sendBatchSms(phoneListByDeviceId, params.toJSONString());
 			log.info("发送信息完成");
 		}

flammable-gas-service/src/main/java/com/zksy/gas/config/NettyServerThread.java

@@ -1,6 +1,7 @@
 package com.zksy.gas.config;
 
 import com.zksy.gas.utils.MessageHandler;
+import com.zksy.common.netty.DeviceFrameDecoder;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelInitializer;
@@ -10,6 +11,8 @@ import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.timeout.ReadTimeoutHandler;
+import io.netty.util.concurrent.DefaultEventExecutorGroup;
+import io.netty.util.concurrent.EventExecutorGroup;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -22,6 +25,8 @@ import org.springframework.stereotype.Component;
 public class NettyServerThread {
 	@Value("${netty.port:8312}")
 	private int port;
+    @Value("${collection.worker-threads:16}")
+    private int businessThreads = 16;
 	private static Logger logger = LoggerFactory.getLogger(NettyServerThread.class);
 	private final MessageHandler messageHandler;
 	@Autowired
@@ -32,6 +37,7 @@ public class NettyServerThread {
 		System.out.println("Netty服务启动端口号" + port);
 		EventLoopGroup acceptor = new NioEventLoopGroup();
 		EventLoopGroup worker = new NioEventLoopGroup();
+        EventExecutorGroup business = new DefaultEventExecutorGroup(businessThreads, null, 1024, io.netty.util.concurrent.RejectedExecutionHandlers.reject());
 		NettyServer.acceptor = acceptor;
 		NettyServer.worker = worker;
 		ServerBootstrap bootstrap = new ServerBootstrap();
@@ -52,8 +58,7 @@ public class NettyServerThread {
 			public void initChannel(SocketChannel ch) throws Exception {
 				logger.info("来自" + ch.remoteAddress() + "的新连接接入");
 				// 注册handler
-				ch.pipeline().addLast(new ReadTimeoutHandler(3600));// 超时时间,1小时内没有从通道（Channel）读取到任何数据
-				ch.pipeline().addLast(messageHandler);
+                configurePipeline(ch.pipeline(), business);
 			}
 		});
 
@@ -66,6 +71,7 @@ public class NettyServerThread {
 		} catch (InterruptedException e) {
 			e.printStackTrace();
 		} finally {
+            business.shutdownGracefully();
 			if (acceptor != null) {
 				acceptor.shutdownGracefully();
 			}
@@ -74,4 +80,10 @@ public class NettyServerThread {
 			}
 		}
 	}
-}
+
+    public void configurePipeline(io.netty.channel.ChannelPipeline pipeline, EventExecutorGroup business) {
+				pipeline.addLast(new ReadTimeoutHandler(3600));// 超时时间,1小时内没有从通道（Channel）读取到任何数据
+				pipeline.addLast(new DeviceFrameDecoder(DeviceFrameDecoder.Protocol.GAS));
+				pipeline.addLast(business, messageHandler);
+    }
+}

flammable-gas-service/src/main/java/com/zksy/gas/utils/MessageHandler.java

@@ -14,7 +14,7 @@ import com.zksy.api.utils.SmsUtil;
 import com.zksy.common.exception.InvalidMessageException;
 import com.zksy.gas.domain.GasMonitorData;
 import com.zksy.gas.service.GasMonitorDataService;
-import com.zksy.utils.DevicePhoneFetchUtil;
+import com.zksy.api.service.DeviceContactService;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandler;
@@ -54,7 +54,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 	@Autowired
 	private SmsUtil smsUtil;
 	@Autowired
-    private DevicePhoneFetchUtil devicePhoneFetchUtil;
+    private DeviceContactService devicePhoneFetchUtil;
     @Autowired
     private AlarmUtil alarmUtil;
     @Autowired
@@ -298,7 +298,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 		params.put("alarmType", alarmType);
 		// 经纬度处理（位置）
 		params.put("location", String.format("经纬度：%.6f,%.8f", lng, lat));
-		List<String> phoneListByDeviceId = devicePhoneFetchUtil.getPhoneListByDeviceId(deviceNo);
+		List<String> phoneListByDeviceId = devicePhoneFetchUtil.getPhoneListByDeviceCode(deviceNo);
 		Map<String, Boolean> sendResults = smsUtil.sendBatchSms(phoneListByDeviceId, params.toJSONString());
 		log.info("发送信息完成");
     }

manhole-service/src/main/java/com/zksy/manhole/config/NettyServerThread.java

@@ -1,6 +1,7 @@
 package com.zksy.manhole.config;
 
 import com.zksy.manhole.utils.MessageHandler;
+import com.zksy.common.netty.DeviceFrameDecoder;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelInitializer;
@@ -11,6 +12,8 @@ import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.handler.timeout.ReadTimeoutHandler;
+import io.netty.util.concurrent.DefaultEventExecutorGroup;
+import io.netty.util.concurrent.EventExecutorGroup;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -25,6 +28,8 @@ import java.util.concurrent.TimeUnit;
 public class NettyServerThread {
 	@Value("${netty.port:20011}")
 	private int port;
+    @Value("${collection.worker-threads:16}")
+    private int businessThreads = 16;
 	private static Logger logger = LoggerFactory.getLogger(NettyServerThread.class);
 	private final MessageHandler messageHandler;
 	@Autowired
@@ -35,6 +40,7 @@ public class NettyServerThread {
 		System.out.println("Netty服务启动端口号" + port);
 		EventLoopGroup acceptor = new NioEventLoopGroup();
 		EventLoopGroup worker = new NioEventLoopGroup();
+        EventExecutorGroup business = new DefaultEventExecutorGroup(businessThreads, null, 1024, io.netty.util.concurrent.RejectedExecutionHandlers.reject());
 		NettyServer.acceptor = acceptor;
 		NettyServer.worker = worker;
 		ServerBootstrap bootstrap = new ServerBootstrap();
@@ -55,8 +61,7 @@ public class NettyServerThread {
 			public void initChannel(SocketChannel ch) throws Exception {
 				logger.info("来自" + ch.remoteAddress() + "的新连接接入");
 				// 注册handler
-				ch.pipeline().addLast(new ReadTimeoutHandler(10));// 超时时间,10s内没有从通道（Channel）读取到任何数据
-				ch.pipeline().addLast(messageHandler);
+                configurePipeline(ch.pipeline(), business);
 			}
 		});
 
@@ -69,6 +74,7 @@ public class NettyServerThread {
 		} catch (InterruptedException e) {
 			e.printStackTrace();
 		} finally {
+            business.shutdownGracefully();
 			if (acceptor != null) {
 				acceptor.shutdownGracefully();
 			}
@@ -77,4 +83,10 @@ public class NettyServerThread {
 			}
 		}
 	}
-}
+
+    public void configurePipeline(io.netty.channel.ChannelPipeline pipeline, EventExecutorGroup business) {
+				pipeline.addLast(new ReadTimeoutHandler(10));// 超时时间,10s内没有从通道（Channel）读取到任何数据
+				pipeline.addLast(new DeviceFrameDecoder(DeviceFrameDecoder.Protocol.MANHOLE));
+				pipeline.addLast(business, messageHandler);
+    }
+}

manhole-service/src/main/java/com/zksy/manhole/utils/AlarmUtil.java

@@ -7,7 +7,7 @@ import com.zksy.api.utils.AlarmLevelUtil;
 import com.zksy.api.utils.SmsUtil;
 import com.zksy.api.domain.AlarmData;
 import com.zksy.api.service.AlarmDataService;
-import com.zksy.utils.DevicePhoneFetchUtil;
+import com.zksy.api.service.DeviceContactService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -34,7 +34,7 @@ public class AlarmUtil {
     @Autowired
     private SmsUtil smsUtil;
     @Autowired
-    private DevicePhoneFetchUtil devicePhoneFetchUtil;
+    private DeviceContactService devicePhoneFetchUtil;
 
     @Autowired
     private AlarmDataService alarmDataService;
@@ -144,7 +144,7 @@ public class AlarmUtil {
     private void sendAlarmSms(String deviceCode, String warningType, BigDecimal actualValue,
                               BigDecimal minValue, BigDecimal maxValue) {
         try {
-            List<String> phoneList = devicePhoneFetchUtil.getPhoneListByDeviceId(deviceCode);
+            List<String> phoneList = devicePhoneFetchUtil.getPhoneListByDeviceCode(deviceCode);
             if (phoneList == null || phoneList.isEmpty()) {
                 log.warn("报警短信发送失败：未找到设备{}的联系人手机号", deviceCode);
                 return;

manhole-service/src/main/java/com/zksy/manhole/utils/MessageHandler.java

@@ -1,5 +1,7 @@
 package com.zksy.manhole.utils;
 
+import static com.zksy.common.utils.CollectionSideEffects.*;
+
 import com.zksy.common.exception.InvalidMessageException;
 import com.zksy.manhole.domain.ManholeData;
 import com.zksy.manhole.service.ManholeDataService;
@@ -9,6 +11,7 @@ import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.handler.timeout.ReadTimeoutException;
+import io.netty.util.ReferenceCountUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -43,6 +46,9 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 			logger.info("接收到 {} 字节的数据，来自: {}", msgByteBuf.readableBytes(), ctx.channel().remoteAddress());
 			byte[] msgBytes = new byte[msgByteBuf.readableBytes()];
 			msgByteBuf.readBytes(msgBytes);
+			if (msgBytes.length != 72) {
+				throw new InvalidMessageException("井盖报文长度应为72字节");
+			}
 			String result = printHexBinary(msgBytes);
 			String msgString = result.replaceAll("\\s", "");
 			String CRCString = msgString.substring(msgString.length() - 4);
@@ -53,15 +59,23 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 				throw new InvalidMessageException("数据校验不成功");
 			} else {
 				ManholeData resultData = DataParser.parseMessage(msgString);
-				// 更新设备最后一次接收数据的时间，并标记设备在线
 				String imeiCardNumber = resultData.getImeiCardNumber();
-				DeviceOfflineCheckTask.deviceLastReceiveTimeMap.put(imeiCardNumber, new Date());
-				deviceOfflineCheckTask.markDeviceOnline(imeiCardNumber);
-				messageParseResultService.saveManholeData(resultData);
+				if (imeiCardNumber == null || imeiCardNumber.trim().isEmpty()) {
+					throw new InvalidMessageException("井盖报文缺少设备编号");
+				}
+				Integer saved = messageParseResultService.saveManholeData(resultData);
+				requireSaved(saved != null && saved > 0);
+				run(imeiCardNumber, "接收时间更新", () -> DeviceOfflineCheckTask.deviceLastReceiveTimeMap.put(imeiCardNumber, new Date()));
+				run(imeiCardNumber, "在线状态同步", () -> deviceOfflineCheckTask.markDeviceOnline(imeiCardNumber));
 			}
 		} catch (InvalidMessageException e) {
 			logger.error("数据入库失败: {}", e.getMessage());
 			ctx.writeAndFlush(Unpooled.copiedBuffer("数据入库失败".getBytes()));
+		} catch (Exception e) {
+			logger.error("井盖数据解析或入库失败，继续接收后续报文", e);
+			ctx.writeAndFlush(Unpooled.copiedBuffer("数据处理异常".getBytes(java.nio.charset.StandardCharsets.UTF_8)));
+		} finally {
+			ReferenceCountUtil.release(msg);
 		}
 	}
 
@@ -96,4 +110,4 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 		logger.info("来自" + ctx.channel().remoteAddress() + "的连接主动断开");
 		ctx.fireChannelUnregistered();
 	}
-}
+}

pom.xml

@@ -8,6 +8,12 @@
     <packaging>pom</packaging>
     <version>1.0.0</version>
 
+    <profiles>
+        <profile>
+            <id>collection-tests</id>
+            <modules><module>collection-integration-tests</module></modules>
+        </profile>
+    </profiles>
     <modules>
         <module>pipe-network-service</module>
         <module>environment-service</module>
@@ -166,4 +172,4 @@
         </pluginManagement>
     </build>
 
-</project>
+</project>

radar-service/src/main/java/com/zksy/radar/config/NettyServerThread.java

@@ -1,6 +1,7 @@
 package com.zksy.radar.config;
 
 import com.zksy.radar.utils.MessageHandler;
+import com.zksy.common.netty.DeviceFrameDecoder;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelInitializer;
@@ -10,6 +11,8 @@ import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.timeout.ReadTimeoutHandler;
+import io.netty.util.concurrent.DefaultEventExecutorGroup;
+import io.netty.util.concurrent.EventExecutorGroup;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -22,6 +25,8 @@ import org.springframework.stereotype.Component;
 public class NettyServerThread {
 	@Value("${netty.port:8313}")
 	private int port;
+    @Value("${collection.worker-threads:16}")
+    private int businessThreads = 16;
 	private static Logger logger = LoggerFactory.getLogger(NettyServerThread.class);
 	private final MessageHandler messageHandler;
 	@Autowired
@@ -32,6 +37,7 @@ public class NettyServerThread {
 		System.out.println("Netty服务启动端口号" + port);
 		EventLoopGroup acceptor = new NioEventLoopGroup();
 		EventLoopGroup worker = new NioEventLoopGroup();
+        EventExecutorGroup business = new DefaultEventExecutorGroup(businessThreads, null, 1024, io.netty.util.concurrent.RejectedExecutionHandlers.reject());
 		NettyServer.acceptor = acceptor;
 		NettyServer.worker = worker;
 		ServerBootstrap bootstrap = new ServerBootstrap();
@@ -52,8 +58,7 @@ public class NettyServerThread {
 			public void initChannel(SocketChannel ch) throws Exception {
 				logger.info("来自" + ch.remoteAddress() + "的新连接接入");
 				// 注册handler
-				ch.pipeline().addLast(new ReadTimeoutHandler(3600));// 超时时间,1小时内没有从通道（Channel）读取到任何数据
-				ch.pipeline().addLast(messageHandler);
+                configurePipeline(ch.pipeline(), business);
 			}
 		});
 
@@ -66,6 +71,7 @@ public class NettyServerThread {
 		} catch (InterruptedException e) {
 			e.printStackTrace();
 		} finally {
+            business.shutdownGracefully();
 			if (acceptor != null) {
 				acceptor.shutdownGracefully();
 			}
@@ -74,4 +80,10 @@ public class NettyServerThread {
 			}
 		}
 	}
-}
+
+    public void configurePipeline(io.netty.channel.ChannelPipeline pipeline, EventExecutorGroup business) {
+				pipeline.addLast(new ReadTimeoutHandler(3600));// 超时时间,1小时内没有从通道（Channel）读取到任何数据
+				pipeline.addLast(new DeviceFrameDecoder(DeviceFrameDecoder.Protocol.PINGSHENG));
+				pipeline.addLast(business, messageHandler);
+    }
+}

radar-service/src/main/java/com/zksy/radar/utils/DataParser.java

@@ -58,19 +58,17 @@ public class DataParser {
                 throw new InvalidMessageException("CRC数据段索引错误");
             }
 
-            // 设备编码开始（内层协议起点）：
-            // 外层结构 3+2+1+1+1+n+1+m => 源地址/目的地址结束后紧接设备编码(0x01)
-            // 简化方式：找到第一个 0x01 2C 序列即可（设备编码 + 功能码）
-            int deviceCodeIndex = -1;
-            for (int i = 0; i < msgBytes.length - 1; i++) {
-                if (msgBytes[i] == 0x01 && msgBytes[i + 1] == 0x2C) {
-                    deviceCodeIndex = i;
-                    break;
-                }
+            // 地址内容也可能包含01 2C，必须根据地址长度定位内层协议。
+            int srcBytes = calculateAddrLength(msgBytes[7] & 0xFF);
+            int destFlagIndex = 8 + srcBytes;
+            if (srcBytes == 0 || srcBytes > 10 || destFlagIndex >= crcEndIndex) {
+                throw new InvalidMessageException("源地址长度无效");
             }
-
-            if (deviceCodeIndex == -1) {
-                throw new InvalidMessageException("未找到设备编码起点 (0x01 2C)");
+            int destBytes = calculateAddrLength(msgBytes[destFlagIndex] & 0xFF);
+            int deviceCodeIndex = destFlagIndex + 1 + destBytes;
+            if (destBytes > 10 || deviceCodeIndex + 2 >= crcEndIndex
+                    || msgBytes[deviceCodeIndex] != 0x01 || msgBytes[deviceCodeIndex + 1] != 0x2C) {
+                throw new InvalidMessageException("内层协议地址或功能码无效");
             }
 
             byte[] crcData = Arrays.copyOfRange(msgBytes, deviceCodeIndex, crcEndIndex);
@@ -127,7 +125,8 @@ public class DataParser {
         // 6. 帧类型 = 0x31
         if (frameType == 0x31) {
             // 设备ID
-            data.setDeviceCode(String.format("%02X", msgBytes[index++]));
+            index++; // 内层仪表地址(01)不是设备台账编号
+            data.setDeviceCode(data.getSourceAddr());
             // 功能码
             data.setFunctionCode(String.format("%02X", msgBytes[index++]));
 

radar-service/src/main/java/com/zksy/radar/utils/MessageHandler.java

@@ -12,7 +12,7 @@ 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.utils.DevicePhoneFetchUtil;
+import com.zksy.api.service.DeviceContactService;
 import io.netty.buffer.ByteBuf;
 import io.netty.util.ReferenceCountUtil;
 import io.netty.buffer.Unpooled;
@@ -47,7 +47,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 	private final AlarmDataService alarmDataService;
 	private final EarlyWarningAutoService earlyWarningAutoService;
 	private final SmsUtil smsUtil;
-	private final DevicePhoneFetchUtil devicePhoneFetchUtil;
+	private final DeviceContactService devicePhoneFetchUtil;
     private final RestTemplate restTemplate;
 
     @Autowired
@@ -59,7 +59,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
     @Autowired
     public MessageHandler(RadarDataService RadarDataService, WarningThresholdService warningThresholdService,
                            AlarmDataService alarmDataService, EarlyWarningAutoService earlyWarningAutoService,
-                           SmsUtil smsUtil, DevicePhoneFetchUtil devicePhoneFetchUtil,
+                           SmsUtil smsUtil, DeviceContactService devicePhoneFetchUtil,
                            RestTemplate restTemplate) {
         this.service = RadarDataService;
         this.warningThresholdService = warningThresholdService;
@@ -112,7 +112,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 				}
 				// 入库
 				requireSaved(service.save(resultData));
-                String deviceCode = resultData.getSystemIdentifier();
+                String deviceCode = resultData.getDeviceCode();
                 run(deviceCode, "接收时间更新", () ->
                         DeviceOfflineCheckTask.deviceLastReceiveTimeMap.put(deviceCode, new java.util.Date()));
                 run(deviceCode, "采集应答", () -> {
@@ -211,7 +211,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 		 * 处理雷达设备告警
 		 */
     private void handleRadarAlarm(RadarData resultData) {
-        String deviceCode = resultData.getDeviceCode() != null ? resultData.getDeviceCode() : resultData.getSystemIdentifier();
+        String deviceCode = resultData.getDeviceCode();
         Integer status = alarmStatus(
                 handleAlarm(deviceCode, "WARN_INSTANT_FLOW", "瞬时流量", resultData.getMeter1InstantFlow()),
                 handleAlarm(deviceCode, "WARN_FLOW_SPEED", "流速", resultData.getFlowSpeed()));
@@ -232,7 +232,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
             BigDecimal actual = new BigDecimal(valueStr);
             WarningThreshold threshold = warningThresholdService.getWarningThresholdByDeviceAndCode(deviceCode, warningCode);
             if (threshold == null || (threshold.getMinValue() == null && threshold.getMaxValue() == null)) {
-                return false;
+                throw new IllegalStateException("阈值缺失，不能判断设备已恢复正常");
             }
             String type = defaultType;
             BigDecimal min = threshold.getMinValue() == null ? null : BigDecimal.valueOf(threshold.getMinValue());
@@ -287,7 +287,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 	private void sendAlarmSms(String deviceCode, String warningType, BigDecimal actualValue,
 	                          BigDecimal minValue, BigDecimal maxValue) {
 		try {
-			List<String> phoneList = devicePhoneFetchUtil.getPhoneListByDeviceId(deviceCode);
+			List<String> phoneList = devicePhoneFetchUtil.getPhoneListByDeviceCode(deviceCode);
 			if (phoneList == null || phoneList.isEmpty()) {
 				logger.warn("报警短信发送失败：未找到设备{}的维护人员手机号", deviceCode);
 				return;

telemetry-service/src/main/java/com/zksy/telemetry/config/NettyServerThread.java

@@ -1,6 +1,7 @@
 package com.zksy.telemetry.config;
 
 import com.zksy.telemetry.utils.MessageHandler;
+import com.zksy.common.netty.DeviceFrameDecoder;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelInitializer;
@@ -10,6 +11,8 @@ import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.timeout.ReadTimeoutHandler;
+import io.netty.util.concurrent.DefaultEventExecutorGroup;
+import io.netty.util.concurrent.EventExecutorGroup;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -22,6 +25,8 @@ import org.springframework.stereotype.Component;
 public class NettyServerThread {
 	@Value("${netty.port:8313}")
 	private int port;
+    @Value("${collection.worker-threads:16}")
+    private int businessThreads = 16;
 	private static Logger logger = LoggerFactory.getLogger(NettyServerThread.class);
 	private final MessageHandler messageHandler;
 	@Autowired
@@ -32,6 +37,7 @@ public class NettyServerThread {
 		System.out.println("Netty服务启动端口号" + port);
 		EventLoopGroup acceptor = new NioEventLoopGroup();
 		EventLoopGroup worker = new NioEventLoopGroup();
+        EventExecutorGroup business = new DefaultEventExecutorGroup(businessThreads, null, 1024, io.netty.util.concurrent.RejectedExecutionHandlers.reject());
 		NettyServer.acceptor = acceptor;
 		NettyServer.worker = worker;
 		ServerBootstrap bootstrap = new ServerBootstrap();
@@ -52,8 +58,7 @@ public class NettyServerThread {
 			public void initChannel(SocketChannel ch) throws Exception {
 				logger.info("来自" + ch.remoteAddress() + "的新连接接入");
 				// 注册handler
-				ch.pipeline().addLast(new ReadTimeoutHandler(3600));// 超时时间,1小时内没有从通道（Channel）读取到任何数据
-				ch.pipeline().addLast(messageHandler);
+                configurePipeline(ch.pipeline(), business);
 			}
 		});
 
@@ -66,6 +71,7 @@ public class NettyServerThread {
 		} catch (InterruptedException e) {
 			e.printStackTrace();
 		} finally {
+            business.shutdownGracefully();
 			if (acceptor != null) {
 				acceptor.shutdownGracefully();
 			}
@@ -74,4 +80,10 @@ public class NettyServerThread {
 			}
 		}
 	}
-}
+
+    public void configurePipeline(io.netty.channel.ChannelPipeline pipeline, EventExecutorGroup business) {
+				pipeline.addLast(new ReadTimeoutHandler(3600));// 超时时间,1小时内没有从通道（Channel）读取到任何数据
+				pipeline.addLast(new DeviceFrameDecoder(DeviceFrameDecoder.Protocol.PINGSHENG));
+				pipeline.addLast(business, messageHandler);
+    }
+}

telemetry-service/src/main/java/com/zksy/telemetry/utils/DataParser.java

@@ -58,19 +58,17 @@ public class DataParser {
                 throw new InvalidMessageException("CRC数据段索引错误");
             }
 
-            // 设备编码开始（内层协议起点）：
-            // 外层结构 3+2+1+1+1+n+1+m => 源地址/目的地址结束后紧接设备编码(0x01)
-            // 简化方式：找到第一个 0x01 2C 序列即可（设备编码 + 功能码）
-            int deviceCodeIndex = -1;
-            for (int i = 0; i < msgBytes.length - 1; i++) {
-                if (msgBytes[i] == 0x01 && msgBytes[i + 1] == 0x2C) {
-                    deviceCodeIndex = i;
-                    break;
-                }
+            // 地址内容也可能包含01 2C，必须根据地址长度定位内层协议。
+            int srcBytes = calculateAddrLength(msgBytes[7] & 0xFF);
+            int destFlagIndex = 8 + srcBytes;
+            if (srcBytes == 0 || srcBytes > 10 || destFlagIndex >= crcEndIndex) {
+                throw new InvalidMessageException("源地址长度无效");
             }
-
-            if (deviceCodeIndex == -1) {
-                throw new InvalidMessageException("未找到设备编码起点 (0x01 2C)");
+            int destBytes = calculateAddrLength(msgBytes[destFlagIndex] & 0xFF);
+            int deviceCodeIndex = destFlagIndex + 1 + destBytes;
+            if (destBytes > 10 || deviceCodeIndex + 2 >= crcEndIndex
+                    || msgBytes[deviceCodeIndex] != 0x01 || msgBytes[deviceCodeIndex + 1] != 0x2C) {
+                throw new InvalidMessageException("内层协议地址或功能码无效");
             }
 
             byte[] crcData = Arrays.copyOfRange(msgBytes, deviceCodeIndex, crcEndIndex);
@@ -127,7 +125,8 @@ public class DataParser {
         // 6. 帧类型 = 0x31
         if (frameType == 0x31) {
             // 设备ID
-            data.setDeviceCode(String.format("%02X", msgBytes[index++]));
+            index++; // 内层仪表地址(01)不是设备台账编号
+            data.setDeviceCode(data.getSourceAddr());
             // 功能码
             data.setFunctionCode(String.format("%02X", msgBytes[index++]));
 

telemetry-service/src/main/java/com/zksy/telemetry/utils/MessageHandler.java

@@ -11,7 +11,7 @@ import com.zksy.api.service.WarningThresholdService;
 import com.zksy.api.utils.AlarmLevelUtil;
 import com.zksy.telemetry.domain.TelemetryData;
 import com.zksy.telemetry.service.TelemetryDataService;
-import com.zksy.utils.DevicePhoneFetchUtil;
+import com.zksy.api.service.DeviceContactService;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandler;
@@ -49,7 +49,9 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 	@Autowired
 	private SmsUtil smsUtil;
 	@Autowired
-	private DevicePhoneFetchUtil devicePhoneFetchUtil;
+	private DeviceContactService devicePhoneFetchUtil;
+	@Autowired
+	private com.zksy.api.utils.CollectionDeviceNotifier deviceNotifier;
 	@Autowired
 	public MessageHandler(TelemetryDataService telemetryDataService) {
 		this.service = telemetryDataService;
@@ -97,7 +99,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 				}
 				// 入库
 				requireSaved(service.save(resultData));
-                String deviceCode = resultData.getSystemIdentifier();
+                String deviceCode = resultData.getDeviceCode();
                 run(deviceCode, "接收时间更新", () ->
                         DeviceOfflineCheckTask.deviceLastReceiveTimeMap.put(deviceCode, new java.util.Date()));
                 run(deviceCode, "采集应答", () -> {
@@ -190,13 +192,17 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
                     earlyWarningAutoService.checkAndGenerate(deviceCode, warningType, warningCode,
                             min, max, actual, null));
         }
+        run(deviceCode, "报警状态同步", () -> deviceNotifier.updateAlarmStatus(deviceCode, over ? 1 : 0));
     }
 
     private String resolveDeviceCode(TelemetryData data) {
+        if (data.getSourceAddr() != null && !data.getSourceAddr().trim().isEmpty()) {
+            return data.getSourceAddr().trim();
+        }
         if (data.getDeviceCode() != null && !data.getDeviceCode().trim().isEmpty()) {
             return data.getDeviceCode().trim();
         }
-        return data.getSystemIdentifier();
+        throw new IllegalArgumentException("供水压力报文缺少设备源地址");
     }
 
     private void sendAlarmSms(String deviceCode, String warningType, BigDecimal actual,
@@ -204,7 +210,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
         if (smsUtil == null || devicePhoneFetchUtil == null) {
             throw new IllegalStateException("一级告警短信依赖未注入");
         }
-        List<String> phones = devicePhoneFetchUtil.getPhoneListByDeviceId(deviceCode);
+        List<String> phones = devicePhoneFetchUtil.getPhoneListByDeviceCode(deviceCode);
         if (phones == null || phones.isEmpty()) {
             log.warn("一级告警短信跳过：未找到设备联系人手机号, deviceCode={}", deviceCode);
             return;

water-level-service/src/main/java/com/zksy/water/config/NettyServerThread.java

@@ -1,6 +1,7 @@
 package com.zksy.water.config;
 
 import com.zksy.water.utils.MessageHandler;
+import com.zksy.common.netty.DeviceFrameDecoder;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelInitializer;
@@ -10,6 +11,8 @@ import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.timeout.ReadTimeoutHandler;
+import io.netty.util.concurrent.DefaultEventExecutorGroup;
+import io.netty.util.concurrent.EventExecutorGroup;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -22,6 +25,8 @@ import org.springframework.stereotype.Component;
 public class NettyServerThread {
 	@Value("${netty.port:8313}")
 	private int port;
+    @Value("${collection.worker-threads:16}")
+    private int businessThreads = 16;
 	private static Logger logger = LoggerFactory.getLogger(NettyServerThread.class);
 	private final MessageHandler messageHandler;
 	@Autowired
@@ -32,6 +37,7 @@ public class NettyServerThread {
 		System.out.println("Netty服务启动端口号" + port);
 		EventLoopGroup acceptor = new NioEventLoopGroup();
 		EventLoopGroup worker = new NioEventLoopGroup();
+        EventExecutorGroup business = new DefaultEventExecutorGroup(businessThreads, null, 1024, io.netty.util.concurrent.RejectedExecutionHandlers.reject());
 		NettyServer.acceptor = acceptor;
 		NettyServer.worker = worker;
 		ServerBootstrap bootstrap = new ServerBootstrap();
@@ -52,8 +58,7 @@ public class NettyServerThread {
 			public void initChannel(SocketChannel ch) throws Exception {
 				logger.info("来自" + ch.remoteAddress() + "的新连接接入");
 				// 注册handler
-				ch.pipeline().addLast(new ReadTimeoutHandler(3600));// 超时时间,1小时内没有从通道（Channel）读取到任何数据
-				ch.pipeline().addLast(messageHandler);
+                configurePipeline(ch.pipeline(), business);
 			}
 		});
 
@@ -66,6 +71,7 @@ public class NettyServerThread {
 		} catch (InterruptedException e) {
 			e.printStackTrace();
 		} finally {
+            business.shutdownGracefully();
 			if (acceptor != null) {
 				acceptor.shutdownGracefully();
 			}
@@ -74,4 +80,10 @@ public class NettyServerThread {
 			}
 		}
 	}
-}
+
+    public void configurePipeline(io.netty.channel.ChannelPipeline pipeline, EventExecutorGroup business) {
+				pipeline.addLast(new ReadTimeoutHandler(3600));// 超时时间,1小时内没有从通道（Channel）读取到任何数据
+				pipeline.addLast(new DeviceFrameDecoder(DeviceFrameDecoder.Protocol.PINGSHENG));
+				pipeline.addLast(business, messageHandler);
+    }
+}

water-level-service/src/main/java/com/zksy/water/utils/DataParser.java

@@ -58,19 +58,17 @@ public class DataParser {
                 throw new InvalidMessageException("CRC数据段索引错误");
             }
 
-            // 设备编码开始（内层协议起点）：
-            // 外层结构 3+2+1+1+1+n+1+m => 源地址/目的地址结束后紧接设备编码(0x01)
-            // 简化方式：找到第一个 0x01 2C 序列即可（设备编码 + 功能码）
-            int deviceCodeIndex = -1;
-            for (int i = 0; i < msgBytes.length - 1; i++) {
-                if (msgBytes[i] == 0x01 && msgBytes[i + 1] == 0x2C) {
-                    deviceCodeIndex = i;
-                    break;
-                }
+            // 地址内容也可能包含01 2C，必须根据地址长度定位内层协议。
+            int srcBytes = calculateAddrLength(msgBytes[7] & 0xFF);
+            int destFlagIndex = 8 + srcBytes;
+            if (srcBytes == 0 || srcBytes > 10 || destFlagIndex >= crcEndIndex) {
+                throw new InvalidMessageException("源地址长度无效");
             }
-
-            if (deviceCodeIndex == -1) {
-                throw new InvalidMessageException("未找到设备编码起点 (0x01 2C)");
+            int destBytes = calculateAddrLength(msgBytes[destFlagIndex] & 0xFF);
+            int deviceCodeIndex = destFlagIndex + 1 + destBytes;
+            if (destBytes > 10 || deviceCodeIndex + 2 >= crcEndIndex
+                    || msgBytes[deviceCodeIndex] != 0x01 || msgBytes[deviceCodeIndex + 1] != 0x2C) {
+                throw new InvalidMessageException("内层协议地址或功能码无效");
             }
 
             byte[] crcData = Arrays.copyOfRange(msgBytes, deviceCodeIndex, crcEndIndex);
@@ -127,7 +125,8 @@ public class DataParser {
         // 6. 帧类型 = 0x31
         if (frameType == 0x31) {
             // 设备ID
-            data.setDeviceCode(String.format("%02X", msgBytes[index++]));
+            index++; // 内层仪表地址(01)不是设备台账编号
+            data.setDeviceCode(data.getSourceAddr());
             // 功能码
             data.setFunctionCode(String.format("%02X", msgBytes[index++]));
 

water-level-service/src/main/java/com/zksy/water/utils/MessageHandler.java

@@ -10,7 +10,7 @@ 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.utils.DevicePhoneFetchUtil;
+import com.zksy.api.service.DeviceContactService;
 import com.zksy.water.domain.WaterMonitorData;
 import com.zksy.water.service.WaterMonitorDataService;
 import io.netty.buffer.ByteBuf;
@@ -47,7 +47,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 	private final AlarmDataService alarmDataService;
 	private final EarlyWarningAutoService earlyWarningAutoService;
 	private final SmsUtil smsUtil;
-	private final DevicePhoneFetchUtil devicePhoneFetchUtil;
+	private final DeviceContactService devicePhoneFetchUtil;
     private final RestTemplate restTemplate;
 
     @Autowired
@@ -59,7 +59,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 	@Autowired
     public MessageHandler(WaterMonitorDataService waterMonitorDataService, WarningThresholdService warningThresholdService,
                            AlarmDataService alarmDataService, EarlyWarningAutoService earlyWarningAutoService,
-                           SmsUtil smsUtil, DevicePhoneFetchUtil devicePhoneFetchUtil,
+                           SmsUtil smsUtil, DeviceContactService devicePhoneFetchUtil,
                            RestTemplate restTemplate) {
         this.service = waterMonitorDataService;
         this.warningThresholdService = warningThresholdService;
@@ -112,7 +112,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 				}
 				// 入库
 				requireSaved(service.save(resultData));
-                String deviceCode = resultData.getSystemIdentifier();
+                String deviceCode = resultData.getDeviceCode();
                 run(deviceCode, "接收时间更新", () ->
                         DeviceOfflineCheckTask.deviceLastReceiveTimeMap.put(deviceCode, new java.util.Date()));
                 run(deviceCode, "采集应答", () -> {
@@ -211,7 +211,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 		 * 处理水位告警
 		 */
     private void handleWaterLevelAlarm(WaterMonitorData resultData) {
-        String deviceCode = resultData.getDeviceCode() != null ? resultData.getDeviceCode() : resultData.getSystemIdentifier();
+        String deviceCode = resultData.getDeviceCode();
         Integer status = alarmStatus(
                 handleAlarm(deviceCode, "WARN-WATER-LEVEL", "水位", resultData.getWaterLevel()));
         if (status != null) {
@@ -231,7 +231,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
             BigDecimal actual = new BigDecimal(valueStr);
             WarningThreshold threshold = warningThresholdService.getWarningThresholdByDeviceAndCode(deviceCode, warningCode);
             if (threshold == null || (threshold.getMinValue() == null && threshold.getMaxValue() == null)) {
-                return false;
+                throw new IllegalStateException("阈值缺失，不能判断设备已恢复正常");
             }
             String type = defaultType;
             BigDecimal min = threshold.getMinValue() == null ? null : BigDecimal.valueOf(threshold.getMinValue());
@@ -286,7 +286,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
 	private void sendAlarmSms(String deviceCode, String warningType, BigDecimal actualValue,
 	                          BigDecimal minValue, BigDecimal maxValue) {
 		try {
-			List<String> phoneList = devicePhoneFetchUtil.getPhoneListByDeviceId(deviceCode);
+			List<String> phoneList = devicePhoneFetchUtil.getPhoneListByDeviceCode(deviceCode);
 			if (phoneList == null || phoneList.isEmpty()) {
 				logger.warn("报警短信发送失败：未找到设备{}的维护人员手机号", deviceCode);
 				return;

zk-api-service/src/main/java/com/zksy/api/mapper/DeviceContactApiMapper.java

@@ -0,0 +1,17 @@
+package com.zksy.api.mapper;
+
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
+import java.util.List;
+
+public interface DeviceContactApiMapper {
+    /** 台账编码先关联设备主键，沿用维护人员和状态负责人的联系人来源。 */
+    @Select("SELECT m.maintain_person_phone FROM app_user.equipment_base b "
+            + "JOIN app_user.equipment_maintain m ON m.equipment_id = b.equipment_id "
+            + "WHERE b.equipment_code = #{deviceCode} "
+            + "UNION SELECT u.phonenumber FROM app_user.equipment_base b "
+            + "JOIN app_user.equipment_status s ON s.equipment_id = b.equipment_id AND s.del_flag = '0' "
+            + "JOIN app_user.sys_user u ON u.user_id = s.user_id AND u.del_flag = '0' "
+            + "WHERE b.equipment_code = #{deviceCode}")
+    List<String> selectPhonesByDeviceCode(@Param("deviceCode") String deviceCode);
+}

zk-api-service/src/main/java/com/zksy/api/service/DeviceContactService.java

@@ -0,0 +1,19 @@
+package com.zksy.api.service;
+
+import com.zksy.api.mapper.DeviceContactApiMapper;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import java.util.*;
+import java.util.stream.Collectors;
+
+@Service
+public class DeviceContactService {
+    @Autowired private DeviceContactApiMapper mapper;
+    public List<String> getPhoneListByDeviceCode(String deviceCode) {
+        if (deviceCode == null || deviceCode.trim().isEmpty()) return Collections.emptyList();
+        List<String> phones = mapper.selectPhonesByDeviceCode(deviceCode);
+        if (phones == null) return Collections.emptyList();
+        return phones.stream().filter(Objects::nonNull).map(String::trim)
+                .filter(p -> p.matches("^1[3-9]\\d{9}$")).distinct().collect(Collectors.toList());
+    }
+}

zk-api-service/src/main/java/com/zksy/api/utils/CollectionDeviceNotifier.java

@@ -0,0 +1,46 @@
+package com.zksy.api.utils;
+
+import com.alibaba.fastjson.JSONObject;
+import com.zksy.api.service.DeviceContactService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.http.*;
+import org.springframework.stereotype.Component;
+import org.springframework.web.client.RestTemplate;
+import java.util.*;
+
+/** 供水压力和漏失服务共用既有设备状态接口与短信入口。调用方分别隔离失败。 */
+@Component
+@org.springframework.context.annotation.Lazy
+public class CollectionDeviceNotifier {
+    @Autowired private RestTemplate restTemplate;
+    @Autowired private SmsUtil smsUtil;
+    @Autowired private DeviceContactService devicePhoneFetchUtil;
+    @Value("${device.status.api.url:http://zk-api-service/equipmentStatus/updateAlarmStatus}")
+    private String statusUrl;
+
+    public void updateAlarmStatus(String deviceCode, int status) {
+        HttpHeaders headers = new HttpHeaders();
+        headers.setContentType(MediaType.APPLICATION_JSON);
+        Map<String,Object> body = new HashMap<>();
+        body.put("deviceCode", deviceCode);
+        body.put("alarmStatus", status);
+        Map<?,?> response = restTemplate.postForObject(statusUrl, new HttpEntity<>(body, headers), Map.class);
+        if (response == null || !Boolean.TRUE.equals(response.get("success"))) {
+            throw new IllegalStateException("设备报警状态同步未成功");
+        }
+    }
+
+    public void sendLevelOneSms(String deviceCode, String type, String detail) {
+        List<String> phones = devicePhoneFetchUtil.getPhoneListByDeviceCode(deviceCode);
+        if (phones == null || phones.isEmpty()) return;
+        JSONObject params = new JSONObject();
+        params.put("deviceNo", deviceCode);
+        params.put("alarmType", type);
+        params.put("location", detail);
+        Map<String,Boolean> result = smsUtil.sendBatchSms(phones, params.toJSONString());
+        if (result == null || phones.stream().anyMatch(p -> !Boolean.TRUE.equals(result.get(p)))) {
+            throw new IllegalStateException("一级告警短信未全部发送成功");
+        }
+    }
+}

zk-common/pom.xml

@@ -17,6 +17,11 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
     <dependencies>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-codec</artifactId>
+            <optional>true</optional>
+        </dependency>
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-pool2</artifactId>
@@ -119,4 +124,4 @@
         </dependency>
     </dependencies>
 
-</project>
+</project>

zk-common/src/main/java/com/zksy/common/netty/DeviceFrameDecoder.java

@@ -0,0 +1,140 @@
+package com.zksy.common.netty;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import java.util.List;
+
+/** 每条连接独立累积报文；校验失败只丢弃坏帧，不关闭设备连接。 */
+public final class DeviceFrameDecoder extends ByteToMessageDecoder {
+    public enum Protocol { PINGSHENG, MANHOLE, GAS, FIRE_PRESSURE }
+    private static final int MAX_FRAME = 8192;
+    private final Protocol protocol;
+
+    public DeviceFrameDecoder(Protocol protocol) { this.protocol = protocol; }
+
+    @Override
+    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
+        int minimum = protocol == Protocol.MANHOLE ? 72 : protocol == Protocol.GAS ? 74 : 5;
+        while (in.readableBytes() >= minimum) {
+            int start = in.readerIndex();
+            int length;
+            if (protocol == Protocol.PINGSHENG) {
+                if (in.getUnsignedMedium(start) != 0x123456) { in.skipBytes(1); continue; }
+                length = in.getUnsignedShort(start + 3);
+                if (length < 10 || length > MAX_FRAME) { in.skipBytes(1); continue; }
+                if (in.readableBytes() < length) {
+                    // 长度字段损坏时，寻找后续完整且两层校验通过的上报帧。
+                    int next = nextPingshengFrame(in, start + 3);
+                    if (next < 0) return;
+                    in.readerIndex(next);
+                    continue;
+                }
+                int xor = 0;
+                for (int i = 0; i < length; i++) xor ^= in.getUnsignedByte(start + i);
+                if (xor != 0) { in.skipBytes(1); continue; }
+            } else if (protocol == Protocol.MANHOLE) {
+                // 当前井盖解析器：35字节参数 + 20字节SIM + 15字节IMEI + CRC16。
+                length = 72;
+                if (crc(in, start, 70) != in.getUnsignedShortLE(start + 70)) { in.skipBytes(1); continue; }
+            } else if (protocol == Protocol.GAS) {
+                // 当前已支持的7指标上报结构固定为74字节，内外两层校验与接收器一致。
+                length = 74;
+                if (in.getUnsignedByte(start) != 0x7E || in.getUnsignedByte(start + 15) != 0xAA
+                        || !gasChecksum(in, start)) { in.skipBytes(1); continue; }
+            } else {
+                if (in.getUnsignedShort(start) != 0x7E7E) { in.skipBytes(1); continue; }
+                // 现有消防设备长度字段不作为可信边界，使用结束符 + 完整CRC确认边界。
+                // 不能单靠03分隔：日期、TLV和压力数据中也可能出现03。
+                length = fireLength(in, start);
+                if (length == 0) {
+                    // 坏帧后可能紧接新报文，不能让坏CRC永久阻塞下一条有效数据。
+                    boolean recovered = false;
+                    for (int next = start + 2; next + 16 < in.writerIndex(); next++) {
+                        if (in.getUnsignedShort(next) == 0x7E7E && fireLength(in, next) > 0) {
+                            in.readerIndex(next);
+                            recovered = true;
+                            break;
+                        }
+                    }
+                    if (recovered) continue;
+                    if (in.readableBytes() >= MAX_FRAME) { in.skipBytes(1); continue; }
+                    return;
+                }
+            }
+            out.add(in.readRetainedSlice(length));
+        }
+    }
+
+    private static boolean gasChecksum(ByteBuf in, int start) {
+        int body = 0, full = 0;
+        for (int i = 16; i < 72; i++) body += in.getUnsignedByte(start + i);
+        for (int i = 3; i < 73; i++) full += in.getUnsignedByte(start + i);
+        return (body & 255) == in.getUnsignedByte(start + 72)
+                && ((255 - full) & 255) == in.getUnsignedByte(start + 73);
+    }
+
+    private static int nextPingshengFrame(ByteBuf in, int from) {
+        for (int start = from; start + 10 <= in.writerIndex(); start++) {
+            if (in.getUnsignedMedium(start) != 0x123456) continue;
+            int length = in.getUnsignedShort(start + 3);
+            if (length < 14 || length > MAX_FRAME || start + length > in.writerIndex()
+                    || in.getUnsignedByte(start + 6) != 0x31) continue;
+            int xor = 0;
+            for (int i = 0; i < length; i++) xor ^= in.getUnsignedByte(start + i);
+            if (xor != 0) continue;
+            int src = (in.getUnsignedByte(start + 7) + 1) / 2;
+            int destIndex = start + 8 + src;
+            if (src == 0 || src > 10 || destIndex >= start + length - 3) continue;
+            int dest = (in.getUnsignedByte(destIndex) + 1) / 2;
+            int body = destIndex + 1 + dest;
+            if (dest > 10 || body + 2 >= start + length - 3) continue;
+            if (crc(in, body, start + length - 3 - body) == in.getUnsignedShortLE(start + length - 3)) return start;
+        }
+        return -1;
+    }
+
+    private static int fireLength(ByteBuf in, int start) {
+        if (in.writerIndex() - start < 14) return 0;
+        if (in.getUnsignedByte(start + 10) != 0xEF) {
+            // 上报数据固定头之后按TLV跨过值区，值中偶然的03+CRC不能作为帧尾。
+            int cursor = start + 37;
+            int limit = Math.min(in.writerIndex(), start + MAX_FRAME);
+            while (cursor + 3 <= limit) {
+                int marker = in.getUnsignedByte(cursor);
+                if (marker == 0x03 || marker == 0x17) {
+                    return crc(in, start, cursor - start + 1) == in.getUnsignedShort(cursor + 1)
+                            ? cursor - start + 3 : 0;
+                }
+                int tag = in.getUnsignedShort(cursor);
+                int bytes = (in.getUnsignedByte(cursor + 2) >>> 3) & 31;
+                if (bytes == 0 && tag == 0x581A) bytes = 2;
+                if (bytes == 0 && tag == 0x4520) bytes = 4;
+                cursor += 3 + bytes;
+            }
+            return 0;
+        }
+        int value = 0xFFFF;
+        int end = Math.min(in.writerIndex() - 2, start + MAX_FRAME - 2);
+        for (int i = start; i < end; i++) {
+            value = crcByte(value, in.getUnsignedByte(i));
+            int marker = in.getUnsignedByte(i);
+            // 消防现有crc16Tall以高字节在前追加CRC。
+            if (i >= start + 14 && (marker == 0x03 || marker == 0x17)
+                    && value == in.getUnsignedShort(i + 1)) return i - start + 3;
+        }
+        return 0;
+    }
+
+    private static int crc(ByteBuf in, int start, int length) {
+        int value = 0xFFFF;
+        for (int i = 0; i < length; i++) value = crcByte(value, in.getUnsignedByte(start + i));
+        return value;
+    }
+
+    private static int crcByte(int value, int b) {
+        value ^= b;
+        for (int i = 0; i < 8; i++) value = (value & 1) == 1 ? (value >>> 1) ^ 0xA001 : value >>> 1;
+        return value;
+    }
+}