初始化项目

云

2026-06-30 24681c81c09022f584a57006f2534b5f74723414

 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java

@@ -12,6 +12,7 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet;

/**
@@ -19,6 +20,8 @@
 * <p>
 * 封装通用的定时器管理、per-device 请求队列限速逻辑。
 * 子类只需实现 {@link #pollPoint(Long, Long)} 定义具体的轮询动作。
 * 如需将多个点位合并为一个轮询任务，可覆盖 {@link #buildPollTasks(IotModbusDeviceConfigRespDTO)}
 * 和 {@link #pollTask(Long, String)}。
 * <p>
 *
 * @author 芋道源码
@@ -38,9 +41,9 @@
    private static final int MAX_QUEUE_SIZE = 1000;

    /**
     * 设备点位的定时器映射：deviceId -> (pointId -> PointTimerInfo)
     * 设备轮询任务的定时器映射：deviceId -> (taskKey -> PollTimerInfo)
     */
    private final Map<Long, Map<Long, PointTimerInfo>> devicePointTimers = new ConcurrentHashMap<>();
    private final Map<Long, Map<String, PollTimerInfo>> devicePollTimers = new ConcurrentHashMap<>();

    /**
     * per-device 请求队列：deviceId -> 待执行请求队列
@@ -60,11 +63,29 @@
    }

    /**
     * 点位定时器信息
     * 轮询任务信息
     */
    @Data
    @AllArgsConstructor
    private static class PointTimerInfo {
    protected static class PollTask {

        /**
         * 任务标识
         */
        private String key;
        /**
         * 轮询间隔（用于判断是否需要更新定时器）
         */
        private Integer pollInterval;

    }

    /**
     * 轮询定时器信息
     */
    @Data
    @AllArgsConstructor
    private static class PollTimerInfo {

        /**
         * Vert.x 定时器 ID
@@ -82,75 +103,84 @@
    /**
     * 更新轮询任务（增量更新）
     *
     * 1. 【删除】点位：停止对应的轮询定时器
     * 2. 【新增】点位：创建对应的轮询定时器
     * 3. 【修改】点位：pollInterval 变化，重建对应的轮询定时器
     *    【修改】其他属性变化：不需要重建定时器（pollPoint 运行时从 configCache 取最新 point）
     * 1. 【删除】任务：停止对应的轮询定时器
     * 2. 【新增】任务：创建对应的轮询定时器
     * 3. 【修改】任务：pollInterval 变化，重建对应的轮询定时器
     *    【修改】其他属性变化：不需要重建定时器（pollTask 运行时从 configCache 取最新配置）
     */
    public void updatePolling(IotModbusDeviceConfigRespDTO config) {
        Long deviceId = config.getDeviceId();
        List<IotModbusPointRespDTO> newPoints = config.getPoints();
        Map<Long, PointTimerInfo> currentTimers = devicePointTimers
        List<PollTask> newTasks = buildPollTasks(config);
        Map<String, PollTimerInfo> currentTimers = devicePollTimers
                .computeIfAbsent(deviceId, k -> new ConcurrentHashMap<>());
        // 1.1 计算新配置中的点位 ID 集合
        Set<Long> newPointIds = convertSet(newPoints, IotModbusPointRespDTO::getId);
        // 1.2 计算删除的点位 ID 集合
        Set<Long> removedPointIds = new HashSet<>(currentTimers.keySet());
        removedPointIds.removeAll(newPointIds);
        // 1.1 计算新配置中的任务 Key 集合
        Set<String> newTaskKeys = convertSet(newTasks, PollTask::getKey);
        // 1.2 计算删除的任务 Key 集合
        Set<String> removedTaskKeys = new HashSet<>(currentTimers.keySet());
        removedTaskKeys.removeAll(newTaskKeys);

        // 2. 处理删除的点位：停止不再存在的定时器
        for (Long pointId : removedPointIds) {
            PointTimerInfo timerInfo = currentTimers.remove(pointId);
        // 2. 处理删除的任务：停止不再存在的定时器
        for (String taskKey : removedTaskKeys) {
            PollTimerInfo timerInfo = currentTimers.remove(taskKey);
            if (timerInfo != null) {
                vertx.cancelTimer(timerInfo.getTimerId());
                log.debug("[updatePolling][设备 {} 点位 {} 定时器已删除]", deviceId, pointId);
                log.debug("[updatePolling][设备 {} 轮询任务 {} 定时器已删除]", deviceId, taskKey);
            }
        }

        // 3. 处理新增和修改的点位
        if (CollUtil.isEmpty(newPoints)) {
        // 3. 处理新增和修改的任务
        if (CollUtil.isEmpty(newTasks)) {
            return;
        }
        for (IotModbusPointRespDTO point : newPoints) {
            Long pointId = point.getId();
            Integer newPollInterval = point.getPollInterval();
            PointTimerInfo existingTimer = currentTimers.get(pointId);
            // 3.1 新增点位：创建定时器
        for (PollTask task : newTasks) {
            String taskKey = task.getKey();
            Integer newPollInterval = task.getPollInterval();
            PollTimerInfo existingTimer = currentTimers.get(taskKey);
            // 3.1 新增任务：创建定时器
            if (existingTimer == null) {
                Long timerId = createPollTimer(deviceId, pointId, newPollInterval);
                Long timerId = createPollTimer(deviceId, taskKey, newPollInterval);
                if (timerId != null) {
                    currentTimers.put(pointId, new PointTimerInfo(timerId, newPollInterval));
                    log.debug("[updatePolling][设备 {} 点位 {} 定时器已创建, interval={}ms]",
                            deviceId, pointId, newPollInterval);
                    currentTimers.put(taskKey, new PollTimerInfo(timerId, newPollInterval));
                    log.debug("[updatePolling][设备 {} 轮询任务 {} 定时器已创建, interval={}ms]",
                            deviceId, taskKey, newPollInterval);
                }
            } else if (!Objects.equals(existingTimer.getPollInterval(), newPollInterval)) {
                // 3.2 pollInterval 变化：重建定时器
                vertx.cancelTimer(existingTimer.getTimerId());
                Long timerId = createPollTimer(deviceId, pointId, newPollInterval);
                Long timerId = createPollTimer(deviceId, taskKey, newPollInterval);
                if (timerId != null) {
                    currentTimers.put(pointId, new PointTimerInfo(timerId, newPollInterval));
                    log.debug("[updatePolling][设备 {} 点位 {} 定时器已更新, interval={}ms -> {}ms]",
                            deviceId, pointId, existingTimer.getPollInterval(), newPollInterval);
                    currentTimers.put(taskKey, new PollTimerInfo(timerId, newPollInterval));
                    log.debug("[updatePolling][设备 {} 轮询任务 {} 定时器已更新, interval={}ms -> {}ms]",
                            deviceId, taskKey, existingTimer.getPollInterval(), newPollInterval);
                } else {
                    currentTimers.remove(pointId);
                    currentTimers.remove(taskKey);
                }
            }
            // 3.3 其他属性变化：无需重建定时器，因为 pollPoint() 运行时从 configCache 获取最新 point，自动使用新配置
            // 3.3 其他属性变化：无需重建定时器，因为 pollTask() 运行时从 configCache 获取最新配置
        }
    }

    /**
     * 构建轮询任务列表
     *
     * 默认每个点位一个轮询任务。TCP Client 等协议可覆盖该方法，将多个点位合并为一个批量读取任务。
     */
    protected List<PollTask> buildPollTasks(IotModbusDeviceConfigRespDTO config) {
        return convertList(config.getPoints(), point -> new PollTask(String.valueOf(point.getId()), point.getPollInterval()));
    }

    /**
     * 创建轮询定时器
     */
    private Long createPollTimer(Long deviceId, Long pointId, Integer pollInterval) {
    private Long createPollTimer(Long deviceId, String taskKey, Integer pollInterval) {
        if (pollInterval == null || pollInterval <= 0) {
            return null;
        }
        return vertx.setPeriodic(pollInterval, timerId -> {
            try {
                submitPollRequest(deviceId, pointId);
                submitPollRequest(deviceId, taskKey);
            } catch (Exception e) {
                log.error("[createPollTimer][轮询点位失败, deviceId={}, pointId={}]", deviceId, pointId, e);
                log.error("[createPollTimer][轮询任务失败, deviceId={}, taskKey={}]", deviceId, taskKey, e);
            }
        });
    }
@@ -160,7 +190,7 @@
    /**
     * 提交轮询请求到设备请求队列（保证同设备请求间隔）
     */
    private void submitPollRequest(Long deviceId, Long pointId) {
    private void submitPollRequest(Long deviceId, String taskKey) {
        // 1. 【重要】将请求添加到设备的请求队列
        Queue<Runnable> queue = deviceRequestQueues.computeIfAbsent(deviceId, k -> new ConcurrentLinkedQueue<>());
        while (queue.size() >= MAX_QUEUE_SIZE) {
@@ -168,7 +198,7 @@
            queue.poll();
            log.warn("[submitPollRequest][设备 {} 请求队列已满({}), 丢弃最旧请求]", deviceId, MAX_QUEUE_SIZE);
        }
        queue.offer(() -> pollPoint(deviceId, pointId));
        queue.offer(() -> pollTask(deviceId, taskKey));

        // 2. 处理设备请求队列（如果没有延迟 timer 在等待）
        processDeviceQueue(deviceId);
@@ -239,6 +269,15 @@
    // ========== 轮询执行 ==========

    /**
     * 轮询任务
     *
     * 默认将任务标识作为点位 ID，执行单点轮询。
     */
    protected void pollTask(Long deviceId, String taskKey) {
        pollPoint(deviceId, Long.valueOf(taskKey));
    }

    /**
     * 轮询单个点位（子类实现具体的读取逻辑）
     *
     * @param deviceId 设备 ID
@@ -252,25 +291,25 @@
     * 停止设备的轮询
     */
    public void stopPolling(Long deviceId) {
        Map<Long, PointTimerInfo> timers = devicePointTimers.remove(deviceId);
        if (CollUtil.isEmpty(timers)) {
            return;
        }
        for (PointTimerInfo timerInfo : timers.values()) {
            vertx.cancelTimer(timerInfo.getTimerId());
        Map<String, PollTimerInfo> timers = devicePollTimers.remove(deviceId);
        if (CollUtil.isNotEmpty(timers)) {
            for (PollTimerInfo timerInfo : timers.values()) {
                vertx.cancelTimer(timerInfo.getTimerId());
            }
        }
        // 清理请求队列
        deviceRequestQueues.remove(deviceId);
        deviceLastRequestTime.remove(deviceId);
        deviceDelayTimerActive.remove(deviceId);
        log.debug("[stopPolling][设备 {} 停止了 {} 个轮询定时器]", deviceId, timers.size());
        log.debug("[stopPolling][设备 {} 停止了 {} 个轮询定时器]", deviceId,
                CollUtil.isEmpty(timers) ? 0 : timers.size());
    }

    /**
     * 停止所有轮询
     */
    public void stopAll() {
        for (Long deviceId : new ArrayList<>(devicePointTimers.keySet())) {
        for (Long deviceId : new ArrayList<>(devicePollTimers.keySet())) {
            stopPolling(deviceId);
        }
    }