20260107:

1、对Session优化，存在下发指令失败问题

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "java.compile.nullAnalysis.mode": "automatic"
+}

fastbee-server/base-server/src/main/java/com/fastbee/base/session/Session.java

@@ -40,6 +40,11 @@ public class Session {
      * 这里必须保证并发可见性与线程安全：Netty IO线程会调用 response()，而业务线程可能在发起 request()。
      */
     private final Map<String, MonoSink> topicSubscribers = new ConcurrentHashMap<>();
+    
+    /**
+     * 请求被拒绝时返回的 Mono
+     */
+    private static final Mono rejected = Mono.error(new RejectedExecutionException("重复的request订阅"));
     /**
      * 是否存在等待中的 requestR()。
      * 用于入站侧判断是否需要尝试 response 匹配（避免无意义的匹配开销，也方便业务侧做更精细的路由）。
@@ -259,27 +264,31 @@ public class Session {
      * 订阅回调 mono.doOnSuccess(处理成功).doOnError(处理异常).subscribe(开始订阅)
      */
     public <T> Mono<T> request(Message message, Class<T> responseClass) {
-        final String key = requestKey(message, responseClass);
+        String key = requestKey(message, responseClass);
-
+        Mono<T> subscribe = this.subscribe(key);
-        // 关键修复：必须“先注册等待回调，再发送数据”
+        if (subscribe == null) {
-        // 否则设备极快回包时，response() 可能发生在订阅登记之前，导致误判为普通上报并落库。
+            return rejected;
-        return Mono.<T>create(sink -> {
-                    MonoSink existing = topicSubscribers.putIfAbsent(key, sink);
-                    if (existing != null) {
-                        sink.error(new RejectedExecutionException("重复的request订阅:" + key));
-                        return;
         }
-
         ChannelFuture channelFuture = channel.writeAndFlush(Packet.of(this, message));
-                    channelFuture.addListener(future -> {
+        return Mono.create(sink -> channelFuture.addListener(future -> {
-                        if (!future.isSuccess()) {
+            if (future.isSuccess()) {
-                            // 发送失败，移除订阅并返回错误
+                sink.success(future);
-                            topicSubscribers.remove(key);
+            } else {
                 sink.error(future.cause());
             }
+        })).then(subscribe).doFinally(signal -> unsubscribe(key));
+    }
+
+    /**
+     * 订阅等待响应
+     */
+    private <T> Mono<T> subscribe(String key) {
+        if (topicSubscribers.containsKey(key)) {
+            return null;
+        }
+        return Mono.create(sink -> {
+            topicSubscribers.put(key, sink);
         });
-                })
-                .doFinally(signal -> unsubscribe(key));
     }
 
     /**

fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/DelimiterBasedFrameDecoder.java

@@ -2,7 +2,10 @@ package com.fastbee.server.handler;
 
 
 import com.fastbee.base.codec.Delimiter;
+import com.fastbee.base.session.Session;
+import com.fastbee.base.util.AttributeUtils;
 import com.fastbee.base.util.ByteBufUtils;
+import io.netty.buffer.ByteBufUtil;
 import com.fastbee.common.utils.gateway.CRC16Utils;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
@@ -62,9 +65,30 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
         }
     }
 
+    /**
+     * 获取设备标识（用于日志）
+     */
+    private String getDeviceId(ChannelHandlerContext ctx) {
+        Session session = AttributeUtils.getSession(ctx.channel());
+        if (session != null && session.getClientId() != null) {
+            return session.getClientId();
+        }
+        // 未注册时使用远程地址
+        return String.valueOf(ctx.channel().remoteAddress());
+    }
+    
     protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) {
+        // 调试：记录收到的原始数据
+        if (log.isDebugEnabled() && buffer.readableBytes() > 0) {
+            byte[] temp = new byte[Math.min(buffer.readableBytes(), 100)];
+            buffer.getBytes(buffer.readerIndex(), temp);
+            log.debug("[{}] 解码器收到数据: {}，长度={}", getDeviceId(ctx), ByteBufUtil.hexDump(temp), buffer.readableBytes());
+        }
+        
+        String deviceId = getDeviceId(ctx);
+        
         // 优先级1：检测7E帧格式（注册包/心跳包）
-        int frame7EResult = check7EFrame(buffer);
+        int frame7EResult = check7EFrame(buffer, deviceId);
         if (frame7EResult > 0) {
             // 是7E帧且数据完整，读取并返回
             return buffer.readRetainedSlice(frame7EResult);
@@ -74,11 +98,11 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
         }
         
         // 优先级2：检测Modbus RTU帧格式（避免数据中的7E被当作分隔符）
-        int modbusResult = checkModbusRtu(buffer);
+        int modbusResult = checkModbusRtu(buffer, deviceId);
         if (modbusResult > 0) {
             // 是Modbus帧且数据完整，读取并返回
             ByteBuf frame = buffer.readRetainedSlice(modbusResult);
-            log.debug("成功解析Modbus RTU帧，长度={}", modbusResult);
+            log.debug("[{}] 成功解析Modbus RTU帧，长度={}", deviceId, modbusResult);
             return frame;
         } else if (modbusResult == 0) {
             // 是Modbus帧但数据不完整，等待更多数据，不要用分隔符处理
@@ -86,6 +110,7 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
         }
         
         // 优先级3：使用分隔符处理（兜底方案，仅当不是7E帧也不是Modbus帧时）
+        log.debug("[{}] 非7E帧非Modbus帧，尝试分隔符处理，7E结果={}, Modbus结果={}", deviceId, frame7EResult, modbusResult);
         int minFrameLength = Integer.MAX_VALUE;
         Delimiter minDelim = null;
         for (Delimiter delim : delimiters) {
@@ -165,7 +190,7 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
      * 
      * @return -1: 不是7E帧; 0: 是7E帧但数据不完整; >0: 帧长度（数据完整）
      */
-    private int check7EFrame(ByteBuf buffer) {
+    private int check7EFrame(ByteBuf buffer, String deviceId) {
         int readable = buffer.readableBytes();
         if (readable < 2) {
             return -1;
@@ -190,18 +215,18 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
         
         // 检查数据是否足够
         if (readable < frameLength) {
-            log.debug("7E帧数据不完整，需要{}字节，当前{}字节，等待更多数据", frameLength, readable);
+            log.debug("[{}] 7E帧数据不完整，需要{}字节，当前{}字节，等待更多数据", deviceId, frameLength, readable);
             return 0; // 是7E帧但数据不完整
         }
         
         // 验证帧尾是否为7E
         int lastByte = buffer.getByte(readerIndex + frameLength - 1) & 0xFF;
         if (lastByte != 0x7E) {
-            log.warn("7E帧格式错误，帧尾不是7E，实际值: {}", lastByte);
+            log.warn("[{}] 7E帧格式错误，帧尾不是7E，实际值: {}", deviceId, lastByte);
             return -1;
         }
         
-        log.debug("成功检测7E帧，类型={}, 长度={}", secondByte == 0x80 ? "注册包" : "心跳包", frameLength);
+        log.debug("[{}] 成功检测7E帧，类型={}, 长度={}", deviceId, secondByte == 0x80 ? "注册包" : "心跳包", frameLength);
         return frameLength;
     }
     
@@ -209,7 +234,7 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
      * 检测Modbus RTU帧格式
      * @return -1: 不是Modbus帧; 0: 是Modbus帧但数据不完整; >0: 帧长度（数据完整）
      */
-    private int checkModbusRtu(ByteBuf buffer) {
+    private int checkModbusRtu(ByteBuf buffer, String deviceId) {
         int readable = buffer.readableBytes();
         if (readable < 2) {
             return -1;
@@ -224,13 +249,16 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
             return -1;
         }
         
-        // 检查设备地址(1-247)
+        // 检查设备地址：标准1-247，另外允许0x00(广播)和0xFF(常用扩展地址)
-        if (deviceAddr < 1 || deviceAddr > 247) {
+        boolean validAddr = (deviceAddr >= 1 && deviceAddr <= 247) 
+                         || deviceAddr == 0x00 
+                         || deviceAddr == 0xFF;
+        if (!validAddr) {
             return -1;
         }
 
         // Modbus RTU：为了避免把请求帧(常见固定8字节)误当响应帧(第3字节是byteCount)导致截断，
-        // 这里采用“候选长度 + CRC校验”方式确定帧边界，兼容：请求/回显/响应/异常响应。
+        // 这里采用"候选长度 + CRC校验"方式确定帧边界，兼容：请求/回显/响应/异常响应。
 
         // 异常响应：地址(1) + 功能码(1|0x80) + 异常码(1) + CRC(2) = 5字节
         if ((functionCode & 0x80) != 0) {
@@ -238,7 +266,7 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
             if (readable < len) {
                 return 0;
             }
-            return isValidModbusCrc(buffer, readerIndex, len) ? len : -1;
+            return isValidModbusCrc(buffer, readerIndex, len, deviceId) ? len : -1;
         }
 
         // 功能码05/06：请求与响应均为8字节（地址 + 功能码 + 地址(2) + 值(2) + CRC(2)）
@@ -247,7 +275,7 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
             if (readable < len) {
                 return 0;
             }
-            return isValidModbusCrc(buffer, readerIndex, len) ? len : -1;
+            return isValidModbusCrc(buffer, readerIndex, len, deviceId) ? len : -1;
         }
 
         // 功能码01/02/03/04：
@@ -256,7 +284,7 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
         if (functionCode >= 0x01 && functionCode <= 0x04) {
             // 先尝试8字节（避免地址高字节=0x00 被误当 byteCount=0 导致5字节截断）
             int reqLen = 8;
-            if (readable >= reqLen && isValidModbusCrc(buffer, readerIndex, reqLen)) {
+            if (readable >= reqLen && isValidModbusCrc(buffer, readerIndex, reqLen, deviceId)) {
                 return reqLen;
             }
 
@@ -265,13 +293,21 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
             }
             int byteCount = buffer.getByte(readerIndex + 2) & 0xFF;
             int respLen = 3 + byteCount + 2;
+            log.debug("[{}] Modbus功能码{}: byteCount={}, 预期帧长={}, 可读={}", 
+                    deviceId, String.format("%02X", functionCode), byteCount, respLen, readable);
             if (respLen > maxFrameLength) {
+                log.warn("[{}] Modbus帧长度{}超过最大限制{}", deviceId, respLen, maxFrameLength);
                 return -1;
             }
             if (readable < respLen) {
+                log.debug("[{}] Modbus帧数据不完整，需要{}字节，当前{}字节", deviceId, respLen, readable);
                 return 0;
             }
-            return isValidModbusCrc(buffer, readerIndex, respLen) ? respLen : -1;
+            boolean crcValid = isValidModbusCrc(buffer, readerIndex, respLen, deviceId);
+            if (!crcValid) {
+                log.debug("[{}] Modbus CRC校验失败，帧长={}", deviceId, respLen);
+            }
+            return crcValid ? respLen : -1;
         }
 
         // 功能码0F/10：
@@ -279,7 +315,7 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
         // - 请求可变：地址(1)+功能码(1)+起始地址(2)+数量(2)+byteCount(1)+values(N)+CRC(2) => 9+N
         if (functionCode == 0x0F || functionCode == 0x10) {
             int respLen = 8;
-            if (readable >= respLen && isValidModbusCrc(buffer, readerIndex, respLen)) {
+            if (readable >= respLen && isValidModbusCrc(buffer, readerIndex, respLen, deviceId)) {
                 return respLen;
             }
 
@@ -294,7 +330,7 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
             if (readable < reqLen) {
                 return 0;
             }
-            return isValidModbusCrc(buffer, readerIndex, reqLen) ? reqLen : -1;
+            return isValidModbusCrc(buffer, readerIndex, reqLen, deviceId) ? reqLen : -1;
         }
 
         return -1; // 不是已知的Modbus功能码
@@ -305,8 +341,9 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
      * @param buffer ByteBuf
      * @param offset 起始偏移（通常为 readerIndex）
      * @param len    帧长度（包含CRC2字节）
+     * @param deviceId 设备标识（用于日志）
      */
-    private boolean isValidModbusCrc(ByteBuf buffer, int offset, int len) {
+    private boolean isValidModbusCrc(ByteBuf buffer, int offset, int len, String deviceId) {
         if (len < 5) {
             return false;
         }
@@ -315,7 +352,13 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
         byte[] crc = CRC16Utils.getCrc16Byte(data); // [low, high]
         int crcLo = buffer.getByte(offset + len - 2) & 0xFF;
         int crcHi = buffer.getByte(offset + len - 1) & 0xFF;
-        return ((crc[0] & 0xFF) == crcLo) && ((crc[1] & 0xFF) == crcHi);
+        boolean valid = ((crc[0] & 0xFF) == crcLo) && ((crc[1] & 0xFF) == crcHi);
+        if (!valid && log.isDebugEnabled()) {
+            log.debug("[{}] CRC校验: 计算值={}{}, 实际值={}{}", deviceId,
+                    String.format("%02X", crc[0] & 0xFF), String.format("%02X", crc[1] & 0xFF),
+                    String.format("%02X", crcLo), String.format("%02X", crcHi));
+        }
+        return valid;
     }
 
     private void fail(long frameLength) {

fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/TCPMessageAdapter.java

@@ -50,13 +50,6 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter {
         ByteBuf buf = (ByteBuf) msg;
         /*session存储*/
         Session session = getSession(ctx);
-        // 临时记录设备原始数据日志
-        // 针对特定设备记录原始数据日志
-        String serialNumber = session.getClientId();
-        ByteBuf copy = buf.duplicate();
-        byte[] source = new byte[buf.writerIndex()];
-        copy.readBytes(source);
-        log.warn("==设备[{}]原始数据:{}", serialNumber, ByteBufUtil.hexDump(source));
         session.access();
         /*将设备端消息传递给下一个处理器*/
         ctx.fireChannelRead(Packet.of(session, buf));