diff --git a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/TCPServer.java b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/TCPServer.java index 59852b04..3fe22167 100644 --- a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/TCPServer.java +++ b/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) @@ -48,22 +48,25 @@ public class TCPServer extends Server { .childOption(NioChannelOption.TCP_NODELAY, true) .childHandler(new ChannelInitializer() { - /*第二个处理器 session处理*/ + /* 第二个处理器 session处理 */ private final TCPMessageAdapter adapter = new TCPMessageAdapter(config.sessionManager); - /*3.解码 适配解码器 解码后业务处理*/ + /* 3.解码 适配解码器 解码后业务处理 */ private final MessageDecoderWrapper decoder = new MessageDecoderWrapper(config.decoder); - /*3.编码 适配编码器-编码后业务处理*/ + /* 3.编码 适配编码器-编码后业务处理 */ private final MessageEncoderWrapper encoder = new MessageEncoderWrapper(config.encoder); - /*4.编解码后消息分发器 同步和异步处理*/ - private final DispatcherHandler dispatcher = new DispatcherHandler(config.handlerMapping, config.handlerInterceptor, bossGroup); + /* 4.编解码后消息分发器 同步和异步处理 */ + 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); + // } } diff --git a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/ProtocolSniffingDecoder.java b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/ProtocolSniffingDecoder.java new file mode 100644 index 00000000..b57a246b --- /dev/null +++ b/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 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()); + } +}