refactor(websocket): 重构频道订阅处理器基类实现

- 提取公共基类 BaseChannelSubscribeHandler 处理订阅逻辑 - 将原处理器拆分为带 GRPC 调用和简单订阅两种实现 - 使用依赖注入替代 Autowired 注解 - 统一订阅和取消订阅的处理流程 - 优化日志记录和异常处理机制
2026-06-05 14:57:57 +08:00 · 2026-06-05 14:57:57 +08:00 · f57c9b9dda
commit f57c9b9dda
parent 05b90f8dc3
22 changed files with 312 additions and 126 deletions
--- a/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/inspection/InspectionWaypointController.java
+++ b/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/inspection/InspectionWaypointController.java
@ -111,4 +111,5 @@ public class InspectionWaypointController extends BaseController
    {
        return toAjax(inspectionWaypointService.deleteInspectionWaypointByIds(ids));
    }
+
 }
--- a/cmvr-iot-framework/src/main/java/com/cmvr/framework/websocket/handler/BaseChannelSubscribeHandler.java
+++ b/cmvr-iot-framework/src/main/java/com/cmvr/framework/websocket/handler/BaseChannelSubscribeHandler.java
@ -0,0 +1,95 @@
+package com.cmvr.framework.websocket.handler;
+
+import com.cmvr.framework.websocket.manager.ChannelSubscriptionManager;
+import com.cmvr.framework.websocket.message.WSChannelSubscriptionMessage;
+import com.cmvr.framework.websocket.service.GrpcClientService;
+import lombok.RequiredArgsConstructor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.web.socket.WebSocketSession;
+
+import java.util.concurrent.CompletableFuture;
+
+@RequiredArgsConstructor
+public abstract class BaseChannelSubscribeHandler implements WSMessageHandler<WSChannelSubscriptionMessage> {
+
+    protected final Logger log = LoggerFactory.getLogger(getClass());
+    protected final ChannelSubscriptionManager channelSubscriptionManager;
+    protected final GrpcClientService grpcClientService;
+
+    /**
+     * 是否需要GRPC远程调用
+     * true=格式 service/method/terminalId/deviceId 走grpc
+     * false=普通频道，只本地订阅，不调用grpc
+     */
+    protected abstract boolean isNeedGrpc(String channel);
+
+    /**
+     * 解析grpc四个参数 [service,method,terminalId,deviceId]
+     * 只有isNeedGrpc=true才会进入执行
+     */
+    protected abstract String[] parseGrpcParam(String channel);
+
+    @Override
+    public void handle(WebSocketSession session, WSChannelSubscriptionMessage message) throws Exception {
+        String action = message.getAction();
+        String userId = session.getAttributes().get("userId").toString();
+        String channel = message.getChannel();
+
+        switch (action) {
+            case "subscribe":
+                doSubscribe(channel, session);
+                break;
+            case "unsubscribe":
+                doUnSubscribe(channel, session);
+                break;
+            default:
+                log.warn("未知订阅动作：{}，channel:{}", action, channel);
+                return;
+        }
+        log.info("用户{} {}频道{}", userId, "subscribe".equals(action) ? "订阅" : "取消订阅", channel);
+    }
+
+    /** 订阅统一模板 */
+    private void doSubscribe(String channel, WebSocketSession session) {
+        boolean needGrpc = isNeedGrpc(channel);
+        // 没人订阅 + 需要grpc → 发起远端订阅
+        if (needGrpc && !channelSubscriptionManager.hasSubscribers(channel)) {
+            String[] params = parseGrpcParam(channel);
+            String service = params[0];
+            String method = params[1];
+            String terminalId = params[2];
+            String deviceId = params[3];
+            CompletableFuture.runAsync(() -> {
+                try {
+                    grpcClientService.send(true, terminalId, deviceId, service, method);
+                } catch (Exception e) {
+                    log.error("频道{} grpc订阅异常", channel, e);
+                }
+            });
+        }
+        // 本地统一订阅
+        channelSubscriptionManager.subscribe(channel, session);
+    }
+
+    /** 取消订阅统一模板 */
+    private void doUnSubscribe(String channel, WebSocketSession session) {
+        channelSubscriptionManager.unsubscribe(channel, session);
+        boolean needGrpc = isNeedGrpc(channel);
+        // 无任何订阅 + 需要grpc → 远端取消
+        if (needGrpc && !channelSubscriptionManager.hasSubscribers(channel)) {
+            String[] params = parseGrpcParam(channel);
+            String service = params[0];
+            String method = params[1];
+            String terminalId = params[2];
+            String deviceId = params[3];
+            CompletableFuture.runAsync(() -> {
+                try {
+                    grpcClientService.send(false, terminalId, deviceId, service, method);
+                } catch (Exception e) {
+                    log.error("频道{} grpc取消订阅异常", channel, e);
+                }
+            });
+        }
+    }
+}
--- a/cmvr-iot-framework/src/main/java/com/cmvr/framework/websocket/handler/ChannelSubscriptionMessageHandler.java
+++ b/cmvr-iot-framework/src/main/java/com/cmvr/framework/websocket/handler/ChannelSubscriptionMessageHandler.java
@ -5,67 +5,26 @@ import com.cmvr.framework.websocket.enums.WSMessageTypeEnum;
 import com.cmvr.framework.websocket.manager.ChannelSubscriptionManager;
 import com.cmvr.framework.websocket.message.WSChannelSubscriptionMessage;
 import com.cmvr.framework.websocket.service.GrpcClientService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.web.socket.WebSocketSession;
