From 30a8100d646c44e6b8dd580a5e930f0db509419f Mon Sep 17 00:00:00 2001 From: zhugongping Date: Thu, 16 Apr 2026 16:34:37 +0800 Subject: [PATCH] =?UTF-8?q?20260416:=E7=BB=99=E8=BD=AE=E8=AE=AD=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E6=B7=BB=E5=8A=A0=E6=89=A7=E8=A1=8C=E6=97=B6=E9=97=B4?= =?UTF-8?q?=EF=BC=88=E4=B8=B4=E6=97=B6=E4=BC=98=E5=8C=96=E4=B9=8B=E5=90=8E?= =?UTF-8?q?=E5=8F=AF=E6=92=A4=E9=94=80=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/main/resources/application-test.yml | 10 ++++---- .../consumer/DevicePropFetchConsumer.java | 24 ++++++++----------- .../DeviceDisallowConcurrentExecution.java | 4 ++-- .../iot/data/job/DeviceJobExecution.java | 4 ++-- .../fastbee/iot/mapper/ModbusJobMapper.java | 5 ++++ .../iot/service/IModbusJobService.java | 7 ++++++ .../service/impl/ModbusJobServiceImpl.java | 17 +++++++++++++ .../resources/mapper/iot/ModbusJobMapper.xml | 8 +++++++ .../views/iot/device/device-modbus-task.vue | 9 ++++++- 9 files changed, 64 insertions(+), 24 deletions(-) diff --git a/fastbee-admin/src/main/resources/application-test.yml b/fastbee-admin/src/main/resources/application-test.yml index 0b2bbe0a..7157c618 100644 --- a/fastbee-admin/src/main/resources/application-test.yml +++ b/fastbee-admin/src/main/resources/application-test.yml @@ -47,10 +47,10 @@ spring: # redis 配置 redis: - host: 47.121.180.229 # 地址 + host: 47.121.180.229 # 地址 port: 6379 # 端口，默认为6379 database: 0 # 数据库索引 - password: fastbee # 密码 + password: Yan@ancekong2026 # 密码 timeout: 10s # 连接超时时间 lettuce: pool: @@ -62,9 +62,9 @@ spring: # mqtt 配置 mqtt: username: fastbee # 账号（仅用于后端自认证） - password: Yan@ancekong2026 # 密码（仅用于后端自认证） - host-url: tcp://47.121.180.229:1883 # 连接 Emqx 消息服务器地址 - #host-url: tcp://47.121.180.229:1883 # 内置netty mqtt broker地址 + password: Yan@ancekong2026 # 密码（仅用于后端自认证） + host-url: tcp://47.121.180.229:1883 # 连接 Emqx 消息服务器地址 + #host-url: tcp://47.121.180.229:1883 # 内置netty mqtt broker地址 client-id: ${random.int} # 客户端Id，不能相同，采用随机数 ${random.value} default-topic: test # 默认主题 timeout: 30 # 超时时间 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 4a330457..e81f91bb 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 @@ -55,7 +55,7 @@ import java.util.stream.Collectors; public class DevicePropFetchConsumer { - //锁集合 + // 锁集合 private final ConcurrentHashMap taskLocks = new ConcurrentHashMap<>(); @Autowired private PubMqttClient pubMqttClient; @@ -84,9 +84,9 @@ public class DevicePropFetchConsumer { ModbusPollMsg task = type == 1 ? job.getPollMsg() : getCommandList(deviceJob); if (CollectionUtils.isEmpty(task.getCommandList())) return; String serialNumber = task.getSerialNumber(); - //获取一个线程锁 + // 获取一个线程锁 Lock lock = taskLocks.computeIfAbsent(serialNumber, k -> new ReentrantLock()); - //阻塞直到获取到锁 + // 阻塞直到获取到锁 lock.lock(); try { Long productId = task.getProductId(); @@ -102,7 +102,7 @@ public class DevicePropFetchConsumer { //通过mqtt内部客户端 下发指令 case MQTT: publish(topic, commandData); - log.info("=>MQTT-线程=[{}],轮询指令:[{}],主题:[{}]", Thread.currentThread().getName(), command, topic); + log.info("=>MQTT-线程=[{}],设备=[{}],轮询指令:[{}],主题:[{}]", Thread.currentThread().getName(), serialNumber, command, topic); break; // 下发TCP客户端 case TCP: @@ -110,13 +110,18 @@ public class DevicePropFetchConsumer { msg.setClientId(serialNumber); msg.setPayload(Unpooled.wrappedBuffer(commandData)); messageManager.requestR(serialNumber, msg, Message.class); - log.info("=>TCP-线程=[{}],轮询指令:[{}]", Thread.currentThread().getName(), command); + log.info("=>TCP-线程=[{}],设备=[{}],轮询指令:[{}]", Thread.currentThread().getName(), serialNumber, command); break; } // 指令间隔时间 Thread.sleep(3000); } + // 临时：仅定时轮询有 deviceJob/jobId；手动下发不更新（避免 NPE） + if (Objects.equals(type, 2) && deviceJob.getJobId() != null) { + modbusJobService.updateTaskExecuteTime(deviceJob.getJobId()); + } + } catch (Exception e) { log.error("线程错误e", e); } finally { @@ -136,21 +141,12 @@ public class DevicePropFetchConsumer { handleSubDeviceStatus(modbusJobList); List commandList = modbusJobList.stream().map(ModbusJob::getCommand).collect(Collectors.toList()); - // 检查是否有设备上电触发任务（cycleType=3） -// boolean hasPowerOnTrigger = modbusJobList.stream() -// .anyMatch(job -> job.getCycleType() != null && job.getCycleType() == 3); - // modbus-tcp 重新处理事务标识符 String subSerialNumber = modbusJobList.get(0).getSubSerialNumber(); String tcpKey = RedisKeyBuilder.buildModbusTcpCacheKey(subSerialNumber); String tcpObject = redisCache.getStrCacheObject(tcpKey); List tcpCommandList = new ArrayList<>(); - // 如果是设备上电触发任务（cycleType=3），跳过事务ID处理，直接使用原始指令 -// if (hasPowerOnTrigger) { -// log.info("检测到设备上电触发任务（cycleType=3），跳过事务ID处理，使用原始指令"); -// tcpCommandList.addAll(commandList); -// } else if (StringUtils.isNotEmpty(tcpObject)) { // 原有的事务ID处理逻辑 String tcpRuntimeCacheKey = RedisKeyBuilder.buildModbusTcpRuntimeCacheKey(subSerialNumber); diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceDisallowConcurrentExecution.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceDisallowConcurrentExecution.java index 83397cdb..62c5bb74 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceDisallowConcurrentExecution.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceDisallowConcurrentExecution.java @@ -6,8 +6,8 @@ import org.quartz.JobExecutionContext; /** * 定时任务处理（禁止并发执行） - * - * @author ruoyi + * 串行定时任务调度器 + * @author sumuxi, 2026-04-16 */ @DisallowConcurrentExecution public class DeviceDisallowConcurrentExecution extends DeviceAbstractQuartzJob { diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceJobExecution.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceJobExecution.java index 3962bef8..87788c0b 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceJobExecution.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/job/DeviceJobExecution.java @@ -5,8 +5,8 @@ import org.quartz.JobExecutionContext; /** * 定时任务处理（允许并发执行） - * - * @author ruoyi + * 并发调度器 + * @author sumuxi, 2026-04-16 */ public class DeviceJobExecution extends DeviceAbstractQuartzJob { @Override diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/ModbusJobMapper.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/ModbusJobMapper.java index b0ad1638..78cf56ce 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/ModbusJobMapper.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/ModbusJobMapper.java @@ -44,6 +44,11 @@ public interface ModbusJobMapper { */ public int updateModbusJob(ModbusJob modbusJob); + /** + * 按设备定时任务 job_id 批量更新执行时间（排除上电触发 cycle_type=3） + */ + int updateTaskExecuteTime(ModbusJob modbusJob); + /** * 清空执行时间 * diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IModbusJobService.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IModbusJobService.java index 059ff4dd..180577c7 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IModbusJobService.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IModbusJobService.java @@ -69,4 +69,11 @@ public interface IModbusJobService { * @return */ List selectDevicesJobByDeviceType(DeviceType deviceType, String serialNumber); + + /** + * 更新任务执行时间 + * + * @param jobId 任务ID + */ + void updateTaskExecuteTime(Long jobId); } 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 4b624e2b..0e118fcd 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 @@ -24,6 +24,7 @@ import org.springframework.util.CollectionUtils; import javax.annotation.Resource; import java.util.ArrayList; +import java.util.Date; import java.util.List; import java.util.Objects; import java.util.stream.Collectors; @@ -341,6 +342,22 @@ public class ModbusJobServiceImpl implements IModbusJobService { } } + /** + * 临时方案：按 Quartz 设备任务 jobId 批量刷新同组轮询行的执行时间（见 Mapper 排除 cycle_type=3） + */ + @Override + public void updateTaskExecuteTime(Long jobId) { + + if (jobId == null) { + return; + } + + ModbusJob modbusJob = new ModbusJob(); + modbusJob.setJobId(jobId); + modbusJob.setExecuteTime(new Date()); + modbusJobMapper.updateTaskExecuteTime(modbusJob); + } + /** * 组装并保存设备任务 * @param modbusJob 待新增的modbus任务 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 250ac44f..089f28a1 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 @@ -95,6 +95,14 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" where task_id = #{taskId} + + + + update iot_modbus_job + set execute_time = #{executeTime} + where job_id = #{jobId} + and (cycle_type is null or cycle_type != 3) + diff --git a/yack-web/src/views/iot/device/device-modbus-task.vue b/yack-web/src/views/iot/device/device-modbus-task.vue index c12ac9ed..226d3507 100644 --- a/yack-web/src/views/iot/device/device-modbus-task.vue +++ b/yack-web/src/views/iot/device/device-modbus-task.vue @@ -77,7 +77,7 @@