From 56cc5bce821104a5eaa2071984bdc12d7678c6a6 Mon Sep 17 00:00:00 2001 From: George Date: Thu, 29 Jan 2026 22:46:50 +0800 Subject: [PATCH] =?UTF-8?q?20260126=EF=BC=9A1=E3=80=81=E4=B8=8A=E7=94=B5?= =?UTF-8?q?=E8=A7=A6=E5=8F=91=E4=BB=BB=E5=8A=A1=E6=94=B9=E9=80=A0=E7=89=88?= =?UTF-8?q?=EF=BC=8C=E7=AE=80=E5=8C=96=E9=80=BB=E8=BE=91=E5=9C=A8TCPMsgAda?= =?UTF-8?q?pter=E8=AF=86=E5=88=AB=E5=8C=B9=E9=85=8D=E4=B8=8A=E7=94=B5?= =?UTF-8?q?=E8=A7=A6=E5=8F=91=E5=9B=9E=E5=8C=85=EF=BC=9B=E5=9C=A8TcpHandle?= =?UTF-8?q?rInterceptor=E5=A4=84=E7=90=86=E6=B3=A8=E5=86=8C=E4=B8=8A?= =?UTF-8?q?=E7=94=B5redisKey=EF=BC=8C=E5=9C=A8=E7=AC=AC=E4=B8=80=E4=B8=AA?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=8C=85=E4=B8=8A=E6=9D=A5=E4=B9=8B=E5=90=8E?= =?UTF-8?q?=E8=BD=AE=E8=AF=A2=E4=B8=8B=E5=8F=91=E4=BB=BB=E5=8A=A1=E3=80=82?= =?UTF-8?q?2=E3=80=81=E5=9C=A8TCPMsgAdapter=E8=AF=86=E5=88=AB=E5=BD=93?= =?UTF-8?q?=E5=89=8D=E6=98=AF=E7=AC=AC=E5=87=A0=E5=B8=A7=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=8C=85=EF=BC=8C=E5=9C=A8Session=E4=B8=AD=E7=94=A8=E5=B8=A7?= =?UTF-8?q?=E8=AE=A1=E6=95=B0=E5=99=A8=E8=AE=B0=E5=BD=95=E5=AE=A2=E6=88=B7?= =?UTF-8?q?=E7=AB=AF=E5=8F=91=E9=80=81=E7=9A=84=E5=B8=A7=E6=95=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../common/constant/FastBeeConstant.java | 14 +- .../common/core/redis/RedisKeyBuilder.java | 39 +--- .../executor/PowerOnTriggerTaskExecutor.java | 214 ++++++++++++------ .../service/impl/DeviceJobServiceImpl.java | 33 +-- .../impl/DeviceReportMessageServiceImpl.java | 2 +- .../com/fastbee/base/session/Session.java | 70 +----- .../tcp/config/TcpHandlerInterceptor.java | 15 ++ .../server/handler/DispatcherHandler.java | 2 +- .../server/handler/MessageEncoderWrapper.java | 4 +- .../server/handler/TCPMessageAdapter.java | 31 ++- .../service/impl/ModbusJobServiceImpl.java | 79 +++---- .../views/iot/device/device-modbus-task.vue | 18 +- .../views/iot/device/instruction-parsing.vue | 18 +- 13 files changed, 253 insertions(+), 286 deletions(-) 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 65dbaa7e..e452b7a6 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,19 +287,9 @@ public interface FastBeeConstant { String DEVICE_OTA_DATA = "device:ota:"; /** - * 设备上电触发指令缓存 + * Redis 上电触发标识前缀 */ - String POWER_ON_TRIGGER_COMMAND = "power:on:trigger:command:"; - - /** - * 设备上电触发任务待触发标记 - */ - String POWER_ON_TRIGGER_PENDING = "power:on:trigger:pending:"; - - /** - * 设备上电触发任务ID列表缓存（用于避免重复查询数据库） - */ - String POWER_ON_TRIGGER_JOB_IDS = "power:on:trigger:jobIds:"; + String POWER_ON_TRIGGER_KEY_PREFIX = "power_on:trigger:"; } 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 18d5ecd2..76a8983a 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 @@ -167,43 +167,10 @@ public class RedisKeyBuilder { } /** - * 设备上电触发指令缓存key - * 用于存储设备上电触发任务下发的指令片段（索引2到3位） + * Redis 上电触发标识key */ - public static String buildPowerOnTriggerCommandKey(String serialNumber) { - return FastBeeConstant.REDIS.POWER_ON_TRIGGER_COMMAND + serialNumber.toUpperCase(); - } - - /** - * 设备上电触发任务待触发标记key - * 用于标记设备是否需要在上电后收到第一个数据包时触发任务 - */ - public static String buildPowerOnTriggerPendingKey(String serialNumber) { - 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(); - } - - /** - * 纯计算物模型缓存key（产品级别） - * 用于缓存产品下的纯计算物模型标识符集合 - */ - public static String buildCalculatedThingsModelKey(Long productId) { - return "CALCULATED_TM:PRODUCT:" + productId; - } - - /** - * 纯计算物模型缓存key（设备级别，未来扩展） - * 用于缓存设备下的纯计算物模型标识符集合（当设备有自定义物模型选择时使用） - */ - public static String buildCalculatedThingsModelDeviceKey(Long deviceId) { - return "CALCULATED_TM:DEVICE:" + deviceId; + public static String buildPowerOnTriggerFlagKey(String serialNumber) { + return FastBeeConstant.REDIS.POWER_ON_TRIGGER_KEY_PREFIX + serialNumber.toUpperCase(); } } 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 ccbe0b74..607389e5 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 @@ -1,48 +1,34 @@ 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.constant.HttpStatus; -import com.fastbee.common.core.domain.AjaxResult; 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.data.redis.core.RedisTemplate; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Component; import org.springframework.util.CollectionUtils; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; import javax.annotation.Resource; import java.time.Duration; -import java.util.ArrayList; import java.util.Date; import java.util.List; -import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; /** * 设备上电触发任务执行器 - * 专门用于执行设备上电触发任务，直接下发指令，不通过队列，确保指令及时准确下发 + * 采用串行发送方案：发送一条指令 → 等待响应（或超时）→ 再发送下一条 + * 这是 Modbus RTU 协议的标准做法，确保请求响应正确匹配 * * @author Sumuxi */ @@ -50,83 +36,179 @@ import java.util.concurrent.atomic.AtomicBoolean; @Component public class PowerOnTriggerTaskExecutor { - @Autowired - private PubMqttClient pubMqttClient; - - @Autowired - private RedisCache redisCache; - + /** + * 响应超时时间（毫秒） + * 如果设备在此时间内没有响应，则判定为失败 + */ + private static final long RESPONSE_TIMEOUT_MS = 5000; + + /** 标识过期时间（秒），防止标识残留 */ + private static final long TRIGGER_FLAG_EXPIRE_SECONDS = 120; + @Resource private MessageManager messageManager; - @Resource - private TopicsUtils topicsUtils; - - @Resource - private IDeviceService deviceService; - @Resource private ISessionStore sessionStore; @Resource private IModbusJobService modbusJobService; + @Resource + private RedisTemplate redisTemplate; + /** - * 每台设备同一时刻仅允许一个“上电触发轮询”流水线运行，避免重复触发/并发下发导致应答错配。 + * 每台设备同一时刻仅允许一个"上电触发轮询"流水线运行，避免重复触发/并发下发导致应答错配 */ private final ConcurrentHashMap running = new ConcurrentHashMap<>(); /** - * 异步检查并触发设备上电任务（在收到第一个数据包时） - * 优化：使用 taskId 列表作为标志位，列表为空或不存在表示没有待执行任务 + * 设置上电触发标识（设备注册/上线成功后调用） + * @param serialNumber 设备编号 + */ + public void setPowerOnTriggerFlag(String serialNumber) { + if (StringUtils.isEmpty(serialNumber)) { + return; + } + String key = RedisKeyBuilder.buildPowerOnTriggerFlagKey(serialNumber); + redisTemplate.opsForValue().set(key, "1", TRIGGER_FLAG_EXPIRE_SECONDS, TimeUnit.SECONDS); + log.debug("设置上电触发标识，设备：{}", serialNumber); + } + + /** + * 检查并消费上电触发标识（原子操作） + * @param serialNumber 设备编号 + * @return true-有标识且已消费，false-无标识 + */ + private boolean consumePowerOnTriggerFlag(String serialNumber) { + String key = RedisKeyBuilder.buildPowerOnTriggerFlagKey(serialNumber); + // 使用 delete 返回值判断 key 是否存在，原子操作 + Boolean deleted = redisTemplate.delete(key); + return Boolean.TRUE.equals(deleted); + } + + + /** + * 异步检查并触发设备上电任务（在收到第一个数据包时） + * 采用串行发送方案：发送一条指令 → 等待响应（或超时）→ 再发送下一条 + * * @param serialNumber 设备编号 - * @param productId 产品ID */ @Async(FastBeeConstant.TASK.TRIGGER_POWER_ON_TASK_ASYNC) - public void checkAndTriggerPowerOnTaskAsync(String serialNumber, Long productId) { + public void checkAndTriggerPowerOnTaskAsync(String serialNumber) { if (StringUtils.isEmpty(serialNumber)) { log.warn("设备编号为空，无法触发设备上电任务"); return; } - // 延迟1秒，等待设备上线 - try {Thread.sleep(1000);} catch (InterruptedException e) {log.error("设备上电触发任务执行失败", e);} - - // 查询设备任务 - 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); + // 检查并消费上电触发标识（原子操作，确保每次上电只执行一次） + if (!consumePowerOnTriggerFlag(serialNumber)) { + log.debug("无上电触发标识，跳过执行，设备：{}", serialNumber); return; } - for (ModbusJob modbusJob : modbusJobList) { - String command = modbusJob.getCommand(); - byte[] commandData; - try { - 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 isRunning = running.computeIfAbsent(serialNumber, k -> new AtomicBoolean(false)); + if (!isRunning.compareAndSet(false, true)) { + log.debug("设备上电触发任务正在执行中，跳过重复触发，设备编号：{}", serialNumber); + return; } + try { + // 延迟1秒，等待设备上线稳定 + Thread.sleep(1000); + + // 查询设备上电触发任务（cycleType=3，状态正常） + 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; + } + + log.info("开始执行设备上电触发任务，设备编号：{}，任务数量：{}", serialNumber, modbusJobList.size()); + + // 串行执行每条任务 + for (ModbusJob modbusJob : modbusJobList) { + boolean success = executeTaskAndWaitResponse(serialNumber, modbusJob); + if (success) { + // 执行成功，更新任务状态为暂停 + updateTaskExecuteTime(modbusJob.getTaskId()); + log.info("上电触发任务执行成功，设备：{}，任务ID：{}，任务名：{}", serialNumber, modbusJob.getTaskId(), modbusJob.getJobName()); + } else { + log.warn("上电触发任务执行失败/超时，设备：{}，任务ID：{}，任务名：{}", serialNumber, modbusJob.getTaskId(), modbusJob.getJobName()); + } + + // 任务间增加100ms间隔，避免设备处理不过来 + Thread.sleep(100); + } + + log.info("设备上电触发任务执行完成，设备编号：{}", serialNumber); + + } catch (InterruptedException e) { + log.error("设备上电触发任务执行被中断，设备编号：{}", serialNumber, e); + Thread.currentThread().interrupt(); + } catch (Exception e) { + log.error("设备上电触发任务执行异常，设备编号：{}", serialNumber, e); + } finally { + // 释放锁 + isRunning.set(false); + } } /** - * 更新任务执行时间（仅在收到响应时调用） + * 执行单条任务并等待响应 + * + * @param serialNumber 设备编号 + * @param modbusJob 任务 + * @return true=收到响应（执行成功），false=超时或失败 + */ + private boolean executeTaskAndWaitResponse(String serialNumber, ModbusJob modbusJob) { + String command = modbusJob.getCommand(); + + // 转换指令 + byte[] commandData; + try { + commandData = ByteUtils.convertToHexByteArray(command); + } catch (Exception e) { + log.warn("指令转换异常，设备：{}，任务ID：{}，指令：{}", serialNumber, modbusJob.getTaskId(), command, e); + return false; + } + + // 构建消息 + Message msg = new Message(); + msg.setClientId(serialNumber); + msg.setPayload(Unpooled.wrappedBuffer(commandData)); + + // 获取会话并设置流水号 + Session session = sessionStore.getSession(serialNumber); + if (session == null) { + log.warn("设备会话不存在，设备：{}", serialNumber); + return false; + } + msg.setSerNo(String.valueOf(session.nextSerialNO())); + + try { + // 发送指令并阻塞等待响应，最多等待200ms + // 只要没超时，就说明收到了响应（不关心响应内容） + Object result = messageManager.requestR(serialNumber, msg, Message.class) + .timeout(Duration.ofMillis(RESPONSE_TIMEOUT_MS)) + .block(); + return result != null; + } catch (Exception e) { + // 超时 + return false; + } + } + + /** + * 更新任务执行时间和状态 + * 收到设备响应后调用，将任务状态设为暂停（status=1），记录执行时间 + * 用户可手动将状态改回正常（status=0）以便下次上电重新触发 * * @param taskId 任务ID */ @@ -138,7 +220,7 @@ public class PowerOnTriggerTaskExecutor { ModbusJob job = new ModbusJob(); job.setTaskId(taskId); job.setExecuteTime(new Date()); - job.setStatus("1"); + job.setStatus("1"); // 暂停状态，防止下次上电重复执行 modbusJobService.updateModbusJob(job); } diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceJobServiceImpl.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceJobServiceImpl.java index 37658072..cde358c5 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceJobServiceImpl.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceJobServiceImpl.java @@ -50,7 +50,7 @@ public class DeviceJobServiceImpl implements IDeviceJobService { List jobList = jobMapper.selectJobAll(); for (DeviceJob deviceJob : jobList) { // 跳过设备上电触发任务，它们不需要加入调度器 - if (!isPowerOnTriggerTask(deviceJob) && StringUtils.isNotEmpty(deviceJob.getCronExpression())) { + if (StringUtils.isNotEmpty(deviceJob.getCronExpression())) { DeviceScheduleUtils.createScheduleJob(scheduler, deviceJob); } } @@ -254,38 +254,14 @@ public class DeviceJobServiceImpl implements IDeviceJobService { public int insertJob(DeviceJob deviceJob) throws SchedulerException, TaskException { int rows = jobMapper.insertJob(deviceJob); if (rows > 0) { - // 检查是否是设备上电触发任务（通过检查remark或cron表达式） - // 设备上电触发任务不需要创建定时任务，因为它是事件触发的 - boolean isPowerOnTrigger = isPowerOnTriggerTask(deviceJob); - if (!isPowerOnTrigger && StringUtils.isNotEmpty(deviceJob.getCronExpression())) { + // CronExpression为空为设备上电触发任务，设备上电触发任务不需要创建定时任务，因为它是事件触发的 + if (StringUtils.isNotEmpty(deviceJob.getCronExpression())) { DeviceScheduleUtils.createScheduleJob(scheduler, deviceJob); } } return rows; } - /** - * 检查是否是设备上电触发任务 - * - * @param deviceJob 设备任务 - * @return true表示是设备上电触发任务 - */ - private boolean isPowerOnTriggerTask(DeviceJob deviceJob) { - if (deviceJob == null) { - return false; - } - String remark = deviceJob.getRemark(); - if (StringUtils.isNotEmpty(remark) && remark.contains("power_on_trigger")) { - return true; - } - // 如果cron表达式为空或null，且remark包含power_on_trigger，则认为是设备上电触发 - String cronExpression = deviceJob.getCronExpression(); - if (StringUtils.isEmpty(cronExpression) && StringUtils.isNotEmpty(remark) && remark.contains("power_on_trigger")) { - return true; - } - return false; - } - /** * 更新任务的时间表达式 * @@ -317,8 +293,7 @@ public class DeviceJobServiceImpl implements IDeviceJobService { scheduler.deleteJob(jobKey); } // 检查是否是设备上电触发任务，如果是则不创建定时任务 - boolean isPowerOnTrigger = isPowerOnTriggerTask(deviceJob); - if (!isPowerOnTrigger && StringUtils.isNotEmpty(deviceJob.getCronExpression())) { + if (StringUtils.isNotEmpty(deviceJob.getCronExpression())) { DeviceScheduleUtils.createScheduleJob(scheduler, deviceJob); } } 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 6d52e298..5d258a54 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 @@ -97,7 +97,7 @@ public class DeviceReportMessageServiceImpl implements IDeviceReportMessageServi Long productId = bo.getProductId(); // 异步检查是否需要触发上电任务（收到第一个数据包时触发） - powerOnTriggerTaskExecutor.checkAndTriggerPowerOnTaskAsync(serialNumber, productId); + powerOnTriggerTaskExecutor.checkAndTriggerPowerOnTaskAsync(serialNumber); switch (bo.getServerType()) { case MQTT: 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 90e35c75..a7986c4b 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 @@ -152,28 +152,6 @@ public class Session { return className; } - /*返回流水号*/ - private static String responseKey(Object response) { - String className = response.getClass().getName(); - if (response instanceof Response) { - 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; - } /** * 增加未注册前的帧计数 @@ -313,46 +291,18 @@ public class Session { } /** - * 消息响应 + * 消息响应（极简版：单请求快速匹配） + * 串行发送场景下，同一时刻只有一个等待请求，直接匹配即可 + * 无需解码、无需 serNo 匹配，收到任何响应都传给等待者 */ - public boolean response(Message message) { - // 先尝试精确匹配 - String key = responseKey(message); - MonoSink monoSink = topicSubscribers.get(key); - if (monoSink != null) { - monoSink.success(message); - return true; + public boolean response(Object message) { + if (topicSubscribers.isEmpty()) { + return false; } - - // 如果精确匹配失败，且响应是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 - 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; + Map.Entry entry = topicSubscribers.entrySet().iterator().next(); + entry.getValue().success(message); + topicSubscribers.remove(entry.getKey()); + return true; } /** diff --git a/fastbee-server/boot-strap/src/main/java/com/fastbee/bootstrap/tcp/config/TcpHandlerInterceptor.java b/fastbee-server/boot-strap/src/main/java/com/fastbee/bootstrap/tcp/config/TcpHandlerInterceptor.java index d1fa8103..361deb62 100644 --- a/fastbee-server/boot-strap/src/main/java/com/fastbee/bootstrap/tcp/config/TcpHandlerInterceptor.java +++ b/fastbee-server/boot-strap/src/main/java/com/fastbee/bootstrap/tcp/config/TcpHandlerInterceptor.java @@ -3,9 +3,12 @@ package com.fastbee.bootstrap.tcp.config; import com.fastbee.base.core.HandlerInterceptor; import com.fastbee.base.session.Session; import com.fastbee.common.core.protocol.Message; +import com.fastbee.iot.data.executor.PowerOnTriggerTaskExecutor; import com.fastbee.modbus.pak.TcpDtu; import lombok.extern.slf4j.Slf4j; +import javax.annotation.Resource; + /** * 消息拦截应答 * @@ -13,6 +16,11 @@ import lombok.extern.slf4j.Slf4j; */ @Slf4j public class TcpHandlerInterceptor implements HandlerInterceptor { + + @Resource + private PowerOnTriggerTaskExecutor powerOnTriggerTaskExecutor; + + @Override public Message notSupported(Message request, Session session) { return null; @@ -44,6 +52,13 @@ public class TcpHandlerInterceptor implements HandlerInterceptor { if (response != null) { //response.setSerialNo(session.nextSerialNO()); } + + // 注册报文处理完成后，设置上电触发标识 + int messageId = Integer.parseInt(request.getMessageId()); + if (messageId == TcpDtu.注册报文 && session.isRegistered()) { + powerOnTriggerTaskExecutor.setPowerOnTriggerFlag(session.getClientId()); + log.debug("设备注册成功，设置上电触发标识，设备：{}", session.getClientId()); + } } @Override 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 f4eae162..ca23ce4d 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 @@ -104,7 +104,7 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter { // 2/3 单个注册包，心跳，数据包上传，匹配消息id处理 } else if (null != request.getMessageId() && !"0".equals(request.getMessageId())) { - //处理心跳和设备注册 + // 处理心跳和设备注册 this.handleMessage(request, packet, ctx); request.setClientId(request.getClientId() == null ? session.getClientId() : request.getClientId()); this.handleOtherMsg(request); diff --git a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/MessageEncoderWrapper.java b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/MessageEncoderWrapper.java index 5b8b918c..dbc3c4ab 100644 --- a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/MessageEncoderWrapper.java +++ b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/MessageEncoderWrapper.java @@ -43,10 +43,10 @@ public class MessageEncoderWrapper extends ChannelOutboundHandlerAdapter { } output = null; } catch (EncoderException e) { - log.error("消息编码异常" + packet.message, e); + log.error("消息编码异常{}", packet.message, e); throw e; } catch (Throwable e) { - log.error("消息编码异常" + packet.message, e); + log.error("消息编码异常{}", packet.message, e); throw new EncoderException(e); } finally { if (output != null) { 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 bbcc5e85..9a078417 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 @@ -48,15 +48,26 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ByteBuf buf = (ByteBuf) msg; - /*session存储*/ + /* session存储 */ Session session = getSession(ctx); session.incrementFrameCount(); if (session.shouldCloseForNoRegister()) { - log.info("=>设备未注册，关闭连接session=[{}]，remoteAddress=[{}]", session.getClientId(),ctx.channel().remoteAddress()); + log.info("=>设备未注册，关闭连接session=[{}]，remoteAddress=[{}]", session.getClientId(), ctx.channel().remoteAddress()); ctx.close(); + buf.release(); // 释放 ByteBuf + return; } + session.access(); - /*将设备端消息传递给下一个处理器*/ + + // 快速响应匹配：串行发送场景，同一时刻只有一个等待请求 + if (session.hasPendingRequest() && session.response(buf.retainedDuplicate())) { + log.debug("=>设备响应匹配成功，设备：{}", session.getClientId()); + buf.release(); // 释放原始 ByteBuf + return; // 直接返回，不传递到后续处理器 + } + + /* 将设备端消息传递给下一个处理器 */ ctx.fireChannelRead(Packet.of(session, buf)); } @@ -66,9 +77,9 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { log.info("=>设备端连接{}", ctx.channel().remoteAddress()); - /*session存储*/ + /* session存储 */ Session session = getSession(ctx); - /*生成上线时间*/ + /* 生成上线时间 */ session.access(); } @@ -78,7 +89,7 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter { @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().remoteAddress(); - //连接IP地址 + // 连接IP地址 String host = socketAddress.getAddress().getHostAddress(); Session session = AttributeUtils.getSession(ctx.channel()); String clientId = session.getClientId(); @@ -112,7 +123,7 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter { @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().remoteAddress(); - //连接IP地址 + // 连接IP地址 String host = socketAddress.getAddress().getHostAddress(); int port = socketAddress.getPort(); if (evt instanceof IdleStateEvent) { @@ -123,7 +134,7 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter { ctx.close(); Session session = AttributeUtils.getSession(ctx.channel()); if (session != null) { - //移除客户端 + // 移除客户端 session.setIp(host); // session.invalidate(); } @@ -138,10 +149,10 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter { session = sessionManager.newInstance(channel); AttributeUtils.setSession(channel, session); } else { - //防止设备掉线不同步 + // 防止设备掉线不同步 boolean registered = session.isRegistered(); if (registered) { - //已经注册查询session是否存在 + // 已经注册查询session是否存在 if (!sessionManager.containKey(session.getClientId())) { sessionManager.add(session); } 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 609d2ed9..63f9ae32 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 @@ -66,10 +66,9 @@ public class ModbusJobServiceImpl implements IModbusJobService { List modbusJobList = modbusJobMapper.selectModbusJobList(modbusJob); if (!CollectionUtils.isEmpty(modbusJobList)) { for (ModbusJob job : modbusJobList) { - // 检查是否是设备上电触发 - if (isPowerOnTrigger(job.getRemark())) { + // 直接使用数据库中的 cycleType 判断 + if (job.getCycleType() != null && job.getCycleType() == 3) { job.setRemarkStr("设备上电触发"); - job.setCycleType(3); // 设备上电触发 } else { job.setRemarkStr(JobCronUtils.handleCronCycle(1, job.getRemark()).getDesc()); job.setCycleType(1); // 周期循环（默认） @@ -81,20 +80,13 @@ public class ModbusJobServiceImpl implements IModbusJobService { } /** - * 检查是否是设备上电触发 + * 检查是否是设备上电触发（直接使用 cycleType 判断） * - * @param remark remark字段 + * @param cycleType 时间周期类型 * @return true表示是设备上电触发 */ - private boolean isPowerOnTrigger(String remark) { - if (remark == null || remark.isEmpty()) { - return false; - } - try { - return remark.contains("power_on_trigger"); - } catch (Exception e) { - return false; - } + private boolean isPowerOnTrigger(Integer cycleType) { + return cycleType != null && cycleType == 3; } /** @@ -106,20 +98,16 @@ public class ModbusJobServiceImpl implements IModbusJobService { @Override public int insertModbusJob(ModbusJob modbusJob) { try { - // 自动填充 cycleType：如果前端没有传递，则根据 remark 判断 + // 如果前端没有传递 cycleType，默认为周期循环 if (modbusJob.getCycleType() == null) { - if (isPowerOnTrigger(modbusJob.getRemark())) { - modbusJob.setCycleType(3); // 设备上电触发 - } else { - modbusJob.setCycleType(1); // 周期循环（默认） - } + modbusJob.setCycleType(1); } - - //查询关联的网关设备 + + // 查询关联的网关设备 DeviceAndProtocol deviceAndProtocol = deviceService.selectProtocolBySerialNumber(modbusJob.getSubSerialNumber()); - //如果是网关子设备 + // 如果是网关子设备 if (deviceAndProtocol.getDeviceType() == DeviceType.SUB_GATEWAY.getCode()) { - //如果有绑定网关 + // 如果有绑定网关 if (!Objects.isNull(deviceAndProtocol.getGwDeviceId())) { DeviceJob deviceJob = new DeviceJob(); deviceJob.setDeviceId(deviceAndProtocol.getGwDeviceId()); @@ -127,7 +115,7 @@ public class ModbusJobServiceImpl implements IModbusJobService { deviceJob.setJobType(5); List list = deviceJobService.selectJobList(deviceJob); if (CollectionUtils.isEmpty(list)) { - //如果不存在定时任务则新建 + // 如果不存在定时任务则新建 deviceJob.setJobName("modbus poll"); deviceJob.setJobGroup("MODBUS"); deviceJob.setMisfirePolicy("2"); @@ -137,9 +125,9 @@ public class ModbusJobServiceImpl implements IModbusJobService { deviceJob.setSerialNumber(deviceAndProtocol.getGwSerialNumber()); deviceJob.setProductId(deviceAndProtocol.getGwProductId()); deviceJob.setCreateTime(DateUtils.getNowDate()); - //处理时间转换,这里原始的表达式放在remark - // 检查是否是设备上电触发 - if (isPowerOnTrigger(deviceJob.getRemark())) { + // 处理时间转换 + // 检查是否是设备上电触发（使用 cycleType 判断） + if (isPowerOnTrigger(modbusJob.getCycleType())) { // 设备上电触发不需要 cron 表达式，设置为null // DeviceJobServiceImpl.insertJob 会检查并跳过创建定时任务 deviceJob.setCronExpression(null); @@ -160,15 +148,15 @@ public class ModbusJobServiceImpl implements IModbusJobService { modbusJob.setDeviceType(deviceAndProtocol.getDeviceType()); return modbusJobMapper.insertModbusJob(modbusJob); } else { - //直连设备 + // 直连设备 DeviceJob deviceJob = new DeviceJob(); deviceJob.setDeviceId(modbusJob.getSubDeviceId()); deviceJob.setRemark(modbusJob.getRemark()); deviceJob.setJobType(5); List list = deviceJobService.selectJobList(deviceJob); if (CollectionUtils.isEmpty(list)) { - //创建定时任务 - //如果不存在定时任务则新建(设备创建定时任务之前会创建一个) + // 创建定时任务 + // 如果不存在定时任务则新建(设备创建定时任务之前会创建一个) deviceJob.setJobName("modbus poll"); deviceJob.setJobGroup("MODBUS"); deviceJob.setMisfirePolicy("2"); @@ -178,9 +166,9 @@ public class ModbusJobServiceImpl implements IModbusJobService { deviceJob.setSerialNumber(deviceAndProtocol.getSerialNumber()); deviceJob.setProductId(deviceAndProtocol.getProductId()); deviceJob.setCreateTime(DateUtils.getNowDate()); - //处理时间转换,这里原始的表达式放在remark - // 检查是否是设备上电触发 - if (isPowerOnTrigger(deviceJob.getRemark())) { + // 处理时间转换 + // 检查是否是设备上电触发（使用 cycleType 判断） + if (isPowerOnTrigger(modbusJob.getCycleType())) { // 设备上电触发不需要 cron 表达式，设置为null // DeviceJobServiceImpl.insertJob 会检查并跳过创建定时任务 deviceJob.setCronExpression(null); @@ -211,15 +199,7 @@ 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); // 周期循环（默认） - } - } - + // cycleType 由前端传递，不再根据 remark 判断 // 如果状态从1（暂停）改为0（正常），需要清空执行时间 if (modbusJob.getTaskId() != null && modbusJob.getStatus() != null && "0".equals(modbusJob.getStatus())) { ModbusJob existingJob = modbusJobMapper.selectModbusJobByTaskId(modbusJob.getTaskId()); @@ -252,16 +232,16 @@ public class ModbusJobServiceImpl implements IModbusJobService { @Transactional(rollbackFor = Exception.class) public int deleteModbusJobByTaskId(ModbusJob modbusJob) { try { - //先删除轮询任务 + // 先删除轮询任务 int result = modbusJobMapper.deleteModbusJobByTaskId(modbusJob.getTaskId()); - //查询是否有轮询任务 + // 查询是否有轮询任务 Long jobId = modbusJob.getJobId(); ModbusJob job = new ModbusJob(); job.setJobId(jobId); List modbusJobList = selectModbusJobList(job); if (CollectionUtils.isEmpty(modbusJobList)) { - //删除任务 - deviceJobService.deleteJobByIds(new Long[]{jobId}); + // 删除任务 + deviceJobService.deleteJobByIds(new Long[] { jobId }); } return result; } catch (SchedulerException e) { @@ -269,7 +249,6 @@ public class ModbusJobServiceImpl implements IModbusJobService { } } - /** * 根据设备类型获取所有轮询任务 * @@ -284,11 +263,11 @@ public class ModbusJobServiceImpl implements IModbusJobService { switch (deviceType) { case DIRECT_DEVICE: case SUB_GATEWAY: - //直连设备直接查询 + // 直连设备直接查询 modbusJob.setSubSerialNumber(serialNumber); return this.selectModbusJobList(modbusJob); case GATEWAY: - //网关设备需要查询所有子设备的modbus任务 + // 网关设备需要查询所有子设备的modbus任务 List subDeviceList = subGatewayService.getSubDeviceListByGw(serialNumber); List serialNumberList = subDeviceList.stream().map(SubDeviceBo::getSubDeviceNo).collect(Collectors.toList()); modbusJob.setSubDeviceList(serialNumberList); diff --git a/vue/src/views/iot/device/device-modbus-task.vue b/vue/src/views/iot/device/device-modbus-task.vue index 8ee62a8c..12d6ea96 100644 --- a/vue/src/views/iot/device/device-modbus-task.vue +++ b/vue/src/views/iot/device/device-modbus-task.vue @@ -298,9 +298,9 @@ {{ $t('device.device-modbus-task.384302-26') }} - @@ -506,7 +506,7 @@ export default { this.handlePush(); return; } - + // 如果指令为空，才生成新的指令 let params = { slaveId: parseInt(this.createForm.path), //从机地址 @@ -556,23 +556,21 @@ export default { handlePush() { /**计算定时时间*/ let cycle = ''; - + // 如果是设备上电触发，不需要 cron 表达式 if (this.createForm.cycleType === 3) { - // 设备上电触发，remark 存储特殊标识 - cycle = JSON.stringify({ type: 'power_on_trigger', sent: false }); + // 设备上电触发，remark 不再存储特殊标识，cycleType=3 即可标识 this.createForm.subDeviceId = this.device.deviceId; this.createForm.subSerialNumber = this.device.serialNumber; this.createForm.command = this.createCode; - this.createForm.remark = cycle; - // 设备上电触发不需要 cron 表达式，设置为空字符串 + this.createForm.remark = ''; // remark 留空或可填写真正的备注 this.createForm.cronExpression = ''; // 确保 cycleType 被传递 this.submitForm(); this.editDialog = false; return; } - + // 原有的周期循环和自定义时间段逻辑 const c = this.cycles1.map((item) => { if (item.interval === 'hour') { diff --git a/vue/src/views/iot/device/instruction-parsing.vue b/vue/src/views/iot/device/instruction-parsing.vue index 16fa3a6f..3713a0d9 100644 --- a/vue/src/views/iot/device/instruction-parsing.vue +++ b/vue/src/views/iot/device/instruction-parsing.vue @@ -5,15 +5,15 @@
{{ $t('device.instruction-parsing.830424-0') }} {{ device.serialNumber }} - - - {{ format }} - - - - {{ item }} - - + + + + + + + + +