+import org.springframework.stereotype.Component;

-import java.util.concurrent.CompletableFuture;

-/**
- * 频道订阅消息处理
- */
+@Component
@WSMessageType(value = WSMessageTypeEnum.CHANNEL_SUBSCRIPTION, messageClass = WSChannelSubscriptionMessage.class)
-public class ChannelSubscriptionMessageHandler implements WSMessageHandler<WSChannelSubscriptionMessage> {
-    private static final Logger log = LoggerFactory.getLogger(ChannelSubscriptionMessageHandler.class);
-    @Autowired
-    private ChannelSubscriptionManager channelSubscriptionManager;
-    @Autowired
-    private GrpcClientService grpcClientService;
-    public void handle(WebSocketSession session, WSChannelSubscriptionMessage message) throws Exception {
-        String action = message.getAction();
-        String userId = session.getAttributes().get("userId").toString();
-        // service/method/terminalId/deviceId
-        String channel = message.getChannel();
-        String[] channelInfoArr = channel.split("\\/");
-        String service = channelInfoArr[0];
-        String method = channelInfoArr[1];
-        String terminalId = channelInfoArr[2];
-        String deviceId = channelInfoArr[3];
-        if ("subscribe".equals(action)) {
-            if (!channelSubscriptionManager.hasSubscribers(channel)) {
-                // 异步调用gRPC
-                CompletableFuture.runAsync(() -> {
-                    try {
-                        grpcClientService.send(true, terminalId, deviceId, service, method);
-                    } catch (Exception e) {
-                        log.error("gRPC订阅失败: {}", e.getMessage());
-                    }
-                });
-            }
-            channelSubscriptionManager.subscribe(channel, session);
-        } else if ("unsubscribe".equals(action)) {
-            channelSubscriptionManager.unsubscribe(channel, session);
-            if (!channelSubscriptionManager.hasSubscribers(channel)) {
-                CompletableFuture.runAsync(() -> {
-                    try {
-                        grpcClientService.send(false, terminalId, deviceId, service, method);
-                    } catch (Exception e) {
-                        log.error("gRPC取消订阅失败: {}", e.getMessage());
-                    }
-                });
-            }
-        } else {
-            System.out.println("未知频道操作: " + action);
-            return;
-        }
+public class ChannelSubscriptionMessageHandler extends BaseChannelSubscribeHandler {

-        // 统一记录日志
-        log.info("用户{} {} 频道{}", userId, getActionDescription(action), channel);
+    public ChannelSubscriptionMessageHandler(ChannelSubscriptionManager channelSubscriptionManager, GrpcClientService grpcClientService) {
+        super(channelSubscriptionManager, grpcClientService);
    }

-    // 根据动作获取描述文本
-    private String getActionDescription(String action) {
-        return "subscribe".equals(action) ? "订阅了" : "取消订阅了";
+    // 当前频道带 / 分段，需要grpc
+    @Override
+    protected boolean isNeedGrpc(String channel) {
+        return channel.contains("/");
    }
+
+    @Override
+    protected String[] parseGrpcParam(String channel) {
+        return channel.split("/");
+    }
+
 }
--- a/cmvr-iot-framework/src/main/java/com/cmvr/framework/websocket/handler/HeartbeatMessageHandler.java
+++ b/cmvr-iot-framework/src/main/java/com/cmvr/framework/websocket/handler/HeartbeatMessageHandler.java
@ -4,12 +4,10 @@ import com.cmvr.framework.websocket.annotation.WSMessageType;
 import com.cmvr.framework.websocket.enums.WSMessageTypeEnum;
 import com.cmvr.framework.websocket.message.WSHeartbeatMessage;
 import org.springframework.web.socket.WebSocketSession;
-
 import java.io.IOException;

@WSMessageType(value = WSMessageTypeEnum.HEARTBEAT, messageClass = WSHeartbeatMessage.class)
 public class HeartbeatMessageHandler implements WSMessageHandler<WSHeartbeatMessage>{
-
    @Override
    public void handle(WebSocketSession session, WSHeartbeatMessage message) throws IOException {
        System.out.println("heartbeat");
--- a/cmvr-iot-framework/src/main/java/com/cmvr/framework/websocket/handler/SimpleChannelSubscribeHandler.java
+++ b/cmvr-iot-framework/src/main/java/com/cmvr/framework/websocket/handler/SimpleChannelSubscribeHandler.java
@ -0,0 +1,29 @@
+package com.cmvr.framework.websocket.handler;
+
+import com.cmvr.framework.websocket.annotation.WSMessageType;
+import com.cmvr.framework.websocket.enums.WSMessageTypeEnum;
+import com.cmvr.framework.websocket.manager.ChannelSubscriptionManager;
+import com.cmvr.framework.websocket.message.WSChannelSubscriptionMessage;
+import com.cmvr.framework.websocket.service.GrpcClientService;
+import org.springframework.stereotype.Component;
+
+@Component
+@WSMessageType(value = WSMessageTypeEnum.CHANNEL_SUBSCRIPTION, messageClass = WSChannelSubscriptionMessage.class)
+public class SimpleChannelSubscribeHandler extends BaseChannelSubscribeHandler {
+
+    public SimpleChannelSubscribeHandler(ChannelSubscriptionManager channelSubscriptionManager, GrpcClientService grpcClientService) {
+        super(channelSubscriptionManager, grpcClientService);
+    }
+
+    // 简单频道不需要grpc调用
+    @Override
+    protected boolean isNeedGrpc(String channel) {
+        return false;
+    }
+
+    // 不会执行到，随便返回
+    @Override
+    protected String[] parseGrpcParam(String channel) {
+        return new String[0];
+    }
+}
--- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionTask.java
+++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionTask.java
@ -51,4 +51,8 @@ public class InspectionTask extends BaseEntity
    @ApiModelProperty("地图id")
    private String mapId;

+    @Excel(name = "点位数量")
+    @ApiModelProperty("点位数量")
+    private Integer waypointCount;
+
 }
--- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionTaskInstance.java
+++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionTaskInstance.java
@ -54,4 +54,8 @@ public class InspectionTaskInstance extends BaseEntity
    @ApiModelProperty("检测任务id")
    private String taskInsId;

+    @Excel(name = "点位数量")
+    @ApiModelProperty("点位数量")
+    private Integer waypointCount;
+
 }
--- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionTaskInstanceVo.java
+++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionTaskInstanceVo.java
@ -94,4 +94,8 @@ public class InspectionTaskInstanceVo extends BaseEntity
    @ApiModelProperty("检测任务id")
    private String taskInsId;

+    @Excel(name = "点位数量")
+    @ApiModelProperty("点位数量")
+    private Integer waypointCount;
+
 }
--- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionTaskVo.java
+++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionTaskVo.java
@ -66,4 +66,8 @@ public class InspectionTaskVo extends BaseEntity
    @ApiModelProperty("地图名称")
    private String mapName;

+    @Excel(name = "点位数量")
+    @ApiModelProperty("点位数量")
+    private Integer waypointCount;
+
 }
--- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/listener/InspectionFlowExecutionListener.java
+++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/listener/InspectionFlowExecutionListener.java
@ -1,6 +1,7 @@
 package com.cmvr.inspection.listener;

-import com.baomidou.mybatisplus.extension.conditions.query.LambdaQueryChainWrapper;
+import cn.hutool.core.map.MapUtil;
+import cn.hutool.core.util.StrUtil;
 import com.cmvr.common.utils.DateUtils;
 import com.cmvr.framework.websocket.service.MessagePushService;
 import com.cmvr.inspection.domain.InspectionTaskInstance;
@ -9,74 +10,144 @@ import com.cmvr.inspection.service.IInspectionTaskInstanceService;
 import com.cmvr.test.flow.runtime.event.FlowExecutionEvent;
 import com.cmvr.test.flow.runtime.event.FlowExecutionListener;
 import lombok.extern.slf4j.Slf4j;
-import org.jetbrains.annotations.NotNull;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;

-import java.util.HashMap;
-import java.util.Map;
+import java.util.*;

@Component
@Slf4j
 public class InspectionFlowExecutionListener implements FlowExecutionListener {
-    @Autowired
-    private IInspectionTaskInstanceService inspectionTaskInstanceService;
-    @Autowired
-    private MessagePushService messagePushService;
+    private final IInspectionTaskInstanceService inspectionTaskInstanceService;
+    private final MessagePushService messagePushService;
+
+    /** key:itemId, value:已完成节点集合 */
+    private final Map<String, Set<String>> itemNodeExecutMap = new HashMap<>();
+    /** key:taskInsId(流程实例id=taskInsId), value:已执行item集合 */
+    private final Map<String, Set<String>> taskItemExecutMap = new HashMap<>();
+
+    public InspectionFlowExecutionListener(IInspectionTaskInstanceService inspectionTaskInstanceService, MessagePushService messagePushService) {
+        this.inspectionTaskInstanceService = inspectionTaskInstanceService;
+        this.messagePushService = messagePushService;
+    }
+
    @Override
    public void onEvent(FlowExecutionEvent event) {
-        // 节点执行事件
-        if (event.getNodeId() != null) {
-            String info = event.getEventType() == FlowExecutionEvent.EventType.NODE_STARTED ? "开始执行" : "执行完成";
-            // 查询巡检任务实例，event中的instId是巡检任务中的taskInsId
-            InspectionTaskInstance inspectionTaskInstance = inspectionTaskInstanceService.lambdaQuery().eq(InspectionTaskInstance::getTaskInsId, event.getInstId())
-                    .one();
-            String insId = inspectionTaskInstance.getId();
-            Map<String, String> data = new HashMap<>();
-            data.put("insId", insId);
-            data.put("logInfo",  info);
-            try {
-                messagePushService.pushToChannel("Inspection/TaskInstance", data);
-            } catch (Exception e) {
-                log.info("如果没人订阅，则吃掉异常");
-            }
-
+        String instId = event.getInstId();
+        String itemId = event.getItemId();
+        String nodeId = event.getNodeId();
+        if (StrUtil.isBlank(instId)) {
+            return;
        }
-        // 通过 instId 找到巡检任务实例，并修改其状态
-        // 1.构造更新条件
-        LambdaQueryChainWrapper<InspectionTaskInstance> wrapper = inspectionTaskInstanceService.lambdaQuery()
-                .eq(InspectionTaskInstance::getTaskInsId, event.getInstId());
-        // 2.要更新的字段
-        InspectionTaskInstance updateEntity = getInspectionTaskInstance();
+
+        // 1.节点完成才记录当前item下node
+        if (StrUtil.isNotBlank(itemId) && StrUtil.isNotBlank(nodeId)
+                && FlowExecutionEvent.EventType.NODE_COMPLETED == event.getEventType()) {
+            itemNodeExecutMap.computeIfAbsent(itemId, k -> new HashSet<>()).add(nodeId);
+        }
+        // 任意事件：当前任务绑定item（item一启动就入Map，所以统计已完成item要-1）
+        if (StrUtil.isNotBlank(instId) && StrUtil.isNotBlank(itemId)) {
+            taskItemExecutMap.computeIfAbsent(instId, k -> new HashSet<>()).add(itemId);
+        }
+
+        InspectionTaskInstance taskInstance = inspectionTaskInstanceService.lambdaQuery()
+                .eq(InspectionTaskInstance::getTaskInsId, instId)
+                .one();
+        if (taskInstance == null) {
+            log.warn("未查询到巡检实例,taskInsId:{}", instId);
+            return;
+        }
+
+        int targetStatus = TaskStatusEnum.RUNNING.getCode();
+        boolean needUpdateDb = false;
+        Date now = DateUtils.getNowDate();
        switch (event.getEventType()) {
            case TASK_COMPLETED:
-                // 设置状态为成功
-                updateEntity.setStatus(TaskStatusEnum.SUCCESS.getCode());
-                // 正确API：update(更新实体, 条件Wrapper)
-                inspectionTaskInstanceService.update(updateEntity, wrapper.getWrapper());
-
+                targetStatus = TaskStatusEnum.SUCCESS.getCode();
+                needUpdateDb = true;
+                clearCache(instId);
+                pushCompleteMsg(taskInstance.getId(), now, targetStatus);
                break;
            case TASK_FAILED:
-                // 设置状态为失败
-                updateEntity.setStatus(TaskStatusEnum.FAILED.getCode());
-                // 正确API：update(更新实体, 条件Wrapper)
-                inspectionTaskInstanceService.update(updateEntity, wrapper.getWrapper());
+                targetStatus = TaskStatusEnum.FAILED.getCode();
+                needUpdateDb = true;
+                clearCache(instId);
                break;
            case NODE_COMPLETED:
-                // 判断当前任务所有节点是否执行完毕
+            default:
                break;
        }
+
+        if (needUpdateDb) {
+            taskInstance.setStatus(targetStatus);
+            taskInstance.setEndTime(now);
+            inspectionTaskInstanceService.updateById(taskInstance);
+        }
+
+        // 节点推送
+        if (StrUtil.isNotBlank(nodeId)) {
+            int waypointCount = taskInstance.getWaypointCount() == null ? 0 : taskInstance.getWaypointCount();
+            double progress = 0D;
+            if (waypointCount > 0) {
+                // 当前任务已执行item数量（item启动就存入集合，已完整跑完的item=总数-1）
+                Set<String> itemSet = taskItemExecutMap.getOrDefault(instId, new HashSet<>());
+                int itemTotal = itemSet.size();
+                int finishedItemNum = itemTotal - 1;
+
+                // 当前item已完成节点数
+                Set<String> finishedNodeSet = itemNodeExecutMap.getOrDefault(itemId, new HashSet<>());
+                int finishedNode = finishedNodeSet.size();
+                int totalNode = event.getNodeCount() == 0 ? 1 : event.getNodeCount();
+
+                // 已完整跑完点位占比
+                double itemTotalRatio = finishedItemNum / (double) waypointCount;
+                // 当前运行item内部节点占比
+                double innerRatio = finishedNode / (double) totalNode / waypointCount;
+
+                // 还原你的兜底：防止循环节点进度超限，最低不低于极小值
+                double singleItemBase = 1d / waypointCount;
+                innerRatio = Math.max(singleItemBase - 0.01, innerRatio);
+
+                progress = (itemTotalRatio + innerRatio) * 100;
+            }
+
+            String info = event.getEventType() == FlowExecutionEvent.EventType.NODE_STARTED ? "开始执行" : "执行完成";
+            Map<String, Object> pushMap = MapUtil.<String, Object>builder()
+                    .put("createTime", now)
+                    .put("insId", taskInstance.getId())
+                    .put("logInfo", "节点【" + event.getNodeName() + "】" + info)
+                    .put("status", targetStatus)
+                    .put("progress", String.format("%.2f", progress))
+                    .put("itemId", itemId)
+                    .put("taskId", taskInstance.getTaskId())
+                    .build();
+            try {
+                messagePushService.pushToChannel("InspectionTaskInstance", pushMap);
+            } catch (Exception e) {
+                log.info("消息推送异常，无订阅忽略,itemId:{}", itemId, e);
+            }
+        }
    }

-    @NotNull
-    private static InspectionTaskInstance getInspectionTaskInstance() {
-        InspectionTaskInstance updateEntity = InspectionTaskInstance.builder()
-                .endTime(DateUtils.getNowDate())
-                .build();
-        updateEntity.setUpdateTime(DateUtils.getNowDate());
-        return updateEntity;
+    /** 任务结束统一清理缓存 */
+    private void clearCache(String taskInsId) {
+        Set<String> itemList = taskItemExecutMap.getOrDefault(taskInsId, new HashSet<>());
+        itemList.forEach(itemNodeExecutMap::remove);
+        taskItemExecutMap.remove(taskInsId);
    }

-
-}
-
+    /**
+     * 推送完成任务信息（参数dbInsId为数据库主键ID）
+     */
+    public void pushCompleteMsg(String dbInsId, Date now, int targetStatus) {
+        Map<String, Object> pushMap = MapUtil.<String, Object>builder()
+                .put("createTime", now)
+                .put("insId", dbInsId)
+                .put("status", targetStatus)
+                .build();
+        try {
+            messagePushService.pushToChannel("InspectionTaskInstance", pushMap);
+        } catch (Exception e) {
+            log.info("消息推送异常，无订阅忽略,insId:{}", dbInsId, e);
+        }
+    }
+}
--- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionTaskWaypointMapper.java
+++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionTaskWaypointMapper.java
@ -50,4 +50,6 @@ public interface InspectionTaskWaypointMapper  extends MPJBaseMapper<InspectionT
     * @param waypointIds 点位ID数组
     */
    void deleteTaskWaypointByWaypointIds(String[] waypointIds);
+
+
 }
--- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionWaypointMapper.java
+++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionWaypointMapper.java
@ -52,4 +52,6 @@ public interface InspectionWaypointMapper  extends MPJBaseMapper<InspectionWaypo
     * @return 巡检点位视图集合
     */
    List<InspectionWaypointVo> selectInspectionWaypointVoList(InspectionWaypoint inspectionWaypoint);
+
+
 }
--- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionTaskInstanceServiceImpl.java
+++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionTaskInstanceServiceImpl.java
@ -88,6 +88,9 @@ public class InspectionTaskInstanceServiceImpl extends ServiceImpl<InspectionTas
        inspectionTaskInstance.setStatus(TaskStatusEnum.NOT_STARTED.getCode());
        inspectionTaskInstance.setCreateTime(DateUtils.getNowDate());
        inspectionTaskInstance.setCreateBy(SecurityUtils.getUsername());
+        // 获取点位数量
+        InspectionTask inspectionTask = inspectionTaskService.selectInspectionTaskById(inspectionTaskInstance.getTaskId());
+        inspectionTaskInstance.setWaypointCount(inspectionTask.getWaypointCount());
        return inspectionTaskInstanceMapper.insertInspectionTaskInstance(inspectionTaskInstance);
    }

@ -106,6 +109,9 @@ public class InspectionTaskInstanceServiceImpl extends ServiceImpl<InspectionTas
        }
        inspectionTaskInstance.setUpdateTime(DateUtils.getNowDate());
        inspectionTaskInstance.setUpdateBy(SecurityUtils.getUsername());
+        // 获取点位数量
+        InspectionTask inspectionTask = inspectionTaskService.selectInspectionTaskById(inspectionTaskInstance.getTaskId());
+        inspectionTaskInstance.setWaypointCount(inspectionTask.getWaypointCount());
        return inspectionTaskInstanceMapper.updateInspectionTaskInstance(inspectionTaskInstance);
    }

--- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionTaskServiceImpl.java
+++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionTaskServiceImpl.java
@ -5,7 +5,6 @@ import java.util.List;
 import java.util.UUID;
 import java.util.ArrayList;
 import java.util.stream.Collectors;
-
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.cmvr.inspection.domain.InspectionTaskWaypoint;
 import com.cmvr.inspection.domain.InspectionWaypoint;
@ -127,7 +126,7 @@ public class InspectionTaskServiceImpl extends ServiceImpl<InspectionTaskMapper,

    /**
     * 查询任务关联的点位列表
-     * 
+     *
     * @param taskId 任务ID
     * @return 点位集合
     */
@ -162,8 +161,10 @@ public class InspectionTaskServiceImpl extends ServiceImpl<InspectionTaskMapper,
            list.add(tw);
        }
        inspectionTaskWaypointMapper.batchInsertTaskWaypoint(list);
-        // 查询巡检任务
+        // 查询巡检任务,并修改点位数量
        InspectionTask inspectionTask = inspectionTaskMapper.selectInspectionTaskById(taskId);
+        inspectionTask.setWaypointCount(waypointIds.length);
+        inspectionTaskMapper.updateInspectionTask(inspectionTask);
        // 查询点位
        List<InspectionWaypoint> waypoints = inspectionTaskWaypointMapper.selectWaypointsByTaskId(taskId);
        // 同步任务编排
--- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionWaypointServiceImpl.java
+++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionWaypointServiceImpl.java
@ -2,7 +2,6 @@ package com.cmvr.inspection.service.impl;

 import java.util.List;
 import java.util.UUID;
-
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.cmvr.inspection.domain.vo.InspectionWaypointVo;
 import com.cmvr.inspection.mapper.InspectionTaskWaypointMapper;
@ -115,5 +114,4 @@ public class InspectionWaypointServiceImpl extends ServiceImpl<InspectionWaypoin
        }
        return inspectionWaypointMapper.deleteInspectionWaypointByIds(ids);
    }
-
 }
--- a/cmvr-iot-inspection/src/main/resources/mapper/inspection/InspectionTaskInstanceMapper.xml
+++ b/cmvr-iot-inspection/src/main/resources/mapper/inspection/InspectionTaskInstanceMapper.xml
@ -41,10 +41,11 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
        <result property="createByName"    column="create_by_name"    />
        <result property="updateByName"    column="update_by_name"    />
        <result property="taskInsId"    column="task_ins_id"    />
+        <result property="waypointCount"    column="waypoint_count"    />
    </resultMap>

    <sql id="selectInspectionTaskInstanceVo">
-        select id, create_by, create_time, update_by, update_time, remark, task_id, robot_id, status, start_time, end_time, task_ins_id from inspection_task_instance
+        select id, create_by, create_time, update_by, update_time, remark, task_id, robot_id, status, start_time, end_time, task_ins_id, waypoint_count from inspection_task_instance
    </sql>

    <select id="selectInspectionTaskInstanceList" parameterType="InspectionTaskInstance" resultMap="InspectionTaskInstanceResult">
