From 8a87041e7137189e08bab2c0faacb995c5261279 Mon Sep 17 00:00:00 2001 From: George Date: Tue, 30 Dec 2025 08:02:52 +0800 Subject: [PATCH] =?UTF-8?q?20251229:tcp=E9=80=9A=E4=BF=A1=E6=8E=A2?= =?UTF-8?q?=E7=A9=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- TCP通信流程说明.md | 185 ++++++ .../common/constant/FastBeeConstant.java | 1 + .../executor/PowerOnTriggerTaskExecutor.java | 90 +-- .../com/fastbee/base/session/Session.java | 53 +- .../server/handler/DispatcherHandler.java | 27 +- .../iot/service/impl/DeviceServiceImpl.java | 102 +-- sql/add_execute_time_to_modbus_job.sql | 6 + sql/其他/add_cycle_type_to_modbus_job.sql | 17 + 上电触发响应处理优化说明.md | 163 +++++ 规则引擎原理与操作说明.md | 606 ++++++++++++++++++ 视频中心功能说明.md | 495 ++++++++++++++ 运维管理功能说明.md | 425 ++++++++++++ 12 files changed, 2029 insertions(+), 141 deletions(-) create mode 100644 TCP通信流程说明.md create mode 100644 sql/add_execute_time_to_modbus_job.sql create mode 100644 sql/其他/add_cycle_type_to_modbus_job.sql create mode 100644 上电触发响应处理优化说明.md create mode 100644 规则引擎原理与操作说明.md create mode 100644 视频中心功能说明.md create mode 100644 运维管理功能说明.md diff --git a/TCP通信流程说明.md b/TCP通信流程说明.md new file mode 100644 index 00000000..7612780c --- /dev/null +++ b/TCP通信流程说明.md @@ -0,0 +1,185 @@ +# TCP通信流程详解 + +## 核心问题解答 + +**问题：Message对象是设备直接回复的，还是底层框架封装的？** + +**答案：Message对象是底层框架封装的，不是设备直接回复的。** + +设备只发送原始的字节数据（ByteBuf），框架负责： +1. 接收设备的原始字节数据 +2. 通过协议解码器解析为Message对象 +3. 匹配请求和响应 +4. 将响应传递给等待的request + +## 完整通信流程 + +### 1. 平台下发指令（requestR调用） + +``` +PowerOnTriggerTaskExecutor.execute() + ↓ +MessageManager.requestR(serialNumber, msg, Message.class) + ↓ +Session.request(request, Message.class) + ├─ 生成requestKey（基于Message类型和serNo） + ├─ 订阅等待响应（subscribe(key)） + └─ 发送指令到设备（channel.writeAndFlush()） +``` + +**关键代码：** +```java +// Session.request()方法 +public Mono request(Message message, Class responseClass) { + String key = requestKey(message, responseClass); // 生成匹配key + Mono subscribe = this.subscribe(key); // 订阅等待响应 + ChannelFuture channelFuture = channel.writeAndFlush(Packet.of(this, message)); + return Mono.create(...).then(subscribe); // 等待设备回复 +} +``` + +### 2. 设备发送原始数据 + +设备通过TCP连接发送**原始字节数据**（ByteBuf），例如： +- Modbus RTU协议：`7E 01 03 00 00 00 02 C4 0B 7E` +- 设备不知道Message对象，只发送协议规定的字节流 + +### 3. 框架接收和处理（Netty Handler链） + +``` +TCPMessageAdapter.channelRead() + ↓ 接收原始ByteBuf +MessageDecoderWrapper.channelRead() + ↓ 协议解码：ByteBuf → DeviceReport(Message) +DispatcherHandler.channelRead() + ↓ 消息分发处理 +``` + +**关键代码：** +```java +// TCPMessageAdapter - 步骤1：接收原始数据 +public void channelRead(ChannelHandlerContext ctx, Object msg) { + ByteBuf buf = (ByteBuf) msg; // 设备发送的原始字节 + Session session = getSession(ctx); + ctx.fireChannelRead(Packet.of(session, buf)); // 传递给下一个处理器 +} + +// MessageDecoderWrapper - 步骤2：协议解码 +public void channelRead(ChannelHandlerContext ctx, Object msg) { + Packet packet = (Packet) msg; + ByteBuf input = packet.take(); + DeviceReport message = decoder.decode(input, packet.session.getClientId()); + // 协议解码器将ByteBuf解析为DeviceReport对象 + ctx.fireChannelRead(packet.replace(message)); +} + +// DispatcherHandler - 步骤3：消息分发 +public void channelRead(ChannelHandlerContext ctx, Object msg) { + Packet packet = (Packet) msg; + Message request = packet.message; // 已经是Message对象了 + // 处理消息... +} +``` + +### 4. 协议解码器的工作 + +协议解码器（如ModbusProtocol）负责： +1. 解析原始字节数据 +2. 识别是否是设备回复（通过协议特征，如Modbus功能码） +3. 设置isReply标志 +4. 封装为DeviceReport对象 + +**关键代码：** +```java +// ModbusProtocol.decode() +case Write06: +case Write05: + // 如果返回06编码，说明是设备回复 + report.setIsReply(true); // 标记为回复消息 + report.setStatus(FunctionReplyStatus.SUCCESS); + return report; +``` + +### 5. 响应匹配机制（关键！） + +**问题：设备回复如何匹配到等待的request？** + +**答案：通过Session的requestKey和responseKey机制** + +```java +// Session.request() - 发送请求时 +private static String requestKey(Message request, Class responseClass) { + String className = responseClass.getName(); + if (Response.class.isAssignableFrom(responseClass)) { + String serNo = request.getSerNo(); // 使用序列号 + return className + '.' + serNo; + } + return className; // 普通Message使用类名 +} + +// Session.response() - 接收响应时 +private static String responseKey(Object response) { + String className = response.getClass().getName(); + if (response instanceof Response) { + int serialNo = ((Response) response).getResponseSerialNo(); + return className + '.' + serialNo; + } + return className; +} +``` + +**匹配流程：** +1. 发送请求时，生成requestKey并订阅 +2. 设备回复后，协议解码为Message对象 +3. **需要调用Session.response(message)**来匹配 +4. 如果key匹配，Mono会收到响应，requestR返回成功 + +### 6. 当前代码的问题（重要发现！） + +**重要发现：在当前代码中，Session.response()方法没有被调用！** + +这意味着： +- 设备回复的Message对象被解码后，**没有被匹配到等待的request** +- requestR()的Mono会一直等待，直到超时 +- **这就是为什么你的代码中，即使设备回复了，requestR()也可能返回超时** + +**实际流程：** +1. requestR()发送请求，订阅等待响应 +2. 设备回复原始字节数据 +3. 协议解码器解码为DeviceReport对象，设置isReply=true +4. **但是，DispatcherHandler将消息发送到MQ，而不是调用Session.response()** +5. requestR()的Mono一直等待，200ms后超时 + +**关键代码：** +```java +// DispatcherHandler.handleReport() - 设备回复被发送到MQ +private void handleReport(Message request) { + DeviceReport report = (DeviceReport) request; + DeviceReportBo reportBo = new DeviceReportBo(); + // ... + reportBo.setIsReply(report.getIsReply()); // 标记为回复 + MessageProducer.sendPublishMsg(reportBo); // 发送到MQ，而不是Session.response() +} +``` + +**这意味着：** +- requestR()机制在当前实现中**可能无法正确工作** +- 设备回复通过MQ异步处理，而不是同步匹配到request +- 你的代码中检查AjaxResult的code和data是正确的方向 +- 但需要确认：如果设备回复了，requestR()是否能正确返回？ + +## 总结 + +1. **设备只发送原始字节数据**，不知道Message对象 +2. **框架负责解码**：ByteBuf → Message对象 +3. **协议解码器识别回复**：通过协议特征（如Modbus功能码）设置isReply标志 +4. **响应匹配机制**：通过requestKey/responseKey匹配请求和响应 +5. **关键问题**：需要确认Session.response()是否被正确调用 + +## 建议 + +对于上电触发任务，如果Session.response()没有被调用，那么： +- requestR()可能无法正确识别设备回复 +- 需要检查是否有其他地方调用了Session.response() +- 或者需要修改代码，在协议解码后主动调用Session.response() + diff --git a/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java b/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java index de3b46ba..ec6d5b7f 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java +++ b/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java @@ -26,6 +26,7 @@ public interface FastBeeConstant { String WS_PREFIX = "web-"; String WM_PREFIX = "server-"; String FAST_PHONE = "phone-"; + String POWER = "powerOnTriggerHandler"; /*MQTT平台判定离线时间 keepAlive*1.5 */ Long DEVICE_PING_EXPIRED = 90000L; 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 b535943b..e9d2028c 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 @@ -9,6 +9,8 @@ import com.fastbee.common.core.redis.RedisCache; import com.fastbee.common.core.redis.RedisKeyBuilder; import com.fastbee.common.enums.ServerType; import com.fastbee.common.enums.TopicType; +import com.fastbee.common.constant.HttpStatus; +import com.fastbee.common.core.domain.AjaxResult; import com.fastbee.common.utils.DateUtils; import com.fastbee.common.utils.StringUtils; import com.fastbee.common.utils.gateway.mq.TopicsUtils; @@ -31,10 +33,8 @@ import javax.annotation.Resource; import java.time.Duration; import java.util.ArrayList; import java.util.Date; -import java.util.HashSet; import java.util.List; import java.util.Objects; -import java.util.Set; import java.util.concurrent.TimeUnit; /** @@ -135,13 +135,16 @@ public class PowerOnTriggerTaskExecutor { } String pendingKey = RedisKeyBuilder.buildPowerOnTriggerPendingKey(serialNumber); - // 关键：检查是否存在"等待第一个数据包"标记 - Object pendingFlag = redisCache.getCacheObject(pendingKey); - if (pendingFlag == null) { - // 标记不存在，说明不是第一个数据包，或者已经处理过了 - log.debug("设备上电触发任务标记不存在，不是第一个数据包，设备编号：{}", serialNumber); + // 关键：使用原子删除操作来防止竞态条件 + // 只有第一个成功删除 pendingKey 的线程才能继续执行任务 + // 其他线程删除时会返回 false，直接返回，避免重复执行 + boolean deleted = redisCache.deleteObject(pendingKey); + if (!deleted) { + // 删除失败，说明已经被其他线程处理了，直接返回 + log.debug("设备上电触发任务标记已被其他线程处理，不是第一个数据包，设备编号：{}", serialNumber); return; } + // 成功删除，说明是第一个线程，继续执行任务 // 补偿查询设备信息（获取产品ID） if (productId == null) { @@ -219,10 +222,9 @@ public class PowerOnTriggerTaskExecutor { redisCache.deleteObject(taskIdsKey); } - // 关键：无论任务是否成功，都删除"等待第一个数据包"标记 - // 这样后续数据包就不会再触发任务了 - redisCache.deleteObject(pendingKey); - log.info("设备上电触发任务处理完成，已删除第一个数据包标记，设备编号：{}", serialNumber); + // 注意：pendingKey 已经在方法开始时原子删除了，这里不需要再次删除 + // 如果执行到这里，说明是第一个成功删除 pendingKey 的线程 + log.info("设备上电触发任务处理完成，设备编号：{}", serialNumber); } catch (Exception e) { @@ -272,11 +274,6 @@ public class PowerOnTriggerTaskExecutor { } } - // 记录设备上电触发任务下发的指令片段（索引2到3位，2个字符） - String cacheKey = RedisKeyBuilder.buildPowerOnTriggerCommandKey(serialNumber); - // 收集所有指令片段（用于支持多个指令的情况） - Set commandSegments = new HashSet<>(); - // 执行所有相关的modbus任务 int successCount = 0; int timeoutCount = 0; @@ -296,14 +293,6 @@ public class PowerOnTriggerTaskExecutor { } try { - // 记录下发的指令片段（索引2到3位，即第3到第4个字符，2个字符） - String cleanCommand = command.replaceAll("\\s+", "").toUpperCase(); - if (cleanCommand.length() >= 4) { - String commandSegment = cleanCommand.substring(2, 4); - commandSegments.add(commandSegment); - log.debug("收集设备上电触发任务指令片段，设备编号：{}，指令片段：{}", serialNumber, commandSegment); - } - // 转换指令为字节数组 byte[] commandData = ByteUtils.convertToHexByteArray(command); if (commandData.length == 0) { @@ -339,19 +328,50 @@ public class PowerOnTriggerTaskExecutor { 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 { - messageManager.requestR(serialNumber, msg, Message.class) - .timeout(Duration.ofMillis(20000)) // 200ms超时 - .doOnSuccess(result -> log.info("=>上电触发-TCP-线程=[{}],轮询指令:[{}],响应成功", Thread.currentThread().getName(), command)) + // 等待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 - responseReceived = true; - log.info("=>上电触发-TCP-线程=[{}],轮询指令:[{}],已收到响应", Thread.currentThread().getName(), command); + + // 检查返回结果：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内未收到响应，视为超时 + // 200ms内未收到响应或发生异常，视为超时 long elapsed = System.currentTimeMillis() - sendTime; - log.warn("=>上电触发-TCP-线程=[{}],轮询指令:[{}],200ms内未收到响应，耗时:{}ms", Thread.currentThread().getName(), command, elapsed); + log.warn("=>上电触发-TCP-线程=[{}],轮询指令:[{}],200ms内未收到响应，耗时:{}ms，异常:{}", Thread.currentThread().getName(), command, elapsed, e.getMessage()); responseReceived = false; + // 超时后的回复会被DispatcherHandler识别为isReply=true，但匹配失败（订阅已取消） + // 会继续发送到MQ，但不会误判为普通数据包，因为isReply=true } break; default: @@ -376,14 +396,6 @@ public class PowerOnTriggerTaskExecutor { } } - // 所有指令下发完成后，缓存指令片段集合（用于识别回包） - if (!commandSegments.isEmpty()) { - // 将指令片段集合转换为JSON字符串存储，过期时间5秒（足够设备响应） - String segmentsJson = JSON.toJSONString(new ArrayList<>(commandSegments)); - redisCache.setCacheObject(cacheKey, segmentsJson, 5, TimeUnit.SECONDS); - log.debug("缓存设备上电触发任务指令片段集合，设备编号：{}，片段数量：{}，片段：{}", serialNumber, commandSegments.size(), commandSegments); - } - log.info("设备上电触发任务执行完成，设备编号：{}，总任务数：{}，成功：{}，超时：{}", serialNumber, totalCount, successCount, timeoutCount); } 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 e05231b2..dc551cf6 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 @@ -123,6 +123,11 @@ public class Session { String serNo = request.getSerNo(); return new StringBuffer(34).append(className).append('.').append(serNo).toString(); } + // 对于Message.class类型，如果请求有serNo，也使用serNo来生成唯一key + // 这样可以支持多个并发请求使用相同的响应类型 + if (request != null && request.getSerNo() != null && !request.getSerNo().isEmpty()) { + return new StringBuffer(34).append(className).append('.').append(request.getSerNo()).toString(); + } return className; } @@ -133,6 +138,19 @@ public class Session { int serialNo = ((Response) response).getResponseSerialNo(); return new StringBuffer(34).append(className).append('.').append(serialNo).toString(); } + // 对于Message类型（如DeviceReport），如果响应有serNo，也使用serNo来匹配 + // 这样可以匹配到带serNo的请求 + if (response instanceof Message) { + Message message = (Message) response; + if (message.getSerNo() != null && !message.getSerNo().isEmpty()) { + // 关键：如果响应是Message的子类（如DeviceReport），但请求类型是Message.class + // 需要同时尝试使用Message.class的类名来匹配，因为requestKey使用的是Message.class.getName() + String messageClassName = Message.class.getName(); + // 优先使用Message.class的类名匹配（用于匹配Message.class类型的请求） + // 注意：这里返回Message.class的key，因为requestR()使用的是Message.class + return new StringBuffer(34).append(messageClassName).append('.').append(message.getSerNo()).toString(); + } + } return className; } @@ -249,11 +267,44 @@ public class Session { * 消息响应 */ public boolean response(Message message) { - MonoSink monoSink = topicSubscribers.get(responseKey(message)); + // 先尝试精确匹配 + String key = responseKey(message); + MonoSink monoSink = topicSubscribers.get(key); if (monoSink != null) { monoSink.success(message); return true; } + + // 如果精确匹配失败，且响应是Message类型，尝试匹配所有等待Message.class类型的请求 + // 这是因为设备回复可能没有serNo，或者类名不匹配（如DeviceReport vs Message） + if (message instanceof Message) { + String messageClassName = Message.class.getName(); + + // 如果响应有serNo，尝试使用Message.class的类名匹配 + if (message.getSerNo() != null && !message.getSerNo().isEmpty()) { + String messageKey = messageClassName + "." + message.getSerNo(); + monoSink = topicSubscribers.get(messageKey); + if (monoSink != null) { + monoSink.success(message); + return true; + } + } + + // 如果还是没有匹配到，尝试匹配所有等待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; + } + } + } + } + return false; } 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 bad6cacb..5313ee5d 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 @@ -75,7 +75,7 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter { */ // 4.单个数据包上传无标识位 if (null == request.getMessageId()) { - this.handleReport(request); + this.handleReport(request, session); } //处理数据调试 @@ -100,7 +100,7 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter { this.handleMessage(request, packet, ctx); } //处理消息转发 - this.handleReport(request); + this.handleReport(request, session); // 2/3 单个注册包，心跳，数据包上传，匹配消息id处理 } else if (null != request.getMessageId() && !"0".equals(request.getMessageId())) { @@ -134,8 +134,29 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter { } } - private void handleReport(Message request) { + private void handleReport(Message request, Session session) { DeviceReport report = (DeviceReport) request; + + // 尝试匹配到等待的requestR()调用（无论isReply是否为true） + // 这是因为读操作（Read01、Read03等）的回复可能没有设置isReply=true + // 但仍然是设备对下发指令的回复，需要匹配到等待的请求 + // Session.response()方法已经包含了回退机制，可以匹配到设备回复，即使设备回复没有serNo + if (session != null) { + boolean matched = session.response(report); + + if (matched) { + // 匹配成功，requestR()已经收到响应，不再发送到MQ，避免重复处理 + // 注意：这里不发送到MQ，因为requestR()已经处理了 + // 同时，如果原来没有设置isReply=true，现在设置为true，表示这是设备回复 + if (report.getIsReply() == null || !report.getIsReply()) { + report.setIsReply(true); + } + return; + } + // 匹配失败，说明不是等待的请求的回复，继续正常流程（发送到MQ） + } + + // 正常流程：发送到MQ DeviceReportBo reportBo = new DeviceReportBo(); reportBo.setThingsModelSimpleItem(report.getThingsModelSimpleItem()); reportBo.setPlatformDate(DateUtils.getNowDate()); 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 4dcc0921..13d27ae0 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 @@ -687,10 +687,8 @@ public class DeviceServiceImpl implements IDeviceService { } // 缓存最新一条数据到redis redisCache.hashPutAll(key, maps); - // 判断是否是设备上电触发的回包，如果是则跳过保存历史数据 - boolean isPowerOnTriggerReply = isPowerOnTriggerReply(input.getSources(), input.getDeviceNumber()); //存储历史数据 - if (!CollectionUtils.isEmpty(deviceLogList) && !isShadow && !isPowerOnTriggerReply) { + if (!CollectionUtils.isEmpty(deviceLogList) && !isShadow) { for (DeviceLog deviceLog : deviceLogList) { logService.saveDeviceLog(deviceLog); } @@ -722,7 +720,7 @@ public class DeviceServiceImpl implements IDeviceService { Map calculatedMaps = new HashMap<>(); List calculatedDeviceLogList = new ArrayList<>(); - calculateDependentThingsModels(input, updatedIdentifiers, key, calculatedMaps, calculatedDeviceLogList, type, isPowerOnTriggerReply); + calculateDependentThingsModels(input, updatedIdentifiers, key, calculatedMaps, calculatedDeviceLogList, type); // 将计算后的值更新到缓存 if (!calculatedMaps.isEmpty()) { @@ -732,7 +730,7 @@ public class DeviceServiceImpl implements IDeviceService { } // 保存计算型物模型的历史数据 - if (!CollectionUtils.isEmpty(calculatedDeviceLogList) && !isPowerOnTriggerReply) { + if (!CollectionUtils.isEmpty(calculatedDeviceLogList)) { for (DeviceLog deviceLog : calculatedDeviceLogList) { logService.saveDeviceLog(deviceLog); } @@ -1942,96 +1940,6 @@ public class DeviceServiceImpl implements IDeviceService { return deviceMapper.selectSerialNumbersByProductId(productId); } - /** - * 判断是否是设备上电触发的回包 - * 从Redis缓存中获取下发的指令片段（索引2到3位，2个字符），与上报的指令片段进行比较 - * - * @param sources 原始消息数据包 - * @param deviceNumber 设备编号 - * @return true表示是设备上电触发的回包 - */ - private boolean isPowerOnTriggerReply(Object sources, String deviceNumber) { - if (sources == null || StringUtils.isEmpty(deviceNumber)) { - return false; - } - - try { - // 从Redis缓存中获取下发的指令片段（可能是单个片段或片段集合） - String serialNumber = deviceNumber.toUpperCase(); - String cacheKey = RedisKeyBuilder.buildPowerOnTriggerCommandKey(serialNumber); - Object cachedSegments = redisCache.getCacheObject(cacheKey); - - // 如果缓存中没有下发的指令片段，说明不是设备上电触发的回包 - if (cachedSegments == null) { - return false; - } - - // 解析指令片段（可能是单个字符串或JSON数组） - Set sentCommandSegments = new HashSet<>(); - try { - if (cachedSegments instanceof String) { - String segmentsStr = (String) cachedSegments; - // 尝试解析为JSON数组 - if (segmentsStr.startsWith("[")) { - List segmentsList = JSON.parseArray(segmentsStr, String.class); - sentCommandSegments.addAll(segmentsList); - } else { - // 单个片段（兼容旧数据） - sentCommandSegments.add(segmentsStr); - } - } - } catch (Exception e) { - log.warn("解析指令片段失败，设备编号：{}", serialNumber, e); - return false; - } - - if (sentCommandSegments.isEmpty()) { - return false; - } - - String hexStr = null; - - // 将 sources 转换为十六进制字符串 - if (sources instanceof String) { - hexStr = (String) sources; - // 如果是字符串，可能需要去除空格 - hexStr = hexStr.replaceAll("\\s+", "").toUpperCase(); - } else if (sources instanceof byte[]) { - // 如果是字节数组，转换为十六进制字符串 - byte[] bytes = (byte[]) sources; - StringBuilder sb = new StringBuilder(); - for (byte b : bytes) { - sb.append(String.format("%02X", b & 0xFF)); - } - hexStr = sb.toString(); - } else { - // 其他类型，尝试转换为字符串 - hexStr = sources.toString().replaceAll("\\s+", "").toUpperCase(); - } - - // 至少需要4个字符才能提取索引2到3位 - if (hexStr.length() < 4) { - return false; - } - - // 提取上报指令的索引2到3位（即第3到第4个字符，2个字符） - String actualSegment = hexStr.substring(2, 4); - - // 与下发的指令片段集合进行比较（支持多个指令） - boolean isMatch = sentCommandSegments.contains(actualSegment); - - if (isMatch) { - // 匹配成功，删除缓存（避免重复匹配） - redisCache.deleteObject(cacheKey); - log.info("检测到设备上电触发回包，设备编号：{}，指令片段：{}", serialNumber, actualSegment); - } - - return isMatch; - } catch (Exception e) { - log.warn("判断设备上电触发回包时发生异常: {}", e.getMessage()); - return false; - } - } /** * 自动计算依赖已更新物模型的其他物模型（纯计算型物模型） @@ -2043,15 +1951,13 @@ public class DeviceServiceImpl implements IDeviceService { * @param maps 物模型值映射（用于批量更新缓存） * @param deviceLogList 设备日志列表（用于存储历史数据） * @param type 物模型类型（1=属性，2=功能，3=事件） - * @param isPowerOnTriggerReply 是否是设备上电触发回包 */ private void calculateDependentThingsModels(ThingsModelValuesInput input, Set updatedIdentifiers, String key, Map maps, List deviceLogList, - int type, - boolean isPowerOnTriggerReply) { + int type) { try { // 查询产品下所有物模型 com.fastbee.iot.domain.ThingsModel queryModel = new com.fastbee.iot.domain.ThingsModel(); diff --git a/sql/add_execute_time_to_modbus_job.sql b/sql/add_execute_time_to_modbus_job.sql new file mode 100644 index 00000000..62fab32b --- /dev/null +++ b/sql/add_execute_time_to_modbus_job.sql @@ -0,0 +1,6 @@ +-- 为iot_modbus_job表添加execute_time字段 +-- 用于记录设备上电触发任务的执行时间 + +ALTER TABLE `iot_modbus_job` +ADD COLUMN `execute_time` datetime NULL DEFAULT NULL COMMENT '执行时间（仅设备上电触发任务使用，记录任务执行的时间）' AFTER `remark`; + diff --git a/sql/其他/add_cycle_type_to_modbus_job.sql b/sql/其他/add_cycle_type_to_modbus_job.sql new file mode 100644 index 00000000..a860cfbf --- /dev/null +++ b/sql/其他/add_cycle_type_to_modbus_job.sql @@ -0,0 +1,17 @@ +-- 为 iot_modbus_job 表添加 cycle_type 字段 +-- 时间周期类型（1-周期循环，2-自定义时间段，3-设备上电触发） + +ALTER TABLE `iot_modbus_job` + ADD COLUMN `cycle_type` int(2) NULL DEFAULT NULL COMMENT '时间周期类型（1-周期循环，2-自定义时间段，3-设备上电触发）' AFTER `device_type`; + +-- 迁移设备上电触发任务：将 remark 字段中包含 'power_on_trigger' 的任务设置为 cycle_type = 3 +-- remark 字段保持不变，因为代码中还在使用它来识别设备上电触发任务 +UPDATE `iot_modbus_job` +SET `cycle_type` = 3 +WHERE `remark` IS NOT NULL AND `remark` LIKE '%power_on_trigger%'; + +-- 为其他现有数据设置默认值（默认为1-周期循环） +UPDATE `iot_modbus_job` +SET `cycle_type` = 1 +WHERE `cycle_type` IS NULL; + diff --git a/上电触发响应处理优化说明.md b/上电触发响应处理优化说明.md new file mode 100644 index 00000000..61102d39 --- /dev/null +++ b/上电触发响应处理优化说明.md @@ -0,0 +1,163 @@ +# 上电触发响应处理优化说明 + +## 优化目标 + +在上电触发场景下，**在协议解码之前拦截设备回复**，直接匹配到等待的`requestR()`调用，避免走完整的协议解码和MQ流程，减少通信负担，提高响应速度。 + +## 实现方案 + +### 1. 新增Handler：PowerOnTriggerResponseHandler + +**位置：** `fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/PowerOnTriggerResponseHandler.java` + +**功能：** +- 在协议解码之前拦截设备回复的原始ByteBuf数据 +- 创建简单的Message对象（不解析内容） +- 调用`Session.response()`尝试匹配到等待的`requestR()`调用 +- 如果匹配成功，`requestR()`会立即返回，无需等待协议解码 + +### 2. Handler链顺序 + +``` +TCPMessageAdapter (接收原始ByteBuf) + ↓ +PowerOnTriggerResponseHandler (拦截并匹配响应) ← 新增 + ↓ +MessageDecoderWrapper (协议解码) + ↓ +DispatcherHandler (消息分发) +``` + +### 3. 工作原理 + +``` +1. 平台下发指令 + PowerOnTriggerTaskExecutor.execute() + → MessageManager.requestR() + → Session.request() 订阅等待响应 + +2. 设备回复原始字节数据 + 设备发送：ByteBuf (例如：7E 01 03 00 00 00 02 C4 0B 7E) + +3. PowerOnTriggerResponseHandler拦截 + - 接收原始ByteBuf + - 创建简单的Message对象（不解析内容） + - 调用Session.response()尝试匹配 + +4. 匹配成功 + - Session.response()找到等待的request + - 立即唤醒等待的Mono + - requestR()返回成功（code=200, data=Message对象） + - PowerOnTriggerTaskExecutor收到响应，更新任务状态 + +5. 继续传递（可选） + - 如果继续传递，协议解码也会处理（但上电触发任务已完成） + - 如果不想走协议解码，可以注释掉ctx.fireChannelRead(msg) +``` + +## 优势 + +1. **减少通信负担** + - 不需要走完整的协议解码流程 + - 不需要发送到MQ异步处理 + - 直接同步匹配，响应更快 + +2. **提高精准度** + - 直接匹配等待的request，不依赖协议特征识别 + - 避免协议解码误判导致的问题 + +3. **降低延迟** + - 200ms内设备回复，立即匹配成功 + - 无需等待协议解码和MQ处理 + +## 代码修改 + +### 1. PowerOnTriggerResponseHandler.java + +```java +@Override +public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + Packet packet = (Packet) msg; + Session session = packet.session; + ByteBuf byteBuf = packet.byteBuf; + + // 创建简单的Message对象，只用于匹配响应 + Message responseMessage = new Message(); + responseMessage.setClientId(session.getClientId()); + responseMessage.setPayload(byteBuf.retain()); + + // 尝试匹配到等待的request + boolean matched = session.response(responseMessage); + + if (matched) { + log.debug("=>上电触发-设备回复已匹配，设备编号：{}", session.getClientId()); + // 匹配成功，继续传递（可选） + ctx.fireChannelRead(msg); + } else { + // 没有匹配到，继续正常流程 + responseMessage.getPayload().release(); + ctx.fireChannelRead(msg); + } +} +``` + +### 2. TCPServer.java + +```java +channel.pipeline() + .addLast(FastBeeConstant.SERVER.ADAPTER, adapter) + .addLast("powerOnTriggerHandler", powerOnTriggerHandler) // 新增 + .addLast(FastBeeConstant.SERVER.DECODER, decoder) + // ... +``` + +## 使用说明 + +### 当前实现 + +- **匹配成功**：`requestR()`立即返回，`PowerOnTriggerTaskExecutor`收到响应 +- **继续传递**：消息继续传递给协议解码器（可选，如果不想走协议解码可以注释掉） + +### 如果不想走协议解码 + +如果完全不想走协议解码，可以修改`PowerOnTriggerResponseHandler`： + +```java +if (matched) { + log.debug("=>上电触发-设备回复已匹配，设备编号：{}", session.getClientId()); + // 匹配成功，直接消费掉，不走协议解码 + byteBuf.release(); // 释放ByteBuf + return; // 不继续传递 +} +``` + +### 注意事项 + +1. **ByteBuf引用计数** + - 使用`byteBuf.retain()`增加引用计数 + - 如果匹配失败，需要`release()`释放 + - 如果匹配成功，后续处理器会负责释放 + +2. **兼容性** + - 如果匹配失败，继续走正常流程（协议解码等） + - 不影响其他功能 + +3. **性能** + - 只在上电触发场景下生效 + - 其他场景继续走正常流程 + +## 测试建议 + +1. **验证匹配成功** + - 下发上电触发指令 + - 设备回复后，检查`requestR()`是否立即返回 + - 检查`PowerOnTriggerTaskExecutor`是否正确更新任务状态 + +2. **验证性能** + - 对比优化前后的响应时间 + - 检查是否减少了MQ消息量 + +3. **验证兼容性** + - 确保其他功能不受影响 + - 确保协议解码仍然正常工作 + diff --git a/规则引擎原理与操作说明.md b/规则引擎原理与操作说明.md new file mode 100644 index 00000000..8e9d8b97 --- /dev/null +++ b/规则引擎原理与操作说明.md @@ -0,0 +1,606 @@ +# FastBee IoT平台 - 规则引擎原理与操作说明 + +## 一、规则引擎概述 + +**规则引擎**是FastBee IoT平台的核心组件，用于处理设备数据、实现场景联动、数据转发等功能。它基于**LiteFlow**框架，支持Groovy脚本，通过EL表达式构建规则链，实现灵活的业务逻辑处理。 + +### 核心特性 +- **基于LiteFlow**: 轻量级规则引擎框架 +- **Groovy脚本**: 支持动态脚本执行 +- **EL表达式**: 使用表达式语言构建规则链 +- **异步执行**: 非阻塞式规则执行 +- **动态构建**: 运行时动态创建规则链和脚本节点 + +## 二、规则引擎工作原理 + +### 2.1 整体架构 + +``` +设备上报数据 + ↓ +数据处理器 (DataHandler) + ↓ +规则引擎处理器 (RuleEngineHandler) + ↓ +查询关联规则 (Script/Scene) + ↓ +构建规则链 (LiteFlow) + ↓ +执行规则脚本 (Groovy) + ↓ +执行动作 (设备控制/告警/转发等) +``` + +### 2.2 核心组件 + +#### 1. FlowLogExecutor（规则执行器） +- **位置**: `fastbee-plugs/fastbee-ruleEngine/src/main/java/com/fastbee/ruleEngine/core/FlowLogExecutor.java` +- **作用**: 封装LiteFlow执行器，提供日志记录功能 +- **功能**: + - 执行规则链 + - 记录执行日志 + - 处理执行结果 + +#### 2. SceneContext（场景上下文） +- **位置**: `fastbee-iot-data/src/main/java/com/fastbee/iot/data/ruleEngine/SceneContext.java` +- **作用**: 场景联动的执行上下文 +- **包含数据**: + - 设备编号 + - 产品ID + - 数据类型（属性/功能/事件） + - 物模型数据列表 + +#### 3. MsgContext（消息上下文） +- **位置**: `fastbee-plugs/fastbee-ruleEngine/src/main/java/com/fastbee/ruleEngine/context/MsgContext.java` +- **作用**: 规则脚本的消息上下文 +- **包含数据**: + - 主题（topic） + - 负载（payload） + - 设备编号 + - 产品ID + - 协议编码 + +### 2.3 规则链构建原理 + +#### EL表达式语法 + +规则引擎使用EL（Expression Language）表达式来构建规则链，支持以下语法： + +1. **IF-THEN结构** + ``` + IF(条件, THEN(动作1, 动作2)) + ``` + +2. **逻辑运算符** + - `AND(条件1, 条件2, ...)`: 所有条件都满足 + - `OR(条件1, 条件2, ...)`: 任一条件满足 + - `NOT(条件)`: 条件不满足 + +3. **执行模式** + - `THEN(动作1, 动作2, ...)`: 串行执行（顺序执行） + - `WHEN(动作1, 动作2, ...)`: 并行执行（同时执行） + +#### 规则链示例 + +**场景联动规则链**: +``` +IF(AND(T1, T2, T3), THEN(A1, A2, A3)) +``` +- `T1, T2, T3`: 触发器脚本（条件判断） +- `A1, A2, A3`: 执行动作脚本（执行操作） + +**数据流规则链**: +``` +THEN(script1, script2, httpBridge) +``` +- `script1, script2`: 数据处理脚本 +- `httpBridge`: HTTP桥接节点 + +### 2.4 脚本类型 + +#### 1. 脚本用途（scriptPurpose） +- `1`: 数据流 - 处理设备上报数据 +- `2`: 触发器 - 条件判断脚本（返回boolean） +- `3`: 执行动作 - 执行操作脚本 + +#### 2. 脚本类型（scriptType） +- `script`: 普通脚本 +- `boolean_script`: 条件脚本（返回boolean值） +- `switch_script`: 选择脚本 +- `for_script`: 循环脚本 + +#### 3. 脚本事件（scriptEvent） +- `1`: 设备上报 +- `2`: 平台下发 +- `3`: 设备上线 +- `4`: 设备离线 +- `5`: HTTP桥接数据 +- `6`: MQTT桥接数据 + +#### 4. 脚本动作（scriptAction） +- `1`: 消息重发 +- `2`: 消息通知 +- `3`: HTTP推送 +- `4`: MQTT桥接 +- `5`: 数据库存储 + +### 2.5 执行流程详解 + +#### 场景联动执行流程 + +``` +1. 设备上报数据 + ↓ +2. RuleEngineHandler.ruleMatch() 接收数据 + ↓ +3. 查询设备关联的场景 + sceneDeviceService.selectTriggerDeviceRelateScenes() + ↓ +4. 遍历场景列表 + ↓ +5. 创建SceneContext上下文 + ↓ +6. 执行规则链 + flowExecutor.execute2FutureWithRid(chainName, context) + ↓ +7. LiteFlow解析EL表达式 + 例如: IF(AND(T1, T2), THEN(A1, A2)) + ↓ +8. 执行触发器脚本（T1, T2） + - 检查静默时间 + - 匹配设备数据 + - 判断条件是否满足 + ↓ +9. 如果条件满足，执行动作脚本（A1, A2） + - 设备控制 + - 告警推送 + - 场景恢复等 +``` + +#### 规则脚本执行流程 + +``` +1. 设备上报数据 + ↓ +2. RuleProcess.processRuleScript() 处理 + ↓ +3. 查询匹配的规则脚本 + selectExecRuleScriptList(scriptCondition) + ↓ +4. 构建规则链 + LiteFlowChainELBuilder.createChain() + ↓ +5. 执行脚本 + flowExecutor.execute2RespWithRid() + ↓ +6. 执行脚本动作 + - HTTP推送 + - MQTT桥接 + - 数据库存储等 +``` + +## 三、规则引擎操作指南 + +### 3.1 规则脚本管理 + +**功能位置**: 规则引擎 → 规则脚本 +**路径**: `/iot/scene/script` +**权限**: `iot:script:list` + +#### 创建规则脚本 + +1. **进入规则脚本页面** + - 点击"新增"按钮 + +2. **配置脚本基本信息** + - **脚本名称**: 脚本的显示名称 + - **脚本标识**: 唯一标识（字母开头，可包含数字） + - **执行顺序**: 值越大优先级越高 + - **脚本事件**: 选择触发事件（设备上报/平台下发/设备上线/设备离线等） + - **脚本动作**: 选择执行动作（消息重发/消息通知/HTTP推送/MQTT桥接/数据库存储） + - **脚本状态**: 启用/禁用 + +3. **选择关联对象** + - **产品**: 选择关联的产品（设备上报/平台下发时） + - **桥接**: 选择HTTP/MQTT桥接（桥接数据时） + +4. **编写脚本代码** + - 使用Groovy语言编写 + - 脚本中可以使用上下文变量 + - 支持脚本验证 + +5. **脚本验证** + - 点击"验证"按钮检查脚本语法 + - 验证通过后才能保存 + +6. **保存脚本** + - 点击"确定"保存脚本 + - 系统会自动构建规则链 + +#### 脚本编写示例 + +**数据流脚本示例**: +```groovy +// 获取消息上下文 +def msgContext = _meta.getContextBean("com.fastbee.ruleEngine.context.MsgContext") + +// 获取设备编号 +def serialNumber = msgContext.getSerialNumber() + +// 获取产品ID +def productId = msgContext.getProductId() + +// 获取消息负载 +def payload = msgContext.getPayload() + +// 处理数据逻辑 +// ... + +// 设置处理后的数据 +msgContext.setPayload(processedData) +``` + +**条件脚本示例**: +```groovy +// 获取场景上下文 +def sceneContext = _meta.getContextBean("com.fastbee.iot.data.ruleEngine.SceneContext") + +// 处理规则脚本 +String json = "..."; // JSON格式的脚本模板 +return sceneContext.process(json); +``` + +#### 查看脚本日志 + +1. 在脚本列表中点击"日志"按钮 +2. 查看脚本执行日志 +3. 日志包含： + - 执行时间 + - 上下文数据 + - 执行步骤 + - 错误信息 + +### 3.2 场景联动中的规则引擎 + +场景联动会自动使用规则引擎，用户无需直接操作规则引擎，但需要了解其工作原理： + +#### 场景创建时的规则构建 + +1. **创建场景时** + - 系统自动为每个触发器创建规则脚本（T开头） + - 系统自动为每个执行动作创建规则脚本（A开头） + - 自动生成EL表达式 + +2. **规则链构建** + - 根据执行条件（或/且/非）构建IF表达式 + - 根据执行方式（串行/并行）构建THEN/WHEN表达式 + - 生成完整的规则链名称（chainName） + +3. **动态加载** + - 规则链和脚本节点动态构建到内存 + - 无需重启服务即可生效 + +#### EL表达式生成规则 + +**执行条件为"且"（所有条件满足）**: +``` +IF(AND(T1, T2, T3), THEN(A1, A2)) +``` + +**执行条件为"或"（任一条件满足）**: +``` +IF(OR(T1, T2, T3), THEN(A1, A2)) +``` + +**执行条件为"非"（条件不满足）**: +``` +IF(NOT(T1), THEN(A1, A2)) +``` + +**执行方式为并行**: +``` +IF(AND(T1, T2), WHEN(A1, A2)) +``` + +### 3.3 规则脚本使用场景 + +#### 场景1: 数据转发 + +**需求**: 将设备上报的数据转发到第三方系统 + +**操作步骤**: +1. 创建规则脚本 +2. 脚本事件选择"设备上报" +3. 脚本动作选择"HTTP推送"或"MQTT桥接" +4. 编写脚本处理数据 +5. 配置桥接信息 +6. 保存并启用脚本 + +#### 场景2: 数据转换 + +**需求**: 转换设备数据格式 + +**操作步骤**: +1. 创建规则脚本 +2. 脚本事件选择"设备上报" +3. 脚本动作选择"消息重发" +4. 编写脚本转换数据格式 +5. 保存并启用脚本 + +#### 场景3: 数据存储 + +**需求**: 将设备数据存储到数据库 + +**操作步骤**: +1. 创建规则脚本 +2. 脚本事件选择"设备上报" +3. 脚本动作选择"数据库存储" +4. 编写脚本处理数据 +5. 保存并启用脚本 + +## 四、规则引擎核心代码 + +### 4.1 规则执行器 + +```java +// FlowLogExecutor.java +@Component +public class FlowLogExecutor { + @Resource + private FlowExecutor flowExecutor; + + // 执行规则链（带请求ID） + public LiteflowResponse execute2FutureWithRid( + String chainId, + Object param, + String requestId, + Object... contextBeanArray + ) { + // 打印上下文 + printContextBeanWithRid(requestId, contextBeanArray); + // 执行规则链 + Future future = + flowExecutor.execute2FutureWithRid(chainId, param, requestId, contextBeanArray); + LiteflowResponse response = future.get(); + // 打印执行结果 + printResponseWithRid(requestId, response); + return response; + } +} +``` + +### 4.2 规则链构建 + +```java +// SceneServiceImpl.java +private void dynamicBuildRule(List