diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 00000000..7b016a89 --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,3 @@ +{ + "java.compile.nullAnalysis.mode": "automatic" +} \ No newline at end of file diff --git a/fastbee-server/base-server/src/main/java/com/fastbee/base/session/Session.java b/fastbee-server/base-server/src/main/java/com/fastbee/base/session/Session.java index 6751aa34..68c45bf7 100644 --- a/fastbee-server/base-server/src/main/java/com/fastbee/base/session/Session.java +++ b/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 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 Mono request(Message message, Class responseClass) { - final String key = requestKey(message, responseClass); + String key = requestKey(message, responseClass); + Mono subscribe = this.subscribe(key); + if (subscribe == null) { + return rejected; + } + ChannelFuture channelFuture = channel.writeAndFlush(Packet.of(this, message)); + return Mono.create(sink -> channelFuture.addListener(future -> { + if (future.isSuccess()) { + sink.success(future); + } else { + sink.error(future.cause()); + } + })).then(subscribe).doFinally(signal -> unsubscribe(key)); + } - // 关键修复：必须“先注册等待回调，再发送数据” - // 否则设备极快回包时，response() 可能发生在订阅登记之前，导致误判为普通上报并落库。 - return Mono.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 -> { - if (!future.isSuccess()) { - // 发送失败，移除订阅并返回错误 - topicSubscribers.remove(key); - sink.error(future.cause()); - } - }); - }) - .doFinally(signal -> unsubscribe(key)); + /** + * 订阅等待响应 + */ + private Mono subscribe(String key) { + if (topicSubscribers.containsKey(key)) { + return null; + } + return Mono.create(sink -> { + topicSubscribers.put(key, sink); + }); } /** diff --git a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/DelimiterBasedFrameDecoder.java b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/DelimiterBasedFrameDecoder.java index 218b707e..7e48b12f 100644 --- a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/DelimiterBasedFrameDecoder.java +++ b/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) - if (deviceAddr < 1 || deviceAddr > 247) { + // 检查设备地址：标准1-247，另外允许0x00(广播)和0xFF(常用扩展地址) + 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) { diff --git a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/TCPMessageAdapter.java b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/TCPMessageAdapter.java index abce2de1..abdb6a08 100644 --- a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/TCPMessageAdapter.java +++ b/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));