@ -76,6 +77,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
            <if test="startTime != null">start_time,</if>
            <if test="endTime != null">end_time,</if>
            <if test="taskInsId != null">task_ins_id,</if>
+            <if test="waypointCount != null">waypoint_count,</if>
         </trim>
        <trim prefix="values (" suffix=")" suffixOverrides=",">
            <if test="id != null">#{id},</if>
@ -90,6 +92,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
            <if test="startTime != null">#{startTime},</if>
            <if test="endTime != null">#{endTime},</if>
            <if test="taskInsId != null">#{taskInsId},</if>
+            <if test="waypointCount != null">#{waypointCount},</if>
         </trim>
    </insert>

@ -107,6 +110,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
            <if test="startTime != null">start_time = #{startTime},</if>
            <if test="endTime != null">end_time = #{endTime},</if>
            <if test="taskInsId != null">task_ins_id = #{taskInsId},</if>
+            <if test="waypointCount != null">waypoint_count = #{waypointCount},</if>
        </trim>
        where id = #{id}
    </update>
@ -123,7 +127,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
    </delete>

    <select id="selectInspectionTaskInstanceVoList" parameterType="InspectionTaskInstanceQuery" resultMap="InspectionTaskInstanceVoResult">
-        select ti.id, ti.create_by, ti.create_time, ti.update_by, ti.update_time, ti.remark,
+        select ti.id, ti.create_by, ti.create_time, ti.update_by, ti.update_time, ti.remark,ti.waypoint_count,
        ti.task_id, t.task_name, t.task_code,
        ti.robot_id, r.robot_name,
        r.current_map_id as robot_current_map_id, m.map_name,
