20260104:

1、对粘包拆包进行探索

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

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<String, AtomicBoolean> running = new ConcurrentHashMap<>();
+
     /**
      * 标记设备需要触发上电任务（等待第一个数据包）
      * 优化：直接查询 ModbusJob，缓存 taskId 列表，避免后续重复查询数据库
@@ -279,127 +288,104 @@ public class PowerOnTriggerTaskExecutor {
             }
         }
 
-        // 执行所有相关的modbus任务
-        int successCount = 0;
-        int timeoutCount = 0;
-        int totalCount = 0;
-        
-        for (ModbusJob modbusJob : modbusJobList) {
-            if (!"0".equals(modbusJob.getStatus())) {
-                // 只执行正常状态的任务
-                continue;
+        AtomicBoolean flag = running.computeIfAbsent(serialNumber, k -> new AtomicBoolean(false));
+        if (!flag.compareAndSet(false, true)) {
+            log.debug("上电触发轮询已在执行中，跳过本次触发，设备编号：{}", serialNumber);
+            return;
         }
 
-            totalCount++;
+        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<Boolean> 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());
-                continue;
+            return Mono.just(false);
         }
 
+        byte[] commandData;
         try {
-                // 转换指令为字节数组
-                byte[] commandData = ByteUtils.convertToHexByteArray(command);
-                if (commandData.length == 0) {
+            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);
-                    continue;
+            return Mono.just(false);
         }
 
-                // 记录发送时间
-                long sendTime = System.currentTimeMillis();
-                boolean responseReceived;
-                
-                // 根据传输协议下发指令
         switch (serverType) {
             case MQTT:
-                        // 通过MQTT内部客户端下发指令
+                return Mono.fromRunnable(() -> {
                             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;
+                        })
+                        .then(Mono.delay(Duration.ofMillis(200)))
+                        .thenReturn(true);
+
             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;
+                    msg.setSerNo(String.valueOf(session.nextSerialNO()));
                 }
 
-                // 只有在200ms内收到响应时才更新任务执行时间和从列表中删除 taskId
-                if (responseReceived) {
-                    successCount++;
-                    // 更新任务执行时间
+                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());
-                    // 从 taskId 列表中删除已成功执行的任务
                                 removeTaskIdFromList(taskIdsKey, modbusJob.getTaskId());
                             } else {
-                    timeoutCount++;
-                    log.warn("设备上电触发任务超时，不更新任务状态，设备编号：{}，任务ID：{}", serialNumber, modbusJob.getTaskId());
+                                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);
+                        });
 
-            } catch (Exception e) {
-                log.error("设备上电触发任务执行异常，设备编号：{}，任务ID：{}，指令：{}", serialNumber, modbusJob.getTaskId(), command, e);
+            default:
+                log.warn("设备上电触发任务执行失败：不支持的传输协议，设备编号：{}，协议：{}", serialNumber, serverType);
+                return Mono.just(false);
         }
     }
 
-        log.info("设备上电触发任务执行完成，设备编号：{}，总任务数：{}，成功：{}，超时：{}", serialNumber, totalCount, successCount, timeoutCount);
-    }
-
     /**
      * 更新任务执行时间（仅在收到响应时调用）
      * 

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<String, MonoSink> topicSubscribers = new HashMap<>();
+    /**
+     * requestR() 等待队列。
+     * 这里必须保证并发可见性与线程安全：Netty IO线程会调用 response()，而业务线程可能在发起 request()。
+     */
+    private final Map<String, MonoSink> topicSubscribers = new ConcurrentHashMap<>();
+    /**
+     * 是否存在等待中的 requestR()。
+     * 用于入站侧判断是否需要尝试 response 匹配（避免无意义的匹配开销，也方便业务侧做更精细的路由）。
+     */
+    public boolean hasPendingRequest() {
+        return !topicSubscribers.isEmpty();
+    }
     protected Channel channel;
     private boolean udp;
     private Function<Session, Boolean> remover;
@@ -248,19 +259,27 @@ public class Session {
      * 订阅回调 mono.doOnSuccess(处理成功).doOnError(处理异常).subscribe(开始订阅)
      */
     public <T> Mono<T> request(Message message, Class<T> responseClass) {
-        String key = requestKey(message, responseClass);
-        Mono<T> subscribe = this.subscribe(key);
-        if (subscribe == null) {
-            return rejected;
+        final String key = requestKey(message, responseClass);
+
+        // 关键修复：必须“先注册等待回调，再发送数据”
+        // 否则设备极快回包时，response() 可能发生在订阅登记之前，导致误判为普通上报并落库。
+        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));
-        return Mono.create(sink -> channelFuture.addListener(future -> {
-            if (future.isSuccess()) {
-                sink.success(future);
-            } else {
+                    channelFuture.addListener(future -> {
+                        if (!future.isSuccess()) {
+                            // 发送失败，移除订阅并返回错误
+                            topicSubscribers.remove(key);
                             sink.error(future.cause());
                         }
-        })).then(subscribe).doFinally(signal -> unsubscribe(key));
+                    });
+                })
+                .doFinally(signal -> unsubscribe(key));
     }
 
     /**
@@ -293,7 +312,6 @@ public class Session {
             // 如果还是没有匹配到，尝试匹配所有等待Message.class类型的请求（不依赖serNo）
             // 找到第一个等待Message.class类型的请求（key以Message.class.getName()开头）
             // 这样可以匹配到设备回复，即使设备回复没有serNo
-            synchronized (topicSubscribers) {
             for (Map.Entry<String, MonoSink> entry : topicSubscribers.entrySet()) {
                 String entryKey = entry.getKey();
                 // 如果key是Message.class的类名，或者是Message.class的类名+serNo
@@ -303,23 +321,10 @@ 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;
-    }
-
     /**
      * 取消订阅
      */

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)

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) //报文编码器

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.
@@ -131,6 +158,166 @@ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
         }
     }
     
+    /**
+     * 检测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) {
             throw new TooLongFrameException("frame length exceeds " + maxFrameLength + ": " + frameLength + " - discarded");

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);
+                log.debug("{}匹配成功{}，requestR()已收到响应（仍继续发送到MQ）", report.getSerialNumber(), report.getProductId());
             }
-                return;
-            }
-            // 匹配失败，说明不是等待的请求的回复，继续正常流程（发送到MQ）
         }
         
         // 正常流程：发送到MQ

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));
-        }
+        log.warn("==设备[{}]原始数据:{}", serialNumber, ByteBufUtil.hexDump(source));
         session.access();
         /*将设备端消息传递给下一个处理器*/
         ctx.fireChannelRead(Packet.of(session, buf));

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);

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 {
+                            // 优先从当前消息中查找
+                            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())) {
-                                // 使用 {identifier} 作为占位符
-                                params.put("{" + refIdentifier + "}", 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");
                         }
                     }
                 }