diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/aspect/DeviceDebugLogAspect.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/aspect/DeviceDebugLogAspect.java index 242433dc..0586faeb 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/aspect/DeviceDebugLogAspect.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/aspect/DeviceDebugLogAspect.java @@ -309,18 +309,18 @@ public class DeviceDebugLogAspect { // 1、从Redis中获取用户ID // 2、从SecurityContext中获取用户ID - try { - Long userId = SecurityUtils.getUserId(); - if (userId == null) { - return false; - } - String cacheKey = CacheConstants.DEVICE_DEBUG_USER_KEY + userId; - String flag = redisCache.getCacheObject(cacheKey); - return "1".equals(flag); - } catch (Exception e) { - // 如果获取用户信息失败，默认不记录日志，避免影响正常功能 - log.debug("无法从SecurityContext获取用户ID（可能在异步线程中）"); - } +// try { +// Long userId = SecurityUtils.getUserId(); +// if (userId == null) { +// return false; +// } +// String cacheKey = CacheConstants.DEVICE_DEBUG_USER_KEY + userId; +// String flag = redisCache.getCacheObject(cacheKey); +// return "1".equals(flag); +// } catch (Exception e) { +// // 如果获取用户信息失败，默认不记录日志，避免影响正常功能 +// log.debug("无法从SecurityContext获取用户ID（可能在异步线程中）"); +// } // 3、从数据库中获取用户ID diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/executor/PowerOnTriggerTaskExecutor.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/executor/PowerOnTriggerTaskExecutor.java index 74e47042..a87c2b52 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/executor/PowerOnTriggerTaskExecutor.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/executor/PowerOnTriggerTaskExecutor.java @@ -27,6 +27,8 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Component; import org.springframework.util.CollectionUtils; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import javax.annotation.Resource; import java.time.Duration; @@ -34,7 +36,9 @@ import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; /** * 设备上电触发任务执行器 @@ -67,6 +71,11 @@ public class PowerOnTriggerTaskExecutor { @Resource private IModbusJobService modbusJobService; + /** + * 每台设备同一时刻仅允许一个“上电触发轮询”流水线运行，避免重复触发/并发下发导致应答错配。 + */ + private final ConcurrentHashMap running = new ConcurrentHashMap<>(); + /** * 标记设备需要触发上电任务（等待第一个数据包） * 优化：直接查询 ModbusJob，缓存 taskId 列表，避免后续重复查询数据库 @@ -279,125 +288,102 @@ public class PowerOnTriggerTaskExecutor { } } - // 执行所有相关的modbus任务 - int successCount = 0; - int timeoutCount = 0; - int totalCount = 0; - - for (ModbusJob modbusJob : modbusJobList) { - if (!"0".equals(modbusJob.getStatus())) { - // 只执行正常状态的任务 - continue; - } - - totalCount++; - String command = modbusJob.getCommand(); - if (StringUtils.isEmpty(command)) { - log.warn("设备上电触发任务执行失败：指令为空，设备编号：{}，任务ID：{}", serialNumber, modbusJob.getTaskId()); - continue; - } - - try { - // 转换指令为字节数组 - byte[] commandData = ByteUtils.convertToHexByteArray(command); - if (commandData.length == 0) { - log.warn("设备上电触发任务执行失败：指令转换失败，设备编号：{}，指令：{}", serialNumber, command); - continue; - } - - // 记录发送时间 - long sendTime = System.currentTimeMillis(); - boolean responseReceived; - - // 根据传输协议下发指令 - switch (serverType) { - case MQTT: - // 通过MQTT内部客户端下发指令 - pubMqttClient.publish(commandData, topic, false, 0); - log.info("=>上电触发-MQTT-线程=[{}],轮询指令:[{}],主题:[{}]", Thread.currentThread().getName(), command, topic); - // MQTT协议无法直接等待响应，等待200ms后继续 - try { - Thread.sleep(200); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - log.warn("MQTT指令等待被中断，设备编号：{}", serialNumber); - } - responseReceived = true; // MQTT协议假设发送成功 - break; - case TCP: - // 下发TCP客户端，等待200ms内的响应 - Message msg = new Message(); - msg.setClientId(serialNumber); - msg.setPayload(Unpooled.wrappedBuffer(commandData)); - // 为每个请求设置唯一的serNo，避免多个并发请求使用相同的requestKey - Session session = sessionStore.getSession(serialNumber); - if (session != null) { - int serNo = session.nextSerialNO(); - msg.setSerNo(String.valueOf(serNo)); - } - try { - // 等待200ms内的响应，直接检查设备是否回复了Message对象 - // DispatcherHandler会在协议解码后，如果识别为设备回复（isReply=true）， - // 主动调用Session.response()来匹配等待的requestR()调用 - AjaxResult result = messageManager.requestR(serialNumber, msg, Message.class) - .timeout(Duration.ofMillis(200)) // 200ms超时 - .doOnError(error -> log.warn("=>上电触发-TCP-线程=[{}],轮询指令:[{}],响应超时或失败:{}", Thread.currentThread().getName(), command, error.getMessage())) - .block(); // 阻塞等待响应，最多200ms - - // 检查返回结果：code为200（SUCCESS）且data字段有Message对象，说明设备回复了 - if (result != null) { - Integer code = (Integer) result.get(AjaxResult.CODE_TAG); - Object data = result.get(AjaxResult.DATA_TAG); - - // 只有code为200且data是Message对象时，才认为设备回复了 - responseReceived = (code != null && code == HttpStatus.SUCCESS && data != null && data instanceof Message); - - if (responseReceived) { - log.info("=>上电触发-TCP-线程=[{}],轮询指令:[{}],设备已回复，code:{}", Thread.currentThread().getName(), command, code); - // 释放Message对象的ByteBuf（DeviceReport的payload） - Message responseMessage = (Message) data; - if (responseMessage != null && responseMessage.getPayload() != null) { - responseMessage.getPayload().release(); - } - } else { - log.warn("=>上电触发-TCP-线程=[{}],轮询指令:[{}],设备未回复，code:{}, data:{}", Thread.currentThread().getName(), command, code, data != null ? data.getClass().getSimpleName() : "null"); - } - } else { - responseReceived = false; - log.warn("=>上电触发-TCP-线程=[{}],轮询指令:[{}],响应结果为空", Thread.currentThread().getName(), command); - } - } catch (Exception e) { - // 200ms内未收到响应或发生异常，视为超时 - long elapsed = System.currentTimeMillis() - sendTime; - log.warn("=>上电触发-TCP-线程=[{}],轮询指令:[{}],200ms内未收到响应，耗时:{}ms，异常:{}", Thread.currentThread().getName(), command, elapsed, e.getMessage()); - responseReceived = false; - // 超时后的回复会被DispatcherHandler识别为isReply=true，但匹配失败（订阅已取消） - // 会继续发送到MQ，但不会误判为普通数据包，因为isReply=true - } - break; - default: - log.warn("设备上电触发任务执行失败：不支持的传输协议，设备编号：{}，协议：{}", serialNumber, serverType); - continue; - } - - // 只有在200ms内收到响应时才更新任务执行时间和从列表中删除 taskId - if (responseReceived) { - successCount++; - // 更新任务执行时间 - updateTaskExecuteTime(modbusJob.getTaskId()); - // 从 taskId 列表中删除已成功执行的任务 - removeTaskIdFromList(taskIdsKey, modbusJob.getTaskId()); - } else { - timeoutCount++; - log.warn("设备上电触发任务超时，不更新任务状态，设备编号：{}，任务ID：{}", serialNumber, modbusJob.getTaskId()); - } - - } catch (Exception e) { - log.error("设备上电触发任务执行异常，设备编号：{}，任务ID：{}，指令：{}", serialNumber, modbusJob.getTaskId(), command, e); - } + AtomicBoolean flag = running.computeIfAbsent(serialNumber, k -> new AtomicBoolean(false)); + if (!flag.compareAndSet(false, true)) { + log.debug("上电触发轮询已在执行中，跳过本次触发，设备编号：{}", serialNumber); + return; } - log.info("设备上电触发任务执行完成，设备编号：{}，总任务数：{}，成功：{}，超时：{}", serialNumber, totalCount, successCount, timeoutCount); + Flux.fromIterable(modbusJobList) + .filter(job -> "0".equals(job.getStatus())) + .concatMap(job -> executeOne(serialNumber, productId, serverType, topic, taskIdsKey, job)) + .collectList() + .doFinally(sig -> flag.set(false)) + .subscribe( + results -> { + long total = results.size(); + long ok = results.stream().filter(Boolean::booleanValue).count(); + long timeout = total - ok; + log.info("设备上电触发任务执行完成，设备编号：{}，总任务数：{}，成功：{}，超时：{}", serialNumber, total, ok, timeout); + }, + err -> log.error("设备上电触发任务执行异常，设备编号：{}", serialNumber, err) + ); + } + + /** + * 单条指令：发送 → 等待应答（200ms）→ 成功则更新任务并从待执行列表删除；失败/超时则保留待执行列表，等待下次上电触发继续执行。 + */ + private Mono executeOne(String serialNumber, + Long productId, + ServerType serverType, + String topic, + String taskIdsKey, + ModbusJob modbusJob) { + String command = modbusJob.getCommand(); + if (StringUtils.isEmpty(command)) { + log.warn("设备上电触发任务执行失败：指令为空，设备编号：{}，任务ID：{}", serialNumber, modbusJob.getTaskId()); + return Mono.just(false); + } + + byte[] commandData; + try { + commandData = ByteUtils.convertToHexByteArray(command); + } catch (Exception e) { + log.warn("设备上电触发任务执行失败：指令转换异常，设备编号：{}，任务ID：{}，指令：{}", serialNumber, modbusJob.getTaskId(), command, e); + return Mono.just(false); + } + if (commandData == null || commandData.length == 0) { + log.warn("设备上电触发任务执行失败：指令转换失败，设备编号：{}，指令：{}", serialNumber, command); + return Mono.just(false); + } + + switch (serverType) { + case MQTT: + return Mono.fromRunnable(() -> { + pubMqttClient.publish(commandData, topic, false, 0); + log.info("=>上电触发-MQTT-线程=[{}],轮询指令:[{}],主题:[{}]", Thread.currentThread().getName(), command, topic); + }) + .then(Mono.delay(Duration.ofMillis(200))) + .thenReturn(true); + + case TCP: + Message msg = new Message(); + msg.setClientId(serialNumber); + msg.setPayload(Unpooled.wrappedBuffer(commandData)); + Session session = sessionStore.getSession(serialNumber); + if (session != null) { + msg.setSerNo(String.valueOf(session.nextSerialNO())); + } + + return messageManager.requestR(serialNumber, msg, Message.class) + .timeout(Duration.ofMillis(20000)) + .map(result -> { + Integer code = result == null ? null : (Integer) result.get(AjaxResult.CODE_TAG); + Object data = result == null ? null : result.get(AjaxResult.DATA_TAG); + boolean ok = code != null && code == HttpStatus.SUCCESS && data instanceof Message; + if (ok) { + log.info("=>上电触发-TCP-线程=[{}],轮询指令:[{}],设备已回复", Thread.currentThread().getName(), command); + updateTaskExecuteTime(modbusJob.getTaskId()); + removeTaskIdFromList(taskIdsKey, modbusJob.getTaskId()); + } else { + log.warn("=>上电触发-TCP-线程=[{}],轮询指令:[{}],设备未回复/响应无效，code:{}, data:{}", + Thread.currentThread().getName(), + command, + code, + data != null ? data.getClass().getSimpleName() : "null"); + } + return ok; + }) + .onErrorResume(e -> { + log.warn("=>上电触发-TCP-线程=[{}],轮询指令:[{}],200ms内未收到应答（保留待执行，等待下次上电触发继续）: {}", + Thread.currentThread().getName(), command, e.getMessage()); + return Mono.just(false); + }); + + default: + log.warn("设备上电触发任务执行失败：不支持的传输协议，设备编号：{}，协议：{}", serialNumber, serverType); + return Mono.just(false); + } } /** 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 dc551cf6..6751aa34 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 @@ -21,6 +21,7 @@ import reactor.core.publisher.MonoSink; import java.net.InetSocketAddress; import java.util.*; import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; @@ -34,8 +35,18 @@ import java.util.function.Function; @Slf4j public class Session { - private static final Mono rejected = Mono.error(new RejectedExecutionException("设备端暂无响应")); - private final Map topicSubscribers = new HashMap<>(); + /** + * requestR() 等待队列。 + * 这里必须保证并发可见性与线程安全：Netty IO线程会调用 response()，而业务线程可能在发起 request()。 + */ + private final Map topicSubscribers = new ConcurrentHashMap<>(); + /** + * 是否存在等待中的 requestR()。 + * 用于入站侧判断是否需要尝试 response 匹配（避免无意义的匹配开销，也方便业务侧做更精细的路由）。 + */ + public boolean hasPendingRequest() { + return !topicSubscribers.isEmpty(); + } protected Channel channel; private boolean udp; private Function remover; @@ -248,19 +259,27 @@ public class Session { * 订阅回调 mono.doOnSuccess(处理成功).doOnError(处理异常).subscribe(开始订阅) */ public Mono request(Message message, Class 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)); + final String key = requestKey(message, responseClass); + + // 关键修复：必须“先注册等待回调，再发送数据” + // 否则设备极快回包时，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)); } /** @@ -293,14 +312,12 @@ public class Session { // 如果还是没有匹配到，尝试匹配所有等待Message.class类型的请求（不依赖serNo） // 找到第一个等待Message.class类型的请求（key以Message.class.getName()开头） // 这样可以匹配到设备回复，即使设备回复没有serNo - synchronized (topicSubscribers) { - for (Map.Entry entry : topicSubscribers.entrySet()) { - String entryKey = entry.getKey(); - // 如果key是Message.class的类名，或者是Message.class的类名+serNo - if (entryKey.equals(messageClassName) || entryKey.startsWith(messageClassName + ".")) { - entry.getValue().success(message); - return true; - } + for (Map.Entry entry : topicSubscribers.entrySet()) { + String entryKey = entry.getKey(); + // 如果key是Message.class的类名，或者是Message.class的类名+serNo + if (entryKey.equals(messageClassName) || entryKey.startsWith(messageClassName + ".")) { + entry.getValue().success(message); + return true; } } } @@ -308,18 +325,6 @@ public class Session { return false; } - /** - * 订阅 - */ - private Mono subscribe(String key) { - synchronized (topicSubscribers) { - if (!topicSubscribers.containsKey(key)) { - return Mono.create(sink -> topicSubscribers.put(key, sink)); - } - } - return null; - } - /** * 取消订阅 */ diff --git a/fastbee-server/boot-strap/src/main/java/com/fastbee/bootstrap/tcp/TCPBootStrap.java b/fastbee-server/boot-strap/src/main/java/com/fastbee/bootstrap/tcp/TCPBootStrap.java index 1ba7bca9..107a0ea7 100644 --- a/fastbee-server/boot-strap/src/main/java/com/fastbee/bootstrap/tcp/TCPBootStrap.java +++ b/fastbee-server/boot-strap/src/main/java/com/fastbee/bootstrap/tcp/TCPBootStrap.java @@ -49,7 +49,7 @@ public class TCPBootStrap { .setIdleStateTime(keepAlive, 0, 0) .setPort(port) //设置报文最大长度 modbus-rtu - .setMaxFrameLength(100) + .setMaxFrameLength(2048) .setDelimiters(new Delimiter(new byte[]{0x7e}, true)) .setDecoder(messageAdapter) .setEncoder(messageAdapter) 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 a1931a6b..44965e17 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 @@ -61,7 +61,7 @@ public class TCPServer extends Server { protected void initChannel(NioSocketChannel channel) throws Exception { channel.pipeline() .addLast(new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, config.allIdleTime)) //设置心跳时间 - // .addLast(FastBeeConstant.SERVER.FRAMEDECODER, frameDecoder())//粘包处理器 + .addLast(FastBeeConstant.SERVER.FRAMEDECODER, frameDecoder())//粘包处理器 .addLast(FastBeeConstant.SERVER.ADAPTER, adapter)//消息适配器 .addLast(FastBeeConstant.SERVER.DECODER, decoder) //报文解码器 .addLast(FastBeeConstant.SERVER.ENCODER, encoder) //报文编码器 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 e68dc6db..218b707e 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 @@ -3,11 +3,13 @@ package com.fastbee.server.handler; import com.fastbee.base.codec.Delimiter; import com.fastbee.base.util.ByteBufUtils; +import com.fastbee.common.utils.gateway.CRC16Utils; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.ByteToMessageDecoder; import io.netty.handler.codec.TooLongFrameException; import io.netty.util.internal.ObjectUtil; +import lombok.extern.slf4j.Slf4j; import java.util.List; @@ -15,9 +17,11 @@ import static io.netty.util.internal.ObjectUtil.checkPositive; /** * 分隔符报文解码器 -消息进站处理步骤1 可选 + * 支持分隔符解码和Modbus RTU帧解码 * * @author bill */ +@Slf4j public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder { /*分隔符 例如报文头 0xFF 报文尾 0x0D*/ @@ -59,7 +63,29 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder { } protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) { - // 使用所有分隔符并选择产生最短帧的分隔符 + // 优先级1：检测7E帧格式（注册包/心跳包） + int frame7EResult = check7EFrame(buffer); + if (frame7EResult > 0) { + // 是7E帧且数据完整，读取并返回 + return buffer.readRetainedSlice(frame7EResult); + } else if (frame7EResult == 0) { + // 是7E帧但数据不完整，等待更多数据，不要用分隔符处理 + return null; + } + + // 优先级2：检测Modbus RTU帧格式（避免数据中的7E被当作分隔符） + int modbusResult = checkModbusRtu(buffer); + if (modbusResult > 0) { + // 是Modbus帧且数据完整，读取并返回 + ByteBuf frame = buffer.readRetainedSlice(modbusResult); + log.debug("成功解析Modbus RTU帧，长度={}", modbusResult); + return frame; + } else if (modbusResult == 0) { + // 是Modbus帧但数据不完整，等待更多数据，不要用分隔符处理 + return null; + } + + // 优先级3：使用分隔符处理（兜底方案，仅当不是7E帧也不是Modbus帧时） int minFrameLength = Integer.MAX_VALUE; Delimiter minDelim = null; for (Delimiter delim : delimiters) { @@ -112,6 +138,7 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder { return frame; } else { + // 没有找到任何可识别的帧格式，等待更多数据或丢弃过长数据 if (!discardingTooLongFrame) { if (buffer.readableBytes() > maxFrameLength) { // Discard the content of the buffer until a delimiter is found. @@ -130,6 +157,166 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder { return null; } } + + /** + * 检测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 check7EFrame(ByteBuf buffer) { + int readable = buffer.readableBytes(); + if (readable < 2) { + return -1; + } + + int readerIndex = buffer.readerIndex(); + int firstByte = buffer.getByte(readerIndex) & 0xFF; + + // 检查是否以7E开头 + if (firstByte != 0x7E) { + return -1; + } + + int secondByte = buffer.getByte(readerIndex + 1) & 0xFF; + + // 注册包(0x80)和心跳包(0x81)都是9字节：7E + 类型 + MAC(6字节) + 7E + if (secondByte != 0x80 && secondByte != 0x81) { + return -1; + } + + int frameLength = 9; + + // 检查数据是否足够 + if (readable < frameLength) { + log.debug("7E帧数据不完整，需要{}字节，当前{}字节，等待更多数据", frameLength, readable); + return 0; // 是7E帧但数据不完整 + } + + // 验证帧尾是否为7E + int lastByte = buffer.getByte(readerIndex + frameLength - 1) & 0xFF; + if (lastByte != 0x7E) { + log.warn("7E帧格式错误，帧尾不是7E，实际值: {}", lastByte); + return -1; + } + + log.debug("成功检测7E帧，类型={}, 长度={}", secondByte == 0x80 ? "注册包" : "心跳包", frameLength); + return frameLength; + } + + /** + * 检测Modbus RTU帧格式 + * @return -1: 不是Modbus帧; 0: 是Modbus帧但数据不完整; >0: 帧长度（数据完整） + */ + private int checkModbusRtu(ByteBuf buffer) { + int readable = buffer.readableBytes(); + if (readable < 2) { + return -1; + } + + int readerIndex = buffer.readerIndex(); + int deviceAddr = buffer.getByte(readerIndex) & 0xFF; + int functionCode = buffer.getByte(readerIndex + 1) & 0xFF; + + // 排除7E开头的数据（可能是7E帧格式，避免误判） + if (deviceAddr == 0x7E) { + return -1; + } + + // 检查设备地址(1-247) + if (deviceAddr < 1 || deviceAddr > 247) { + return -1; + } + + // Modbus RTU：为了避免把请求帧(常见固定8字节)误当响应帧(第3字节是byteCount)导致截断， + // 这里采用“候选长度 + CRC校验”方式确定帧边界，兼容：请求/回显/响应/异常响应。 + + // 异常响应：地址(1) + 功能码(1|0x80) + 异常码(1) + CRC(2) = 5字节 + if ((functionCode & 0x80) != 0) { + int len = 5; + if (readable < len) { + return 0; + } + return isValidModbusCrc(buffer, readerIndex, len) ? len : -1; + } + + // 功能码05/06：请求与响应均为8字节（地址 + 功能码 + 地址(2) + 值(2) + CRC(2)） + if (functionCode == 0x05 || functionCode == 0x06) { + int len = 8; + if (readable < len) { + return 0; + } + return isValidModbusCrc(buffer, readerIndex, len) ? len : -1; + } + + // 功能码01/02/03/04： + // - 请求/回显固定8字节：地址(1)+功能码(1)+起始地址(2)+数量(2)+CRC(2) + // - 响应可变：地址(1)+功能码(1)+byteCount(1)+data(N)+CRC(2) + if (functionCode >= 0x01 && functionCode <= 0x04) { + // 先尝试8字节（避免地址高字节=0x00 被误当 byteCount=0 导致5字节截断） + int reqLen = 8; + if (readable >= reqLen && isValidModbusCrc(buffer, readerIndex, reqLen)) { + return reqLen; + } + + if (readable < 3) { + return 0; + } + int byteCount = buffer.getByte(readerIndex + 2) & 0xFF; + int respLen = 3 + byteCount + 2; + if (respLen > maxFrameLength) { + return -1; + } + if (readable < respLen) { + return 0; + } + return isValidModbusCrc(buffer, readerIndex, respLen) ? respLen : -1; + } + + // 功能码0F/10： + // - 响应固定8字节：地址(1)+功能码(1)+起始地址(2)+数量(2)+CRC(2) + // - 请求可变：地址(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)) { + return respLen; + } + + if (readable < 7) { + return 0; + } + int byteCount = buffer.getByte(readerIndex + 6) & 0xFF; + int reqLen = 9 + byteCount; + if (reqLen > maxFrameLength) { + return -1; + } + if (readable < reqLen) { + return 0; + } + return isValidModbusCrc(buffer, readerIndex, reqLen) ? reqLen : -1; + } + + return -1; // 不是已知的Modbus功能码 + } + + /** + * 校验Modbus RTU CRC16（低位在前，高位在后）。 + * @param buffer ByteBuf + * @param offset 起始偏移（通常为 readerIndex） + * @param len 帧长度（包含CRC2字节） + */ + 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 fail(long frameLength) { if (frameLength > 0) { diff --git a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/DispatcherHandler.java b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/DispatcherHandler.java index 5313ee5d..c89cd0f8 100644 --- a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/DispatcherHandler.java +++ b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/DispatcherHandler.java @@ -137,23 +137,13 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter { private void handleReport(Message request, Session session) { DeviceReport report = (DeviceReport) request; - // 尝试匹配到等待的requestR()调用（无论isReply是否为true） - // 这是因为读操作（Read01、Read03等）的回复可能没有设置isReply=true - // 但仍然是设备对下发指令的回复，需要匹配到等待的请求 - // Session.response()方法已经包含了回退机制，可以匹配到设备回复，即使设备回复没有serNo - if (session != null) { + // 若存在等待中的 requestR()，尝试把当前上行包作为“应答候选”进行匹配。 + // 注意：匹配成功后仍然走正常MQ流程，避免吞掉正常上报（同时 requestR 也能拿到应答）。 + if (session != null && session.hasPendingRequest()) { boolean matched = session.response(report); - if (matched) { - // 匹配成功，requestR()已经收到响应，不再发送到MQ，避免重复处理 - // 注意：这里不发送到MQ，因为requestR()已经处理了 - // 同时，如果原来没有设置isReply=true，现在设置为true，表示这是设备回复 - if (report.getIsReply() == null || !report.getIsReply()) { - report.setIsReply(true); - } - return; + log.debug("{}匹配成功{}，requestR()已收到响应（仍继续发送到MQ）", report.getSerialNumber(), report.getProductId()); } - // 匹配失败，说明不是等待的请求的回复，继续正常流程（发送到MQ） } // 正常流程：发送到MQ 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 3586771f..abce2de1 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 @@ -53,12 +53,10 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter { // 临时记录设备原始数据日志 // 针对特定设备记录原始数据日志 String serialNumber = session.getClientId(); - if ("D3C611176213".equalsIgnoreCase(serialNumber) || "DEE3FBBF0132".equalsIgnoreCase(serialNumber)) { - ByteBuf copy = buf.duplicate(); - byte[] source = new byte[buf.writerIndex()]; - copy.readBytes(source); - log.warn("=>设备[{}]原始数据:{}", serialNumber, ByteBufUtil.hexDump(source)); - } + 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)); diff --git a/fastbee-server/sip-server/src/main/java/com/fastbee/sip/handler/req/RegisterReqHandler.java b/fastbee-server/sip-server/src/main/java/com/fastbee/sip/handler/req/RegisterReqHandler.java index d158a841..d8ee9b4f 100644 --- a/fastbee-server/sip-server/src/main/java/com/fastbee/sip/handler/req/RegisterReqHandler.java +++ b/fastbee-server/sip-server/src/main/java/com/fastbee/sip/handler/req/RegisterReqHandler.java @@ -89,7 +89,7 @@ public class RegisterReqHandler extends ReqAbstractHandler implements Initializi AuthorizationHeader authorhead = (AuthorizationHeader) request.getHeader(AuthorizationHeader.NAME); // 校验密码是否正确 if (authorhead == null && !ObjectUtils.isEmpty(sipConfig.getPassword())) { - log.info("未携带授权头 回复401，sipId：" + sipId); + log.info("未携带授权头 回复401，sipId：{}", sipId); response = getMessageFactory().createResponse(Response.UNAUTHORIZED, request); new DigestAuthUtil().generateChallenge(getHeaderFactory(), response, sipConfig.getDomain()); sipSender.transmitRequest(Sipreq.getLocalAddress().getHostAddress(), response); diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java index 5b1717e5..24fd58cb 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java @@ -547,23 +547,40 @@ public class DeviceServiceImpl implements IDeviceService { for (String refIdentifier : referencedIdentifiers) { if (!refIdentifier.equals(identity)) { // 避免循环引用自己 try { - ValueItem refValueItem = itslValueCache.getCacheIdentifier( - input.getProductId(), - input.getDeviceNumber(), - refIdentifier - ); - if (refValueItem != null && StringUtils.isNotEmpty(refValueItem.getValue())) { - // 使用 {identifier} 作为占位符 - params.put("{" + refIdentifier + "}", refValueItem.getValue()); + // 优先从当前消息中查找 + String refValue = null; + for (ThingsModelSimpleItem currentItem : input.getThingsModelValueRemarkItem()) { + String currentId = currentItem.getId(); + if (currentId.startsWith("array_")) { + currentId = currentId.substring(9); + } + if (currentId.equals(refIdentifier)) { + refValue = currentItem.getValue(); + break; + } + } + + // 如果当前消息中没有，再从缓存读取 + if (refValue == null) { + ValueItem refValueItem = itslValueCache.getCacheIdentifier( + input.getProductId(), + input.getDeviceNumber(), + refIdentifier + ); + if (refValueItem != null && StringUtils.isNotEmpty(refValueItem.getValue())) { + refValue = refValueItem.getValue(); + } + } + + if (refValue != null && !refValue.isEmpty()) { + params.put("{" + refIdentifier + "}", refValue); } else { - // 如果引用的物模型值不存在，使用0作为默认值 params.put("{" + refIdentifier + "}", "0"); log.warn("=>计算公式引用的物模型值不存在[{}],标识符:[{}],使用默认值0", input.getProductId(), refIdentifier); } } catch (Exception e) { log.error("=>获取计算公式引用的物模型值失败[{}],标识符:[{}],错误:{}", input.getProductId(), refIdentifier, e.getMessage()); - params.put("{" + refIdentifier + "}", "0"); } } }