--- a/cmvr-iot-inspection/src/main/resources/mapper/inspection/InspectionTaskMapper.xml
+++ b/cmvr-iot-inspection/src/main/resources/mapper/inspection/InspectionTaskMapper.xml
@ -36,7 +36,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
    </resultMap>

    <sql id="selectInspectionTaskVo">
-        select id, create_by, create_time, update_by, update_time, remark, task_code, task_name, task_type, task_config_id, map_id from inspection_task
+        select id, create_by, create_time, update_by, update_time, remark, task_code, task_name, task_type, task_config_id, map_id, waypoint_count from inspection_task
    </sql>

    <select id="selectInspectionTaskList" parameterType="InspectionTask" resultMap="InspectionTaskResult">
@ -67,6 +67,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
            <if test="taskType != null">task_type,</if>
            <if test="taskConfigId != null">task_config_id,</if>
            <if test="mapId != null">map_id,</if>
+            <if test="waypointCount != null">waypoint_count,</if>
         </trim>
        <trim prefix="values (" suffix=")" suffixOverrides=",">
            <if test="id != null">#{id},</if>
@ -80,6 +81,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
            <if test="taskType != null">#{taskType},</if>
            <if test="taskConfigId != null">#{taskConfigId},</if>
            <if test="mapId != null">#{mapId},</if>
+            <if test="waypointCount != null">#{waypointCount},</if>
         </trim>
    </insert>

