20260126：建立协议解码器，协议ASCII\JSON\MODBUS分开解码

fastbee-server/iot-server-core/src/main/java/com/fastbee/server/TCPServer.java

@@ -1,6 +1,5 @@
 package com.fastbee.server;
 
-
 import com.fastbee.common.constant.FastBeeConstant;
 import com.fastbee.server.config.NettyConfig;
 import com.fastbee.server.handler.*;
@@ -11,7 +10,6 @@ import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioChannelOption;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
-import io.netty.handler.codec.ByteToMessageDecoder;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.concurrent.DefaultThreadFactory;
 
@@ -35,10 +33,12 @@ public class TCPServer extends Server {
     @Override
     protected AbstractBootstrap initialize() {
         bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory(config.name, Thread.MAX_PRIORITY));
-        workerGroup = new NioEventLoopGroup(config.workerCore, new DefaultThreadFactory(config.name, Thread.MAX_PRIORITY));
+        workerGroup = new NioEventLoopGroup(config.workerCore,
+                new DefaultThreadFactory(config.name, Thread.MAX_PRIORITY));
         if (config.businessCore > 0) {
             businessService = new ThreadPoolExecutor(config.businessCore, config.businessCore, 1L,
-                    TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new DefaultThreadFactory(config.name + "-B", true, Thread.NORM_PRIORITY));
+                    TimeUnit.SECONDS, new LinkedBlockingQueue<>(),
+                    new DefaultThreadFactory(config.name + "-B", true, Thread.NORM_PRIORITY));
         }
         return new ServerBootstrap()
                 .group(bossGroup, workerGroup)
