diff --git a/TCP通信流程说明.md b/TCP通信流程说明.md deleted file mode 100644 index 7612780c..00000000 --- a/TCP通信流程说明.md +++ /dev/null @@ -1,185 +0,0 @@ -# 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-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java index 5011158b..321fbec2 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java @@ -137,8 +137,8 @@ public class DevicePropFetchConsumer { List commandList = modbusJobList.stream().map(ModbusJob::getCommand).collect(Collectors.toList()); // 检查是否有设备上电触发任务（cycleType=3） - boolean hasPowerOnTrigger = modbusJobList.stream() - .anyMatch(job -> job.getCycleType() != null && job.getCycleType() == 3); +// boolean hasPowerOnTrigger = modbusJobList.stream() +// .anyMatch(job -> job.getCycleType() != null && job.getCycleType() == 3); // modbus-tcp 重新处理事务标识符 String subSerialNumber = modbusJobList.get(0).getSubSerialNumber(); @@ -147,10 +147,11 @@ public class DevicePropFetchConsumer { List tcpCommandList = new ArrayList<>(); // 如果是设备上电触发任务（cycleType=3），跳过事务ID处理，直接使用原始指令 - if (hasPowerOnTrigger) { - log.info("检测到设备上电触发任务（cycleType=3），跳过事务ID处理，使用原始指令"); - tcpCommandList.addAll(commandList); - } else if (StringUtils.isNotEmpty(tcpObject)) { +// if (hasPowerOnTrigger) { +// log.info("检测到设备上电触发任务（cycleType=3），跳过事务ID处理，使用原始指令"); +// tcpCommandList.addAll(commandList); +// } else + if (StringUtils.isNotEmpty(tcpObject)) { // 原有的事务ID处理逻辑 String tcpRuntimeCacheKey = RedisKeyBuilder.buildModbusTcpRuntimeCacheKey(subSerialNumber); for (String command : commandList) { 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 ff6eaa1a..ccbe0b74 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 @@ -76,59 +76,6 @@ public class PowerOnTriggerTaskExecutor { */ private final ConcurrentHashMap running = new ConcurrentHashMap<>(); - /** - * 标记设备需要触发上电任务（等待第一个数据包） - * 优化：直接查询 ModbusJob，缓存 taskId 列表，避免后续重复查询数据库 - * @param device 设备信息 - */ - @Async(FastBeeConstant.TASK.TRIGGER_POWER_ON_TASK_ASYNC) - public void markPowerOnTriggerPending(Device device) { - try { - String serialNumber = device.getSerialNumber().toUpperCase(); - String taskIdsKey = RedisKeyBuilder.buildPowerOnTriggerJobIdsKey(serialNumber); - String pendingKey = RedisKeyBuilder.buildPowerOnTriggerPendingKey(serialNumber); - - // 直接查询该设备的上电触发任务（cycleType=3，即 remark 包含 power_on_trigger） - ModbusJob queryJob = new ModbusJob(); - queryJob.setSubSerialNumber(serialNumber); - queryJob.setStatus("0"); // 正常状态 - queryJob.setCycleType(3); // 上电触发任务 - - List modbusJobList = modbusJobService.selectModbusJobList(queryJob); - if (CollectionUtils.isEmpty(modbusJobList)) { - // 清除可能存在的缓存 - redisCache.deleteObject(taskIdsKey); - return; - } - - // 筛选出上电触发任务，并收集 taskId 列表 - List powerOnTriggerTaskIds = new ArrayList<>(); - for (ModbusJob job : modbusJobList) { - if (Objects.equals(job.getCycleType(), 3)) { - powerOnTriggerTaskIds.add(job.getTaskId()); - } - } - - // 如果有上电触发任务，缓存 taskId 列表 - if (!powerOnTriggerTaskIds.isEmpty()) { - // 缓存 taskId 列表，过期时间10分钟 - redisCache.setCacheObject(taskIdsKey, JSON.toJSONString(powerOnTriggerTaskIds), 600, TimeUnit.SECONDS); - // 关键：设置"等待第一个数据包"标记（简单的标记即可，比如 "1"） - redisCache.setCacheObject(pendingKey, "1", 600, TimeUnit.SECONDS); - log.debug("标记设备需要触发上电任务，等待第一个数据包，设备编号：{}，任务数量：{}", - serialNumber, powerOnTriggerTaskIds.size()); - } else { - // 如果没有上电触发任务，清除可能存在的旧缓存 - redisCache.deleteObject(taskIdsKey); - redisCache.deleteObject(pendingKey); - log.debug("清除设备上电触发任务标记，设备编号：{}", serialNumber); - - } - } catch (Exception e) { - log.error("标记设备上电触发任务异常，设备编号：{}", device.getSerialNumber(), e); - } - } - /** * 异步检查并触发设备上电任务（在收到第一个数据包时） * 优化：使用 taskId 列表作为标志位，列表为空或不存在表示没有待执行任务 @@ -142,248 +89,40 @@ public class PowerOnTriggerTaskExecutor { return; } - try { - Thread.sleep(1000); // 延迟1秒，等待设备上线 - } catch (InterruptedException e) { - log.error("设备上电触发任务执行失败", e); - } + // 延迟1秒，等待设备上线 + try {Thread.sleep(1000);} catch (InterruptedException e) {log.error("设备上电触发任务执行失败", e);} - String pendingKey = RedisKeyBuilder.buildPowerOnTriggerPendingKey(serialNumber); - // 关键：使用原子删除操作来防止竞态条件 - // 只有第一个成功删除 pendingKey 的线程才能继续执行任务 - // 其他线程删除时会返回 false，直接返回，避免重复执行 - boolean deleted = redisCache.deleteObject(pendingKey); - if (!deleted) { - // 删除失败，说明已经被其他线程处理了，直接返回 - log.debug("设备上电触发任务标记已被其他线程处理，不是第一个数据包，设备编号：{}", serialNumber); + // 查询设备任务 + ModbusJob queryJob = new ModbusJob(); + queryJob.setSubSerialNumber(serialNumber); + queryJob.setStatus("0"); // 正常状态 + queryJob.setCycleType(3); // 设备上电触发类型 + + List modbusJobList = modbusJobService.selectModbusJobList(queryJob); + if (CollectionUtils.isEmpty(modbusJobList)) { + log.debug("设备上电触发任务执行跳过：无任务，设备编号：{}", serialNumber); return; } - // 成功删除，说明是第一个线程，继续执行任务 - - // 补偿查询设备信息（获取产品ID） - if (productId == null) { - Device device = deviceService.selectDeviceBySerialNumber(serialNumber); - if (device == null) { - log.warn("设备编号：{}，产品ID为空，无法触发设备上电任务", serialNumber); - return; - } - productId = device.getProductId(); - } - try { - serialNumber = serialNumber.toUpperCase(); - String taskIdsKey = RedisKeyBuilder.buildPowerOnTriggerJobIdsKey(serialNumber); - - // 从缓存读取 taskId 列表（列表本身作为标志位） - List taskIds = null; - Object cachedTaskIds = redisCache.getCacheObject(taskIdsKey); - if (cachedTaskIds != null) { - try { - taskIds = JSON.parseArray(cachedTaskIds.toString(), Long.class); - // 如果列表为空，表示没有待执行任务，直接返回（天然标志位） - if (CollectionUtils.isEmpty(taskIds)) { - log.debug("设备上电触发任务列表为空，设备编号：{}", serialNumber); - return; - } - log.debug("从缓存读取上电触发任务ID列表，设备编号：{}，任务数量：{}", serialNumber, taskIds.size()); - } catch (Exception e) { - log.warn("解析缓存的任务ID列表失败，设备编号：{}，将查询数据库", serialNumber, e); - } - } else { - // 缓存不存在，表示没有待执行任务，直接返回 - log.debug("设备上电触发任务缓存不存在，设备编号：{}", serialNumber); - return; - } - - // 如果缓存中没有或解析失败，则查询数据库（兼容性处理） - if (taskIds == null || taskIds.isEmpty()) { - ModbusJob queryJob = new ModbusJob(); - queryJob.setSubSerialNumber(serialNumber); - queryJob.setStatus("0"); // 正常状态 - queryJob.setCycleType(3); - - List modbusJobList = modbusJobService.selectModbusJobList(queryJob); - if (CollectionUtils.isEmpty(modbusJobList)) { - return; - } - - // 筛选出上电触发任务的 taskId - taskIds = new ArrayList<>(); - for (ModbusJob job : modbusJobList) { - if (Objects.equals(job.getCycleType(), 3) && "0".equals(job.getStatus())) { - taskIds.add(job.getTaskId()); - } - } - - if (taskIds.isEmpty()) { - return; - } - } - - // 根据 taskId 列表查询 ModbusJob 任务 - List modbusJobList = new ArrayList<>(); - for (Long taskId : taskIds) { - ModbusJob job = modbusJobService.selectModbusJobByTaskId(taskId); - if (job != null && "0".equals(job.getStatus()) && Objects.equals(job.getCycleType(), 3)) { - modbusJobList.add(job); - } - } - - // 执行所有上电触发任务（传入 taskIdsKey，用于成功后删除 taskId） - if (!modbusJobList.isEmpty()) { - execute(serialNumber, productId, modbusJobList, taskIdsKey); - } else { - // 如果没有有效任务，清除缓存 - redisCache.deleteObject(taskIdsKey); - } - - // 注意：pendingKey 已经在方法开始时原子删除了，这里不需要再次删除 - // 如果执行到这里，说明是第一个成功删除 pendingKey 的线程 - log.info("设备上电触发任务处理完成，设备编号：{}", serialNumber); - - - } catch (Exception e) { - log.error("处理设备上电触发任务异常，设备编号：{}", serialNumber, e); - // 异常时也删除标记，避免一直阻塞 + for (ModbusJob modbusJob : modbusJobList) { + String command = modbusJob.getCommand(); + byte[] commandData; try { - redisCache.deleteObject(pendingKey); - } catch (Exception ex) { - // 忽略删除异常 - } - } - } - - /** - * 执行设备上电触发任务 - * 直接下发指令，不通过队列，确保指令及时准确下发 - * 优化：TCP协议使用block()等待响应，确保上一条指令处理完成后再发送下一条 - * 优化：任务执行成功后，从 taskId 列表中删除，列表为空时删除整个 key - * - * @param serialNumber 设备编号 - * @param productId 产品ID - * @param modbusJobList Modbus任务列表 - * @param taskIdsKey Redis key，用于存储待执行的 taskId 列表 - */ - public void execute(String serialNumber, Long productId, List modbusJobList, String taskIdsKey) { - if (StringUtils.isEmpty(serialNumber) || productId == null || CollectionUtils.isEmpty(modbusJobList)) { - log.warn("设备上电触发任务执行失败：设备或任务列表为空"); - return; - } - - // 获取设备状态和传输协议 - DeviceStatusVO deviceStatusVO = deviceService.selectDeviceStatusAndTransportStatus(serialNumber); - if (deviceStatusVO == null) { - log.warn("设备上电触发任务执行失败：无法获取设备状态，设备编号：{}", serialNumber); - return; - } - - ServerType serverType = ServerType.explain(deviceStatusVO.getTransport()); - String topic = topicsUtils.buildTopic(productId, serialNumber, TopicType.FUNCTION_GET); - - // 检查设备是否在线（TCP需要检查session） - if (ServerType.TCP.equals(serverType)) { - Session session = sessionStore.getSession(serialNumber); - if (session == null) { - log.warn("设备上电触发任务执行失败：TCP设备不在线，设备编号：{}", serialNumber); + commandData = ByteUtils.convertToHexByteArray(command); + } catch (Exception e) { + log.warn("设备上电触发任务执行失败：指令转换异常，设备编号：{}，任务ID：{}，指令：{}", serialNumber, modbusJob.getTaskId(), command, e); return; } + 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())); + } + messageManager.requestR(serialNumber, msg, Message.class) + .timeout(Duration.ofMillis(200)); } - AtomicBoolean flag = running.computeIfAbsent(serialNumber, k -> new AtomicBoolean(false)); - if (!flag.compareAndSet(false, true)) { - log.debug("上电触发轮询已在执行中，跳过本次触发，设备编号：{}", serialNumber); - return; - } - - 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(200)) - .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); - } } /** @@ -403,55 +142,5 @@ public class PowerOnTriggerTaskExecutor { modbusJobService.updateModbusJob(job); } - /** - * 从 taskId 列表中删除已成功执行的任务 - * 如果列表为空，则删除整个 key - * - * @param taskIdsKey Redis key - * @param taskId 要删除的任务ID - */ - private void removeTaskIdFromList(String taskIdsKey, Long taskId) { - if (StringUtils.isEmpty(taskIdsKey) || taskId == null) { - return; - } - - try { - // 读取当前列表 - Object cachedTaskIds = redisCache.getCacheObject(taskIdsKey); - if (cachedTaskIds == null) { - log.debug("taskId 列表不存在，key：{}", taskIdsKey); - return; - } - - // 解析为 List - List taskIds = JSON.parseArray(cachedTaskIds.toString(), Long.class); - if (taskIds == null || CollectionUtils.isEmpty(taskIds)) { - // 列表为空或解析失败，删除整个 key - redisCache.deleteObject(taskIdsKey); - log.debug("taskId 列表为空或解析失败，删除 key：{}", taskIdsKey); - return; - } - - // 删除指定的 taskId - boolean removed = taskIds.remove(taskId); - if (!removed) { - log.debug("taskId {} 不在列表中，key：{}", taskId, taskIdsKey); - return; - } - - // 如果列表为空，删除整个 key；否则更新列表 - if (taskIds.isEmpty()) { - redisCache.deleteObject(taskIdsKey); - log.debug("上电触发任务列表已清空，删除 key：{}", taskIdsKey); - } else { - // 更新列表，保持过期时间 - redisCache.setCacheObject(taskIdsKey, JSON.toJSONString(taskIds), 600, TimeUnit.SECONDS); - log.debug("从列表中删除 taskId：{}，剩余任务数：{}", taskId, taskIds.size()); - } - } catch (Exception e) { - log.error("从 taskId 列表中删除任务失败，key：{}，taskId：{}", taskIdsKey, taskId, e); - } - } - } diff --git a/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/codec/ModbusDecoder.java b/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/codec/ModbusDecoder.java index 74b08d93..cc64b888 100644 --- a/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/codec/ModbusDecoder.java +++ b/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/codec/ModbusDecoder.java @@ -47,11 +47,11 @@ public class ModbusDecoder { int messageId = in.getUnsignedByte(1); int currentMessageId = messageId; if (!(start == TcpDtu.起始位 && (messageId == TcpDtu.注册报文 || messageId == TcpDtu.心跳包))) { - //03解析 + // 03解析 currentMessageId = 0; if (deviceData.getCode() == ModbusCode.Write06 || messageId == ModbusCode.Write06.getCode() || deviceData.getCode() == ModbusCode.Write05 || messageId == ModbusCode.Write05.getCode()) { - //设备回复06解析 + // 设备回复06解析 currentMessageId = 2; } else if (deviceData.getCode() == ModbusCode.Read01 || messageId == ModbusCode.Read01.getCode() || deviceData.getCode() == ModbusCode.Read02 || messageId == ModbusCode.Read02.getCode()) { diff --git a/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/test/ModbusTest.java b/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/test/ModbusTest.java index 73e12e61..f3c16e37 100644 --- a/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/test/ModbusTest.java +++ b/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/test/ModbusTest.java @@ -75,6 +75,7 @@ public class ModbusTest { String s = Integer.toHexString(Integer.parseInt("16878")); String s1 = Integer.toHexString(16878); System.out.println(s); + System.out.println(s1); String hexToBytes = "41EE8000"; byte[] bytes = ByteBufUtil.decodeHexDump(hexToBytes); String v = CaculateUtils.toFloat(bytes); 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 3fe22167..e6a913b3 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 @@ -19,7 +19,7 @@ import java.util.concurrent.TimeUnit; /** * TCP服务 - * 粘包处理为 分隔符和固定长度 + * 粘包处理为 多协议嗅探解码器，支持json、ascii、modbus * 需要其他方式处理粘包，按照流程添加 * * @author bill @@ -62,11 +62,9 @@ public class TCPServer extends Server { protected void initChannel(NioSocketChannel channel) throws Exception { channel.pipeline() // 设置心跳时间 - .addLast(new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, - config.allIdleTime)) + .addLast(new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, config.allIdleTime)) // 第一步、添加粘包处理器 - .addLast(FastBeeConstant.SERVER.FRAMEDECODER, - new ProtocolSniffingDecoder(config.maxFrameLength)) + .addLast(FastBeeConstant.SERVER.FRAMEDECODER, new ProtocolSniffingDecoder(config.maxFrameLength)) // 第二步、添加消息适配器 .addLast(FastBeeConstant.SERVER.ADAPTER, adapter) // 第三步、添加报文解码器 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 c89cd0f8..f4eae162 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 @@ -63,7 +63,7 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter { Packet packet = (Packet) msg; Message request = packet.message; - //判断是否注册 + // 判断是否注册 Session session = packet.session; /** * TCP的数据包四种情况 @@ -79,7 +79,7 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter { } //处理数据调试 - //1. 整包上传: 包含：设备编号(注册包)，数据包(心跳包) -->处理注册上传 和 上报数据 + // 1. 整包上传: 包含：设备编号(注册包)，数据包(心跳包) -->处理注册上传 和 上报数据 if (request.getIsPackage()) { //获取设备编号 List items = ((DeviceReport) request).getThingsModelSimpleItem(); @@ -137,15 +137,6 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter { private void handleReport(Message request, Session session) { DeviceReport report = (DeviceReport) request; - // 若存在等待中的 requestR()，尝试把当前上行包作为“应答候选”进行匹配。 - // 注意：匹配成功后仍然走正常MQ流程，避免吞掉正常上报（同时 requestR 也能拿到应答）。 - if (session != null && session.hasPendingRequest()) { - boolean matched = session.response(report); - if (matched) { - log.debug("{}匹配成功{}，requestR()已收到响应（仍继续发送到MQ）", report.getSerialNumber(), report.getProductId()); - } - } - // 正常流程：发送到MQ DeviceReportBo reportBo = new DeviceReportBo(); reportBo.setThingsModelSimpleItem(report.getThingsModelSimpleItem()); @@ -158,7 +149,6 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter { reportBo.setProtocolCode(FastBeeConstant.PROTOCOL.ModbusRtu); reportBo.setSources(report.getSources()); MessageProducer.sendPublishMsg(reportBo); - return; } /** 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 9d09acdb..bbcc5e85 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 @@ -43,7 +43,7 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter { /** * TCP -1 报文入口 - * 设备端消息响应 ，只处理设备session，报文传递到下一个处理器 + * 设备端消息响应 ，处理设备session与帧计数器计数（3帧内无注册包关闭改客户端连接），报文传递到下一个处理器 */ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/domain/ModbusJob.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/domain/ModbusJob.java index dfb46754..67e7c89b 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/domain/ModbusJob.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/domain/ModbusJob.java @@ -73,6 +73,9 @@ public class ModbusJob extends BaseEntity { @ApiModelProperty("设备类型") private Integer deviceType; + /** + * 子设备编号列表 + */ private List subDeviceList; /** 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 f4ebbee3..9faeeaf8 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 @@ -1399,17 +1399,6 @@ public class DeviceServiceImpl implements IDeviceService { } eventLogMapper.insert(event); - // 设备上线时，标记需要触发上电任务（等待第一个数据包） - if (device.getStatus() == 3) { - try { - Object executor = applicationContext.getBean("powerOnTriggerTaskExecutor"); - Method executeMethod = executor.getClass().getMethod("markPowerOnTriggerPending", Device.class); - executeMethod.invoke(executor, device); - } catch (Exception e) { - log.error("标记需要触发上电任务失败，设备编号：{}", device.getSerialNumber(), e); - } - } - return result; } diff --git a/上电触发响应处理优化说明.md b/上电触发响应处理优化说明.md deleted file mode 100644 index 61102d39..00000000 --- a/上电触发响应处理优化说明.md +++ /dev/null @@ -1,163 +0,0 @@ -# 上电触发响应处理优化说明 - -## 优化目标 - -在上电触发场景下，**在协议解码之前拦截设备回复**，直接匹配到等待的`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 deleted file mode 100644 index 8e9d8b97..00000000 --- a/规则引擎原理与操作说明.md +++ /dev/null @@ -1,606 +0,0 @@ -# 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