@ -96,6 +98,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
            <if test="taskType != null">task_type = #{taskType},</if>
            <if test="taskConfigId != null">task_config_id = #{taskConfigId},</if>
            <if test="mapId != null">map_id = #{mapId},</if>
+            <if test="waypointCount != null">waypoint_count = #{waypointCount},</if>
        </trim>
        where id = #{id}
    </update>
@ -113,7 +116,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"

    <select id="selectInspectionTaskVoList" parameterType="InspectionTask" resultMap="InspectionTaskVoResult">
        select t.id, t.create_by, t.create_time, t.update_by, t.update_time, t.remark,
-               t.task_code, t.task_name, t.task_type, t.task_config_id, t.map_id,
+               t.task_code, t.task_name, t.task_type, t.task_config_id, t.map_id, t.waypoint_count,
               m.map_name,
               u1.user_name as create_by_name, u2.user_name as update_by_name
        from inspection_task t
--- a/cmvr-iot-inspection/src/main/resources/mapper/inspection/InspectionWaypointMapper.xml
+++ b/cmvr-iot-inspection/src/main/resources/mapper/inspection/InspectionWaypointMapper.xml
@ -40,14 +40,6 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
        select id, create_by, create_time, update_by, update_time, remark, waypoint_name, enabled, sort_order, detect_item_id, map_id from inspection_waypoint
    </sql>

-    <select id="selectInspectionWaypointList" parameterType="InspectionWaypoint" resultMap="InspectionWaypointResult">
-        <include refid="selectInspectionWaypointVo"/>
-        <where>  
-            <if test="waypointName != null  and waypointName != ''"> and waypoint_name like concat('%', #{waypointName}, '%')</if>
-            <if test="enabled != null  and enabled != ''"> and enabled = #{enabled}</if>
-        </where>
-    </select>
-    
    <select id="selectInspectionWaypointById" parameterType="String" resultMap="InspectionWaypointResult">
        <include refid="selectInspectionWaypointVo"/>
        where id = #{id}
--- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionEvent.java
+++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionEvent.java
@ -32,6 +32,11 @@ public class FlowExecutionEvent {
     */
    private String taskId;

+    /**
+     * 节点数量
+     */
+    private Integer nodeCount;
+
    /**
     * 当前检测项 ID
     */
--- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowAfterInterceptor.java
+++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowAfterInterceptor.java
@ -40,6 +40,7 @@ public class FlowAfterInterceptor extends AbstractFlowMsgPreInterceptor{
                .taskId(message.getTaskId())
                .itemId(message.getItemId())
                .nodeId(message.getNodeId())
+                .nodeCount(message.getGraph().allNodeIds().size())
                .nodeType(message.getNodeType())
                .errorMessage(errorMessage)
                .build();
--- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowBeforeInterceptor.java
+++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowBeforeInterceptor.java
@ -27,6 +27,7 @@ public class FlowBeforeInterceptor extends AbstractFlowMsgPreInterceptor{
                .taskId(message.getTaskId())
                .itemId(message.getItemId())
                .nodeId(message.getNodeId())
+                .nodeCount(message.getGraph().allNodeIds().size())
                .nodeName(message.getNodeName())
                .nodeType(message.getNodeType())
                .build();
--- a/sql/inspection_module.sql
+++ b/sql/inspection_module.sql
@ -57,6 +57,7 @@ CREATE TABLE `inspection_task` (
  `task_type` char(1) DEFAULT '1' COMMENT '任务类型(1巡检任务 2讲解任务)',
  `task_config_id` varchar(64) DEFAULT NULL COMMENT '任务配置ID',
  `map_id` varchar(64) DEFAULT NULL COMMENT '地图ID',
+  `waypoint_count` int(11) DEFAULT '0' COMMENT '点位总数',
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_task_code` (`task_code`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='巡检任务表';
@ -104,6 +105,7 @@ CREATE TABLE `inspection_task_instance` (
  `start_time` datetime DEFAULT NULL COMMENT '开始时间',
  `end_time` datetime DEFAULT NULL COMMENT '结束时间',
  `task_ins_id` varchar(64) DEFAULT NULL COMMENT '检测任务id',
+  `waypoint_count` int(11) DEFAULT '0' COMMENT '点位总数',
  PRIMARY KEY (`id`),
  KEY `idx_task_id` (`task_id`),
  KEY `idx_robot_id` (`robot_id`),