@@ -55,15 +55,18 @@ public class TCPServer extends Server {
                     /* 3.编码 适配编码器-编码后业务处理 */
                     private final MessageEncoderWrapper encoder = new MessageEncoderWrapper(config.encoder);
                     /* 4.编解码后消息分发器 同步和异步处理 */
-                    private final DispatcherHandler dispatcher = new DispatcherHandler(config.handlerMapping, config.handlerInterceptor, bossGroup);
+                    private final DispatcherHandler dispatcher = new DispatcherHandler(config.handlerMapping,
+                            config.handlerInterceptor, bossGroup);
 
                     @Override
                     protected void initChannel(NioSocketChannel channel) throws Exception {
                         channel.pipeline()
                                 // 设置心跳时间
-                                .addLast(new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, config.allIdleTime))
+                                .addLast(new IdleStateHandler(config.readerIdleTime, config.writerIdleTime,
+                                        config.allIdleTime))
                                 // 第一步、添加粘包处理器
-                                .addLast(FastBeeConstant.SERVER.FRAMEDECODER, frameDecoder())
+                                .addLast(FastBeeConstant.SERVER.FRAMEDECODER,
+                                        new ProtocolSniffingDecoder(config.maxFrameLength))
                                 // 第二步、添加消息适配器
                                 .addLast(FastBeeConstant.SERVER.ADAPTER, adapter)
                                 // 第三步、添加报文解码器
@@ -79,12 +82,14 @@ public class TCPServer extends Server {
     /**
      * 添加TCP粘包处理器
      */
-    private ByteToMessageDecoder frameDecoder() {
-        if (config.lengthField == null) {
-            /*分隔符处理器，报文以固定包头包尾结束*/
-            return new DelimiterBasedFrameDecoder(config.maxFrameLength, config.delimiters);
-        }
-        /*报文长度的，以长度固定处理器和分隔符处理器 处理*/
-        return new LengthFieldAndDelimiterFrameDecoder(config.maxFrameLength, config.lengthField, config.delimiters);
-    }
+    // private ByteToMessageDecoder frameDecoder() {
+    // if (config.lengthField == null) {
+    // /*分隔符处理器，报文以固定包头包尾结束*/
+    // return new DelimiterBasedFrameDecoder(config.maxFrameLength,
+    // config.delimiters);
+    // }
+    // /*报文长度的，以长度固定处理器和分隔符处理器 处理*/
+    // return new LengthFieldAndDelimiterFrameDecoder(config.maxFrameLength,
+    // config.lengthField, config.delimiters);
+    // }
 }

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

@@ -0,0 +1,419 @@
+package com.fastbee.server.handler;
+
+import com.fastbee.base.session.Session;
+import com.fastbee.base.util.AttributeUtils;
+import com.fastbee.common.utils.gateway.CRC16Utils;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.codec.TooLongFrameException;
+import io.netty.handler.timeout.IdleState;
+import io.netty.handler.timeout.IdleStateEvent;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.List;
+
+/**
+ * 多协议嗅探解码器
+ * 根据首字节特征自动检测协议类型，应用对应的帧切分策略：
+ * 1. JSON: 以 '{' 开头，大括号匹配切分
+ * 2. 7E帧: 以 0x7E 开头，固定长度9字节（注册包/心跳包）
+ * 3. Modbus RTU: 地址+功能码+CRC校验
+ * 4. ASCII: 其他全部当作ASCII，在channelReadComplete时整块输出
+ *
+ * @author bill
+ */
+@Slf4j
+public class ProtocolSniffingDecoder extends ByteToMessageDecoder {
+
+    private final int maxFrameLength;
+    
+    /**
+     * 标记是否有未识别的数据（ASCII）
+     */
+    private boolean hasUnrecognizedData = false;
+
+    public ProtocolSniffingDecoder(int maxFrameLength) {
+        this.maxFrameLength = maxFrameLength;
+    }
+
+    @Override
+    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+        while (in.isReadable()) {
+            // 调试日志
+            if (log.isDebugEnabled() && in.readableBytes() > 0) {
+                byte[] temp = new byte[Math.min(in.readableBytes(), 100)];
+                in.getBytes(in.readerIndex(), temp);
+                log.debug("[{}] 协议嗅探器收到数据: {}, 长度={}", getDeviceId(ctx), ByteBufUtil.hexDump(temp), in.readableBytes());
+            }
+
+            int firstByte = in.getByte(in.readerIndex()) & 0xFF;
+            String deviceId = getDeviceId(ctx);
+
+            /** ========== 1. JSON检测: 以 '{' | '['开头 ========== */
+            if (firstByte == 0x7B || firstByte == 0x5B) {
+                ByteBuf frame = decodeJson(in, deviceId);
+                if (frame != null) {
+                    out.add(frame);
+                    hasUnrecognizedData = false;
+                    continue;
+                }
+                // JSON不完整，等待更多数据
+                return;
+            }
+
+            /** 2. 7E帧检测【7E + 80(注册包) 或 7E + 81(心跳包)】 */
+
+            if (firstByte == 0x7E) {
+                int len = decode7EFrame(in, deviceId);
+                if (len > 0) {
+                    out.add(in.readRetainedSlice(len));
+                    hasUnrecognizedData = false;
+                    continue;
+                } else if (len == 0) {
+                    // 数据不完整，等待
+                    return;
+                }
+                // len == -1: 不是有效7E帧，当作ASCII处理
+            }
+
+            /** 3. Modbus RTU检测；地址+功能码+CRC校验 */
+            int modbusLen = decodeModbusRtu(in, deviceId);
+            if (modbusLen > 0) {
+                out.add(in.readRetainedSlice(modbusLen));
+                hasUnrecognizedData = false;
+                continue;
+            } else if (modbusLen == 0) {
+                // 数据不完整，等待
+                return;
+            }
+
+            /** 4.在ASCII数据中查找7E帧、JSON、Modbus的起始位置（目前只只检测注册/心跳） */
+            ByteBuf asciiFrame = decodeAsciiWithProtocolScan(in, deviceId);
+            if (asciiFrame != null) {
+                out.add(asciiFrame);
+                hasUnrecognizedData = false;
+                // 继续循环，处理可能的粘包数据
+                continue;
+            }
+            
+            // ASCII数据不完整或没有找到其他协议边界，等待channelReadComplete
+            hasUnrecognizedData = true;
+            return;
+        }
+    }
+
+    /**
+     * ASCII解码 - 扫描是否有其他协议数据粘包
+     * 在ASCII数据中查找JSON、7E帧、Modbus的起始位置
+     *
+     * @return ASCII帧（截断到其他协议起始位置），或null（等待更多数据）
+     */
+    private ByteBuf decodeAsciiWithProtocolScan(ByteBuf in, String deviceId) {
+        int start = in.readerIndex();
+        int end = in.writerIndex();
+
+        // 从第二个字节开始扫描（第一个字节已经确认是ASCII）
+        for (int i = start + 1; i < end; i++) {
+            int b = in.getByte(i) & 0xFF;
+
+            // ========== 检测7E帧起始 ==========
+            if (b == 0x7E && i + 1 < end) {
+                int nextByte = in.getByte(i + 1) & 0xFF;
+                // 7E + 80(注册包) 或 7E + 81(心跳包)
+                if (nextByte == 0x80 || nextByte == 0x81) {
+                    log.debug("[{}] 在ASCII中检测到7E帧起始，位置={}，类型={}", deviceId, i - start, nextByte == 0x80 ? "注册包" : "心跳包");
+                    return extractAsciiFrame(in, start, i, deviceId);
+                }
+            }
+        }
+
+        // 没有找到其他协议的起始位置;检查是否超长
+        if (in.readableBytes() > maxFrameLength) {
+            log.warn("[{}] ASCII数据超长，强制输出，长度={}", deviceId, in.readableBytes());
+            return in.readRetainedSlice(in.readableBytes());
+        }
+
+        return null; // 等待更多数据或channelReadComplete
+    }
+
+    /**
+     * 提取ASCII帧（从start到protocolStart之间的数据）
+     */
+    private ByteBuf extractAsciiFrame(ByteBuf in, int start, int protocolStart, String deviceId) {
+        int length = protocolStart - start;
+        if (length > 0) {
+            ByteBuf frame = in.readRetainedSlice(length);
+            log.debug("[{}] 提取ASCII帧，长度={}", deviceId, length);
+            return frame;
+        }
+        return null;
+    }
+
+    /**
+     * 每次read完成后调用
+     * 如果有未识别的数据（ASCII），直接全部输出
+     */
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+        if (hasUnrecognizedData) {
+            ByteBuf cumulation = internalBuffer();
+            if (cumulation != null && cumulation.isReadable()) {
+                log.debug("[{}] channelReadComplete: 输出ASCII数据，长度={}", getDeviceId(ctx), cumulation.readableBytes());
+                ctx.fireChannelRead(cumulation.readRetainedSlice(cumulation.readableBytes()));
+            }
+            hasUnrecognizedData = false;
+        }
+        super.channelReadComplete(ctx);
+    }
+
+    /**
+     * 处理空闲事件 - 超时兜底
+     */
+    @Override
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+        if (evt instanceof IdleStateEvent) {
+            IdleState state = ((IdleStateEvent) evt).state();
+            if (state == IdleState.READER_IDLE) {
+                // 读取空闲，强制输出缓冲区
+                ByteBuf cumulation = internalBuffer();
+                if (cumulation != null && cumulation.isReadable()) {
+                    log.debug("[{}] 读取超时，强制输出缓冲区，长度={}", getDeviceId(ctx), cumulation.readableBytes());
+                    ctx.fireChannelRead(cumulation.readRetainedSlice(cumulation.readableBytes()));
+                    hasUnrecognizedData = false;
+                }
+            }
+        }
+        super.userEventTriggered(ctx, evt);
+    }
+
+    /**
+     * JSON帧解码 - 通过大括号匹配切分
+     */
+    private ByteBuf decodeJson(ByteBuf in, String deviceId) {
+        int start = in.readerIndex();
+        int depth = 0;
+        boolean inString = false;
+        boolean escape = false;
+
+        for (int i = start; i < in.writerIndex(); i++) {
+            byte b = in.getByte(i);
+
+            // 处理字符串内的转义
+            if (escape) {
+                escape = false;
+                continue;
+            }
+            if (b == '\\' && inString) {
+                escape = true;
+                continue;
+            }
+
+            // 字符串边界
+            if (b == '"') {
+                inString = !inString;
+                continue;
+            }
+
+            // 只在字符串外计算括号
+            if (!inString) {
+                if (b == '{' || b == '[') {
+                    depth++;
+                } else if (b == '}' || b == ']') {
+                    depth--;
+                }
+
+                // 括号匹配完成 = JSON完整
+                if (depth == 0 && i > start) {
+                    int length = i - start + 1;
+                    ByteBuf frame = in.readRetainedSlice(length);
+                    // 跳过尾部的空白字符
+                    skipWhitespace(in);
+                    log.debug("[{}] 成功解析JSON帧，长度={}", deviceId, length);
+                    return frame;
+                }
+            }
+        }
+
+        // 检查是否超长
+        if (in.readableBytes() > maxFrameLength) {
+            log.warn("[{}] JSON帧超长，丢弃，长度={}", deviceId, in.readableBytes());
+            in.skipBytes(in.readableBytes());
+            throw new TooLongFrameException("JSON frame length exceeds " + maxFrameLength);
+        }
+
+        return null; // JSON不完整，等待更多数据
+    }
+
+    /**
+     * 7E帧解码 - 固定长度切分
+     * 注册包格式：7E(1) + 80(1) + MAC地址(6) + 7E(1) = 9字节
+     * 心跳包格式：7E(1) + 81(1) + MAC地址(6) + 7E(1) = 9字节
+     *
+     * @return -1: 不是7E帧; 0: 是7E帧但数据不完整; >0: 帧长度
+     */
+    private int decode7EFrame(ByteBuf in, String deviceId) {
+        int readable = in.readableBytes();
+        if (readable < 2) {
+            return -1;
+        }
+
+        int readerIndex = in.readerIndex();
+        int secondByte = in.getByte(readerIndex + 1) & 0xFF;
+
+        // 只处理注册包(0x80)和心跳包(0x81)
+        if (secondByte != 0x80 && secondByte != 0x81) {
+            return -1;
+        }
+
+        int frameLength = 9;
+        if (readable < frameLength) {
+            log.debug("[{}] 7E帧数据不完整，需要{}字节，当前{}字节", deviceId, frameLength, readable);
+            return 0; // 数据不完整
+        }
+
+        // 验证帧尾
+        int lastByte = in.getByte(readerIndex + frameLength - 1) & 0xFF;
+        if (lastByte != 0x7E) {
+            log.warn("[{}] 7E帧格式错误，帧尾不是7E，实际值: 0x{}", deviceId, Integer.toHexString(lastByte));
+            return -1;
+        }
+
+        log.debug("[{}] 成功检测7E帧，类型={}, 长度={}", deviceId, secondByte == 0x80 ? "注册包" : "心跳包", frameLength);
+        return frameLength;
+    }
+
+    /**
+     * Modbus RTU解码 - 长度计算 + CRC校验切分
+     *
+     * @return -1: 不是Modbus帧; 0: 是Modbus帧但数据不完整; >0: 帧长度
+     */
+    private int decodeModbusRtu(ByteBuf in, String deviceId) {
+        int readable = in.readableBytes();
+        if (readable < 2) {
+            return -1;
+        }
+
+        int readerIndex = in.readerIndex();
+        int deviceAddr = in.getByte(readerIndex) & 0xFF;
+        int functionCode = in.getByte(readerIndex + 1) & 0xFF;
+
+        // 排除7E开头的数据
+        if (deviceAddr == 0x7E) {
+            return -1;
+        }
+
+        // 检查设备地址：标准1-247，另外允许0x00(广播)和0xFF(常用扩展地址)
+        boolean validAddr = (deviceAddr >= 1 && deviceAddr <= 247)
+                || deviceAddr == 0x00
+                || deviceAddr == 0xFF;
+        if (!validAddr) {
+            return -1;
+        }
+
+        // 异常响应：地址(1) + 功能码(1|0x80) + 异常码(1) + CRC(2) = 5字节
+        if ((functionCode & 0x80) != 0) {
+            int len = 5;
+            if (readable < len) {
+                return 0;
+            }
+            return isValidModbusCrc(in, readerIndex, len) ? len : -1;
+        }
+
+        // 功能码05/06：请求与响应均为8字节
+        if (functionCode == 0x05 || functionCode == 0x06) {
+            int len = 8;
+            if (readable < len) {
+                return 0;
+            }
+            return isValidModbusCrc(in, readerIndex, len) ? len : -1;
+        }
+
+        // 功能码01/02/03/04：
+        if (functionCode >= 0x01 && functionCode <= 0x04) {
+            // 先尝试8字节（请求/回显固定长度）
+            int reqLen = 8;
+            if (readable >= reqLen && isValidModbusCrc(in, readerIndex, reqLen)) {
+                return reqLen;
+            }
+
+            if (readable < 3) {
+                return 0;
+            }
+            int byteCount = in.getByte(readerIndex + 2) & 0xFF;
+            int respLen = 3 + byteCount + 2;
+            if (respLen > maxFrameLength) {
+                log.warn("[{}] Modbus帧长度{}超过最大限制{}", deviceId, respLen, maxFrameLength);
+                return -1;
+            }
+            if (readable < respLen) {
+                return 0;
+            }
+            return isValidModbusCrc(in, readerIndex, respLen) ? respLen : -1;
+        }
+
+        // 功能码0F/10：
+        if (functionCode == 0x0F || functionCode == 0x10) {
+            int respLen = 8;
+            if (readable >= respLen && isValidModbusCrc(in, readerIndex, respLen)) {
+                return respLen;
+            }
+
+            if (readable < 7) {
+                return 0;
+            }
+            int byteCount = in.getByte(readerIndex + 6) & 0xFF;
+            int reqLen = 9 + byteCount;
+            if (reqLen > maxFrameLength) {
+                return -1;
+            }
+            if (readable < reqLen) {
+                return 0;
+            }
+            return isValidModbusCrc(in, readerIndex, reqLen) ? reqLen : -1;
+        }
+
+        return -1; // 不是已知的Modbus功能码
+    }
+
+    /**
+     * 校验Modbus RTU CRC16（低位在前，高位在后）
+     */
+    private boolean isValidModbusCrc(ByteBuf buffer, int offset, int len) {
+        if (len < 5) {
+            return false;
+        }
+        byte[] data = new byte[len - 2];
+        buffer.getBytes(offset, data);
+        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);
+    }
+
+    /**
+     * 跳过空白字符
+     */
+    private void skipWhitespace(ByteBuf in) {
+        while (in.isReadable()) {
+            byte b = in.getByte(in.readerIndex());
+            if (b == ' ' || b == '\r' || b == '\n' || b == '\t') {
+                in.skipBytes(1);
+            } else {
+                break;
+            }
+        }
+    }
+
+    /**
+     * 获取设备标识（用于日志）
+     */
+    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());
+    }
+}