From b35a7b81625a5c0d469fcb3a5982affe27b5b3a1 Mon Sep 17 00:00:00 2001 From: George Date: Mon, 29 Dec 2025 08:19:29 +0800 Subject: [PATCH] =?UTF-8?q?20251229:=E8=AE=BE=E5=A4=87=E4=B8=8A=E7=94=B5?= =?UTF-8?q?=E8=A7=A6=E5=8F=91=E4=BB=BB=E5=8A=A1=E4=BF=AE=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/main/resources/application-prod.yml | 2 +- .../common/constant/FastBeeConstant.java | 5 + .../common/core/redis/RedisKeyBuilder.java | 9 +- .../iot/data/aspect/DeviceDebugLogAspect.java | 17 +- .../executor/PowerOnTriggerTaskExecutor.java | 458 ++++++++++++++++++ .../service/IDeviceReportMessageService.java | 6 - .../impl/DeviceReportMessageServiceImpl.java | 51 +- .../server/handler/MessageDecoderWrapper.java | 2 +- .../fastbee/iot/service/IDeviceService.java | 12 - .../iot/service/impl/DeviceServiceImpl.java | 243 ++-------- .../service/impl/ModbusJobServiceImpl.java | 18 + .../resources/mapper/iot/ModbusJobMapper.xml | 7 +- sql/yack.sql | 2 + vue/src/views/dataCenter/history.vue | 32 +- 14 files changed, 588 insertions(+), 276 deletions(-) create mode 100644 fastbee-iot-data/src/main/java/com/fastbee/iot/data/executor/PowerOnTriggerTaskExecutor.java diff --git a/fastbee-admin/src/main/resources/application-prod.yml b/fastbee-admin/src/main/resources/application-prod.yml index e85c8171..e831169c 100644 --- a/fastbee-admin/src/main/resources/application-prod.yml +++ b/fastbee-admin/src/main/resources/application-prod.yml @@ -141,7 +141,7 @@ sip: # 日志配置 logging: level: - com.fastbee: debug + com.fastbee: info com.yomahub: warn org.dromara: warn org.springframework: warn 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 b7bcc242..de3b46ba 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 @@ -287,6 +287,11 @@ public interface FastBeeConstant { */ String POWER_ON_TRIGGER_PENDING = "power:on:trigger:pending:"; + /** + * 设备上电触发任务ID列表缓存（用于避免重复查询数据库） + */ + String POWER_ON_TRIGGER_JOB_IDS = "power:on:trigger:jobIds:"; + } interface PROTOCOL { diff --git a/fastbee-common/src/main/java/com/fastbee/common/core/redis/RedisKeyBuilder.java b/fastbee-common/src/main/java/com/fastbee/common/core/redis/RedisKeyBuilder.java index a56f4b25..8981507a 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/core/redis/RedisKeyBuilder.java +++ b/fastbee-common/src/main/java/com/fastbee/common/core/redis/RedisKeyBuilder.java @@ -160,7 +160,7 @@ public class RedisKeyBuilder { /** * 设备上电触发指令缓存key - * 用于存储设备上电触发任务下发的指令片段（索引2到11位） + * 用于存储设备上电触发任务下发的指令片段（索引2到3位） */ public static String buildPowerOnTriggerCommandKey(String serialNumber) { return FastBeeConstant.REDIS.POWER_ON_TRIGGER_COMMAND + serialNumber.toUpperCase(); @@ -174,5 +174,12 @@ public class RedisKeyBuilder { return FastBeeConstant.REDIS.POWER_ON_TRIGGER_PENDING + serialNumber.toUpperCase(); } + /** + * 设备上电触发任务ID列表缓存key + * 用于缓存设备上电触发任务的ID列表，避免重复查询数据库 + */ + public static String buildPowerOnTriggerJobIdsKey(String serialNumber) { + return FastBeeConstant.REDIS.POWER_ON_TRIGGER_JOB_IDS + serialNumber.toUpperCase(); + } } diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/aspect/DeviceDebugLogAspect.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/aspect/DeviceDebugLogAspect.java index d3ab9686..79890733 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/aspect/DeviceDebugLogAspect.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/aspect/DeviceDebugLogAspect.java @@ -37,6 +37,7 @@ import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.Objects; +import java.util.function.Function; /** * 设备调试错误日志AOP切面 @@ -232,7 +233,7 @@ public class DeviceDebugLogAspect { * 统一异常处理入口 */ private void handleException(JoinPoint joinPoint, Throwable e, - java.util.function.Function contextExtractor, + Function contextExtractor, String errorLogPrefix) { try { Object[] args = joinPoint.getArgs(); @@ -378,9 +379,11 @@ public class DeviceDebugLogAspect { // 构建并保存设备日志 DeviceLog deviceLog = buildDeviceLog(device, context); logService.saveDeviceLog(deviceLog); - log.info("已保存错误日志到数据库 - 设备: {}, 错误类型: {}, 错误信息: {}", - serialNumber, context.getIdentity(), - StringUtils.substring(context.getErrorMsg(), 0, MAX_ERROR_MSG_DISPLAY_LENGTH)); + if (log.isDebugEnabled()) { + log.debug("已保存错误日志到数据库 - 设备: {}, 错误类型: {}, 错误信息: {}", + serialNumber, context.getIdentity(), + StringUtils.substring(context.getErrorMsg(), 0, MAX_ERROR_MSG_DISPLAY_LENGTH)); + } // 推送错误信息到前端页面 pushErrorToFrontend(serialNumber, productId, context); @@ -462,8 +465,10 @@ public class DeviceDebugLogAspect { // 推送消息到MQTT mqttRemoteManager.pushCommon(messageBo); - log.info("已推送错误信息到前端 - 设备: {}, 产品ID: {}, 错误类型: {}", - serialNumber, productId, context.getIdentity()); + if (log.isDebugEnabled()) { + log.debug("已推送错误信息到前端 - 设备: {}, 产品ID: {}, 错误类型: {}", + serialNumber, productId, context.getIdentity()); + } } catch (Exception ex) { log.error("推送错误信息到前端失败: {}", ex.getMessage(), ex); } 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 new file mode 100644 index 00000000..b535943b --- /dev/null +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/executor/PowerOnTriggerTaskExecutor.java @@ -0,0 +1,458 @@ +package com.fastbee.iot.data.executor; + +import com.alibaba.fastjson2.JSON; +import com.fastbee.base.service.ISessionStore; +import com.fastbee.base.session.Session; +import com.fastbee.common.constant.FastBeeConstant; +import com.fastbee.common.core.protocol.Message; +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.utils.DateUtils; +import com.fastbee.common.utils.StringUtils; +import com.fastbee.common.utils.gateway.mq.TopicsUtils; +import com.fastbee.common.utils.gateway.protocol.ByteUtils; +import com.fastbee.iot.data.service.impl.MessageManager; +import com.fastbee.iot.domain.Device; +import com.fastbee.iot.domain.ModbusJob; +import com.fastbee.iot.model.DeviceStatusVO; +import com.fastbee.iot.service.IDeviceService; +import com.fastbee.iot.service.IModbusJobService; +import com.fastbee.mqttclient.PubMqttClient; +import io.netty.buffer.Unpooled; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.scheduling.annotation.Async; +import org.springframework.stereotype.Component; +import org.springframework.util.CollectionUtils; + +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; + +/** + * 设备上电触发任务执行器 + * 专门用于执行设备上电触发任务，直接下发指令，不通过队列，确保指令及时准确下发 + * + * @author Sumuxi + */ +@Slf4j +@Component +public class PowerOnTriggerTaskExecutor { + + @Autowired + private PubMqttClient pubMqttClient; + + @Autowired + private RedisCache redisCache; + + @Resource + private MessageManager messageManager; + + @Resource + private TopicsUtils topicsUtils; + + @Resource + private IDeviceService deviceService; + + @Resource + private ISessionStore sessionStore; + + @Resource + private IModbusJobService modbusJobService; + + /** + * 标记设备需要触发上电任务（等待第一个数据包） + * 优化：直接查询 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 列表作为标志位，列表为空或不存在表示没有待执行任务 + * @param serialNumber 设备编号 + * @param productId 产品ID + */ + @Async(FastBeeConstant.TASK.TRIGGER_POWER_ON_TASK_ASYNC) + public void checkAndTriggerPowerOnTaskAsync(String serialNumber, Long productId) { + if (StringUtils.isEmpty(serialNumber)) { + log.warn("设备编号为空，无法触发设备上电任务"); + return; + } + + String pendingKey = RedisKeyBuilder.buildPowerOnTriggerPendingKey(serialNumber); + // 关键：检查是否存在"等待第一个数据包"标记 + Object pendingFlag = redisCache.getCacheObject(pendingKey); + if (pendingFlag == null) { + // 标记不存在，说明不是第一个数据包，或者已经处理过了 + 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); + } + + // 关键：无论任务是否成功，都删除"等待第一个数据包"标记 + // 这样后续数据包就不会再触发任务了 + redisCache.deleteObject(pendingKey); + log.info("设备上电触发任务处理完成，已删除第一个数据包标记，设备编号：{}", serialNumber); + + + } catch (Exception e) { + log.error("处理设备上电触发任务异常，设备编号：{}", serialNumber, e); + // 异常时也删除标记，避免一直阻塞 + 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); + return; + } + } + + // 记录设备上电触发任务下发的指令片段（索引2到3位，2个字符） + String cacheKey = RedisKeyBuilder.buildPowerOnTriggerCommandKey(serialNumber); + // 收集所有指令片段（用于支持多个指令的情况） + Set commandSegments = new HashSet<>(); + + // 执行所有相关的modbus任务 + int successCount = 0; + int timeoutCount = 0; + int totalCount = 0; + + for (ModbusJob modbusJob : modbusJobList) { + if (!"0".equals(modbusJob.getStatus())) { + // 只执行正常状态的任务 + continue; + } + + totalCount++; + String command = modbusJob.getCommand(); + if (StringUtils.isEmpty(command)) { + log.warn("设备上电触发任务执行失败：指令为空，设备编号：{}，任务ID：{}", serialNumber, modbusJob.getTaskId()); + continue; + } + + try { + // 记录下发的指令片段（索引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) { + log.warn("设备上电触发任务执行失败：指令转换失败，设备编号：{}，指令：{}", serialNumber, command); + continue; + } + + // 记录指令到Redis缓存（用于轮询任务管理） + String runtimeCacheKey = RedisKeyBuilder.buildModbusRuntimeCacheKey(serialNumber); + redisCache.zSetAdd(runtimeCacheKey, command, DateUtils.getTimestampSeconds()); + + // 记录发送时间 + long sendTime = System.currentTimeMillis(); + boolean responseReceived; + + // 根据传输协议下发指令 + switch (serverType) { + case MQTT: + // 通过MQTT内部客户端下发指令 + pubMqttClient.publish(commandData, topic, false, 0); + log.info("=>上电触发-MQTT-线程=[{}],轮询指令:[{}],主题:[{}]", Thread.currentThread().getName(), command, topic); + // MQTT协议无法直接等待响应，等待200ms后继续 + try { + Thread.sleep(200); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.warn("MQTT指令等待被中断，设备编号：{}", serialNumber); + } + responseReceived = true; // MQTT协议假设发送成功 + break; + case TCP: + // 下发TCP客户端，等待200ms内的响应 + Message msg = new Message(); + msg.setClientId(serialNumber); + msg.setPayload(Unpooled.wrappedBuffer(commandData)); + try { + messageManager.requestR(serialNumber, msg, Message.class) + .timeout(Duration.ofMillis(20000)) // 200ms超时 + .doOnSuccess(result -> log.info("=>上电触发-TCP-线程=[{}],轮询指令:[{}],响应成功", Thread.currentThread().getName(), command)) + .doOnError(error -> log.warn("=>上电触发-TCP-线程=[{}],轮询指令:[{}],响应超时或失败:{}", Thread.currentThread().getName(), command, error.getMessage())) + .block(); // 阻塞等待响应，最多200ms + responseReceived = true; + log.info("=>上电触发-TCP-线程=[{}],轮询指令:[{}],已收到响应", Thread.currentThread().getName(), command); + } catch (Exception e) { + // 200ms内未收到响应，视为超时 + long elapsed = System.currentTimeMillis() - sendTime; + log.warn("=>上电触发-TCP-线程=[{}],轮询指令:[{}],200ms内未收到响应，耗时:{}ms", Thread.currentThread().getName(), command, elapsed); + responseReceived = false; + } + break; + default: + log.warn("设备上电触发任务执行失败：不支持的传输协议，设备编号：{}，协议：{}", serialNumber, serverType); + continue; + } + + // 只有在200ms内收到响应时才更新任务执行时间和从列表中删除 taskId + if (responseReceived) { + successCount++; + // 更新任务执行时间 + updateTaskExecuteTime(modbusJob.getTaskId()); + // 从 taskId 列表中删除已成功执行的任务 + removeTaskIdFromList(taskIdsKey, modbusJob.getTaskId()); + } else { + timeoutCount++; + log.warn("设备上电触发任务超时，不更新任务状态，设备编号：{}，任务ID：{}", serialNumber, modbusJob.getTaskId()); + } + + } catch (Exception e) { + log.error("设备上电触发任务执行异常，设备编号：{}，任务ID：{}，指令：{}", serialNumber, modbusJob.getTaskId(), command, e); + } + } + + // 所有指令下发完成后，缓存指令片段集合（用于识别回包） + 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); + } + + /** + * 更新任务执行时间（仅在收到响应时调用） + * + * @param taskId 任务ID + */ + private void updateTaskExecuteTime(Long taskId) { + if (taskId == null) { + return; + } + + ModbusJob job = new ModbusJob(); + job.setTaskId(taskId); + job.setExecuteTime(new Date()); + job.setStatus("1"); + 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-iot-data/src/main/java/com/fastbee/iot/data/service/IDeviceReportMessageService.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/IDeviceReportMessageService.java index bf31c8a3..63029181 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/IDeviceReportMessageService.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/IDeviceReportMessageService.java @@ -47,10 +47,4 @@ public interface IDeviceReportMessageService { */ IProtocol selectedProtocol(Long productId); - /** - * 异步检查并触发设备上电任务（在收到第一个数据包时） - * @param serialNumber 设备编号 - */ - void checkAndTriggerPowerOnTaskAsync(String serialNumber); - } diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceReportMessageServiceImpl.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceReportMessageServiceImpl.java index ebb9614d..78947714 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceReportMessageServiceImpl.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceReportMessageServiceImpl.java @@ -19,6 +19,7 @@ import com.fastbee.common.utils.DateUtils; import com.fastbee.common.utils.StringUtils; import com.fastbee.common.utils.gateway.mq.TopicsUtils; import com.fastbee.iot.cache.IOtaTaskCache; +import com.fastbee.iot.data.executor.PowerOnTriggerTaskExecutor; import com.fastbee.iot.data.service.IDataHandler; import com.fastbee.iot.data.service.IDeviceReportMessageService; import com.fastbee.iot.data.service.IFirmwareTaskDetailService; @@ -83,9 +84,9 @@ public class DeviceReportMessageServiceImpl implements IDeviceReportMessageServi private TopicsUtils topicsUtils; @Resource private MqttRemoteManager remoteManager; - + @Resource - private ApplicationContext applicationContext; // self注入，用于异步调用 + private PowerOnTriggerTaskExecutor powerOnTriggerTaskExecutor; /** * 处理设备主动上报数据 @@ -93,17 +94,16 @@ public class DeviceReportMessageServiceImpl implements IDeviceReportMessageServi @Override public void parseReportMsg(DeviceReportBo bo) { String serialNumber = bo.getSerialNumber(); - + Long productId = bo.getProductId(); + // 异步检查是否需要触发上电任务（收到第一个数据包时触发） - DeviceReportMessageServiceImpl bean = applicationContext.getBean(DeviceReportMessageServiceImpl.class); - bean.checkAndTriggerPowerOnTaskAsync(serialNumber); - + powerOnTriggerTaskExecutor.checkAndTriggerPowerOnTaskAsync(serialNumber, productId); + switch (bo.getServerType()) { case MQTT: log.debug("=>MQ*收到设备主题[{}],消息:[{}]", bo.getTopicName(), bo.getData()); //构建消息 DeviceStatusVO deviceStatusVO = buildReport(bo); - Long productId = bo.getProductId(); /*获取协议处理器*/ IProtocol protocol = selectedProtocol(productId); DeviceData data = DeviceData.builder() @@ -245,43 +245,6 @@ public class DeviceReportMessageServiceImpl implements IDeviceReportMessageServi } } - /** - * 异步检查并触发设备上电任务（在收到第一个数据包时） - * @param serialNumber 设备编号 - */ - @Async(FastBeeConstant.TASK.TRIGGER_POWER_ON_TASK_ASYNC) - public void checkAndTriggerPowerOnTaskAsync(String serialNumber) { - try { - if (StringUtils.isEmpty(serialNumber)) { - return; - } - - String cacheKey = RedisKeyBuilder.buildPowerOnTriggerPendingKey(serialNumber.toUpperCase()); - // 检查是否有待触发的上电任务标记 - Object pendingFlag = redisCache.getCacheObject(cacheKey); - if (pendingFlag == null) { - // 没有待触发的任务，直接返回 - return; - } - - // 有待触发的任务，删除标记（确保只触发一次） - redisCache.deleteObject(cacheKey); - - // 获取设备信息 - Device device = deviceService.selectDeviceBySerialNumber(serialNumber); - if (device == null) { - log.warn("触发上电任务时设备不存在，设备编号：{}", serialNumber); - return; - } - - // 触发上电任务（异步执行） - log.info("收到设备第一个数据包，触发上电任务，设备编号：{}", serialNumber); - deviceService.triggerPowerOnTasks(device); - - } catch (Exception e) { - log.error("检查并触发设备上电任务异常，设备编号：{}", serialNumber, e); - } } - /** * 构建消息 * diff --git a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/MessageDecoderWrapper.java b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/MessageDecoderWrapper.java index 069fb17e..6d72b03c 100644 --- a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/MessageDecoderWrapper.java +++ b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/MessageDecoderWrapper.java @@ -44,7 +44,7 @@ public class MessageDecoderWrapper extends ChannelInboundHandlerAdapter { /*将当前readerIndex增加指定的长度，已经解码完成*/ input.skipBytes(input.readableBytes()); } catch (Exception e) { - log.error("消息解码异常[" + ByteBufUtil.hexDump(input, 0, input.writerIndex()) + "]", e); + log.error("消息解码异常[{}]", ByteBufUtil.hexDump(input, 0, input.writerIndex()), e); throw new DecoderException(e); } finally { input.release(); diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IDeviceService.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IDeviceService.java index b6f5b78d..66476543 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IDeviceService.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IDeviceService.java @@ -185,18 +185,6 @@ public interface IDeviceService { */ public int updateDeviceStatusAndLocation(Device device, String ipAddress); - /** - * 标记设备上电任务待处理 - * @param device 设备信息 - */ - void markPowerOnTriggerPending(Device device); - - /** - * 触发设备上电任务（供外部调用，如收到第一个数据包时） - * @param device 设备信息 - */ - public void triggerPowerOnTasks(Device device); - /** * 更新设备状态 * 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 f31e41ae..4dcc0921 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 @@ -149,9 +149,6 @@ public class DeviceServiceImpl implements IDeviceService { @Resource private IModbusJobService modbusJobService; @Resource - @Lazy - private IDeviceJobService deviceJobService; - @Resource private ProductSubGatewayMapper productSubGatewayMapper; @Resource private ModbusParamsMapper modbusParamsMapper; @@ -1373,201 +1370,18 @@ public class DeviceServiceImpl implements IDeviceService { // 设备上线时，标记需要触发上电任务（等待第一个数据包） if (device.getStatus() == 3) { - DeviceServiceImpl bean = applicationContext.getBean(DeviceServiceImpl.class); - bean.markPowerOnTriggerPending(device); + 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; } - /** - * 标记设备需要触发上电任务（等待第一个数据包） - * @param device 设备信息 - */ - @Async(FastBeeConstant.TASK.TRIGGER_POWER_ON_TASK_ASYNC) - @Override - public void markPowerOnTriggerPending(Device device) { - try { - // 查询该设备的所有设备上电触发任务 - DeviceJob queryJob = new DeviceJob(); - queryJob.setDeviceId(device.getDeviceId()); - queryJob.setJobType(5); // modbus轮询任务 - queryJob.setStatus("0"); // 正常状态 - - List jobList = deviceJobService.selectJobList(queryJob); - if (CollectionUtils.isEmpty(jobList)) { - return; - } - - // 检查是否有上电触发任务 - boolean hasPowerOnTrigger = false; - for (DeviceJob job : jobList) { - if (isPowerOnTrigger(job.getRemark())) { - hasPowerOnTrigger = true; - break; - } - } - - // 如果有上电触发任务，则在Redis中标记，等待第一个数据包 - // 无论标记是否存在，都重新设置标记，确保恢复任务后能重新触发 - String serialNumber = device.getSerialNumber().toUpperCase(); - String cacheKey = RedisKeyBuilder.buildPowerOnTriggerPendingKey(serialNumber); - if (hasPowerOnTrigger) { - // 设置过期时间为10分钟，防止长时间未收到数据包导致标记一直存在 - // 无论之前是否有标记，都重新设置，确保恢复任务后能重新触发 - redisCache.setCacheObject(cacheKey, "1", 600, TimeUnit.SECONDS); - log.info("标记设备需要触发上电任务，等待第一个数据包，设备编号：{}", serialNumber); - } else { - // 如果没有上电触发任务，清除可能存在的旧标记 - if (redisCache.hasKey(cacheKey)) { - redisCache.deleteObject(cacheKey); - log.debug("清除设备上电触发任务标记，设备编号：{}", serialNumber); - } - } - } catch (Exception e) { - log.error("标记设备上电触发任务异常，设备编号：{}", device.getSerialNumber(), e); - } - } - - /** - * 触发设备上电任务（收到第一个数据包时） - * @param device 设备信息 - */ - public void triggerPowerOnTasks(Device device) { - try { - - // 查询该设备的所有设备上电触发任务 - DeviceJob queryJob = new DeviceJob(); - queryJob.setDeviceId(device.getDeviceId()); - queryJob.setJobType(5); // modbus轮询任务 - // 只查询正常状态的任务（如果用户手动改成正常，下次上电会再次触发） - queryJob.setStatus("0"); // 正常状态（"0"=正常，"1"=暂停） - - List jobList = deviceJobService.selectJobList(queryJob); - if (CollectionUtils.isEmpty(jobList)) { - return; - } - - // 查找设备上电触发任务（状态为正常即可，不需要检查sent状态） - for (DeviceJob job : jobList) { - if (isPowerOnTrigger(job.getRemark())) { - // 执行设备上电触发任务 - executePowerOnTriggerTask(job, device); - } - } - } catch (Exception e) { - log.error("处理设备上电触发任务异常，设备编号：{}", device.getSerialNumber(), e); - } - } - - /** - * 检查是否是设备上电触发任务 - * - * @param remark remark字段 - * @return true表示是设备上电触发任务 - */ - private boolean isPowerOnTrigger(String remark) { - if (remark == null || remark.isEmpty()) { - return false; - } - return remark.contains("power_on_trigger"); - } - - /** - * 执行设备上电触发任务 - * - * @param deviceJob 设备任务 - * @param device 设备信息 - */ - private void executePowerOnTriggerTask(DeviceJob deviceJob, Device device) { - try { - // 查询对应的ModbusJob - ModbusJob queryJob = new ModbusJob(); - queryJob.setJobId(deviceJob.getJobId()); - List modbusJobList = modbusJobService.selectModbusJobList(queryJob); - - if (CollectionUtils.isEmpty(modbusJobList)) { - return; - } - - // 执行所有相关的modbus任务 - boolean hasActiveTask = false; - // 记录设备上电触发任务下发的指令片段（索引2到11位） - String serialNumber = device.getSerialNumber().toUpperCase(); - String cacheKey = RedisKeyBuilder.buildPowerOnTriggerCommandKey(serialNumber); - - for (ModbusJob modbusJob : modbusJobList) { - if ("0".equals(modbusJob.getStatus())) { // 只执行正常状态的任务 - hasActiveTask = true; - // 记录下发的指令片段（索引2到11位，即第3到第12个字符） - String command = modbusJob.getCommand(); - if (StringUtils.isNotEmpty(command) && command.length() >= 12) { - // 去除空格并转为大写 - String cleanCommand = command.replaceAll("\\s+", "").toUpperCase(); - if (cleanCommand.length() >= 12) { - // 提取索引2到11位（即第3到第12个字符） - String commandSegment = cleanCommand.substring(2, 12); - // 存储到Redis缓存，设置过期时间为5分钟（设备上报应该在5分钟内完成） - redisCache.setCacheObject(cacheKey, commandSegment, 300, TimeUnit.SECONDS); - log.info("记录设备上电触发任务下发指令片段，设备编号：{}，指令片段：{}", serialNumber, commandSegment); - } - } - } - } - - if (hasActiveTask) { - // 通过反射调用DeviceJobInvoke.invokeMethod来执行任务 - try { - Class clazz = Class.forName("com.fastbee.iot.data.job.DeviceJobInvoke"); - Method method = clazz.getMethod("invokeMethod", DeviceJob.class); - method.invoke(null, deviceJob); - log.info("设备上电触发任务执行成功，设备编号：{}，任务ID：{}", device.getSerialNumber(), deviceJob.getJobId()); - - // 执行成功后，将任务状态改为暂停 - updatePowerOnTriggerTaskStatus(modbusJobList); - } catch (Exception e) { - log.error("执行设备上电触发任务异常，设备编号：{}，任务ID：{}", device.getSerialNumber(), deviceJob.getJobId(), e); - } - } - - } catch (Exception e) { - log.error("处理设备上电触发任务异常，设备编号：{}，任务ID：{}", device.getSerialNumber(), deviceJob.getJobId(), e); - } - } - - /** - * 更新设备上电触发任务状态为暂停 - * - * @param modbusJobList 设备任务 - */ - private void updatePowerOnTriggerTaskStatus(List modbusJobList) { - if (CollectionUtils.isEmpty(modbusJobList)) { - return; - } - log.info("更新设备上电触发任务状态为暂停，任务数量：{}", modbusJobList.size()); - java.util.Date executeTime = DateUtils.getNowDate(); - modbusJobList.forEach(modbusJob -> { - // 只更新status=0的任务为status=1，如果已经是status=1则跳过 - if (!"0".equals(modbusJob.getStatus())) { - log.debug("任务ID：{}的状态已经是{}，跳过更新", modbusJob.getTaskId(), modbusJob.getStatus()); - return; - } - ModbusJob job = new ModbusJob(); - job.setStatus("1"); - job.setTaskId(modbusJob.getTaskId()); - // 记录执行时间 - job.setExecuteTime(executeTime); - try { - // 将任务状态改为暂停（"1"=暂停，"0"=正常），并记录执行时间 - modbusJobService.updateModbusJob(job); - log.info("设备上电触发任务已设置为暂停状态并记录执行时间，任务ID：{}，执行时间：{}", modbusJob.getTaskId(), executeTime); - } catch (Exception e) { - log.error("更新设备上电触发任务状态异常，任务ID：{}", modbusJob.getTaskId(), e); - } - }); - - } - /** * @param device 设备状态 * @return 结果 @@ -2130,7 +1944,7 @@ public class DeviceServiceImpl implements IDeviceService { /** * 判断是否是设备上电触发的回包 - * 从Redis缓存中获取下发的指令片段（索引2到11位），与上报的指令片段进行比较 + * 从Redis缓存中获取下发的指令片段（索引2到3位，2个字符），与上报的指令片段进行比较 * * @param sources 原始消息数据包 * @param deviceNumber 设备编号 @@ -2142,13 +1956,36 @@ public class DeviceServiceImpl implements IDeviceService { } try { - // 从Redis缓存中获取下发的指令片段 + // 从Redis缓存中获取下发的指令片段（可能是单个片段或片段集合） String serialNumber = deviceNumber.toUpperCase(); String cacheKey = RedisKeyBuilder.buildPowerOnTriggerCommandKey(serialNumber); - String sentCommandSegment = redisCache.getCacheObject(cacheKey); + Object cachedSegments = redisCache.getCacheObject(cacheKey); // 如果缓存中没有下发的指令片段，说明不是设备上电触发的回包 - if (StringUtils.isEmpty(sentCommandSegment)) { + 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; } @@ -2172,15 +2009,16 @@ public class DeviceServiceImpl implements IDeviceService { hexStr = sources.toString().replaceAll("\\s+", "").toUpperCase(); } - if (hexStr.length() < 12) { + // 至少需要4个字符才能提取索引2到3位 + if (hexStr.length() < 4) { return false; } - // 提取上报指令的索引2到11位（即第3到第12个字符） - String actualSegment = hexStr.substring(2, 12); + // 提取上报指令的索引2到3位（即第3到第4个字符，2个字符） + String actualSegment = hexStr.substring(2, 4); - // 与下发的指令片段进行比较 - boolean isMatch = sentCommandSegment.equals(actualSegment); + // 与下发的指令片段集合进行比较（支持多个指令） + boolean isMatch = sentCommandSegments.contains(actualSegment); if (isMatch) { // 匹配成功，删除缓存（避免重复匹配） @@ -2350,3 +2188,4 @@ public class DeviceServiceImpl implements IDeviceService { } } + diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/ModbusJobServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/ModbusJobServiceImpl.java index 6da6537a..609d2ed9 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/ModbusJobServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/ModbusJobServiceImpl.java @@ -106,6 +106,15 @@ public class ModbusJobServiceImpl implements IModbusJobService { @Override public int insertModbusJob(ModbusJob modbusJob) { try { + // 自动填充 cycleType：如果前端没有传递，则根据 remark 判断 + if (modbusJob.getCycleType() == null) { + if (isPowerOnTrigger(modbusJob.getRemark())) { + modbusJob.setCycleType(3); // 设备上电触发 + } else { + modbusJob.setCycleType(1); // 周期循环（默认） + } + } + //查询关联的网关设备 DeviceAndProtocol deviceAndProtocol = deviceService.selectProtocolBySerialNumber(modbusJob.getSubSerialNumber()); //如果是网关子设备 @@ -202,6 +211,15 @@ public class ModbusJobServiceImpl implements IModbusJobService { */ @Override public int updateModbusJob(ModbusJob modbusJob) { + // 自动填充 cycleType：如果前端没有传递，则根据 remark 判断 + if (modbusJob.getCycleType() == null && modbusJob.getRemark() != null) { + if (isPowerOnTrigger(modbusJob.getRemark())) { + modbusJob.setCycleType(3); // 设备上电触发 + } else { + modbusJob.setCycleType(1); // 周期循环（默认） + } + } + // 如果状态从1（暂停）改为0（正常），需要清空执行时间 if (modbusJob.getTaskId() != null && modbusJob.getStatus() != null && "0".equals(modbusJob.getStatus())) { ModbusJob existingJob = modbusJobMapper.selectModbusJobByTaskId(modbusJob.getTaskId()); diff --git a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/ModbusJobMapper.xml b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/ModbusJobMapper.xml index a22ee698..250ac44f 100644 --- a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/ModbusJobMapper.xml +++ b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/ModbusJobMapper.xml @@ -17,10 +17,11 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" + - select task_id,job_name, sub_device_id, sub_serial_number, command,device_type, job_id, status, create_by, create_time, remark, execute_time from iot_modbus_job + select task_id,job_name, sub_device_id, sub_serial_number, command,device_type, job_id, status, create_by, create_time, remark, execute_time, cycle_type from iot_modbus_job