diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowGraph.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowGraph.java index ce4dfa3..ea3a24d 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowGraph.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowGraph.java @@ -174,7 +174,6 @@ public class FlowGraph { * 获取开始节点对象 */ public FlowNodeWrapper getStartNode() { - System.out.println("这是什么线程：" + Thread.currentThread().getName()); String id = findStartNodeId(); return getNode(id); } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskInstHolder.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskInstHolder.java index 0056c36..632137e 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskInstHolder.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskInstHolder.java @@ -10,6 +10,8 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import java.util.List; + /** * 任务实例状态协调器：负责同步日志记录、上下文状态、数据库状态 */ @@ -40,8 +42,8 @@ public class TaskInstHolder { public void markRunning(String instId, String taskId, String itemId, String nodeId, String nodeType, String operate, - String action, String paramsIn) { - nodeInstService.logRunning(instId, taskId, itemId, nodeId, nodeType, operate, action, paramsIn); + String action, String paramsIn, List iterations) { + nodeInstService.logRunning(instId, taskId, itemId, nodeId, nodeType, operate, action, paramsIn, iterations); } public void markSuccess(String instId, String itemId, String nodeId, int pendingItemCount, String nodeType, diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/control/FlowControlService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/control/FlowControlService.java index b597d2d..c12f004 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/control/FlowControlService.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/control/FlowControlService.java @@ -1,5 +1,6 @@ package com.cmvr.test.flow.control; +import cn.hutool.core.bean.BeanUtil; import com.cmvr.common.exception.GlobalException; import com.cmvr.test.enums.NodeTypeEnum; import com.cmvr.test.enums.TaskStatusEnum; @@ -10,6 +11,7 @@ import com.cmvr.test.flow.context.TaskInstHolder; import com.cmvr.test.flow.context.TaskThreadRegistry; import com.cmvr.test.flow.runtime.engine.FlowItemExecutor; import com.cmvr.test.flow.runtime.message.TaskNodeExecuteContext; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; import com.cmvr.test.service.ITeNodeInstService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -17,6 +19,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; import javax.annotation.Resource; +import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.stream.Collectors; @@ -118,7 +121,7 @@ public class FlowControlService { nodeInstService.update(instId, nodeId, TaskStatusEnum.RUNNING); continue; } else { - nodeInstService.delete(instId, nodeId); + nodeInstService.delete(instId, nodeId, pendingNode.getIterations()); } if (graph.isSub()) { // 找到子图对应的loop节点 @@ -135,11 +138,21 @@ public class FlowControlService { // 子图中断点，先执行中断的子图节点 executor.execute(() -> { CountDownLatch latch = taskThreadRegistry.getLatch(instId, loopCtx.getNode().getNodeId()); + + // 恢复时构造新的 message，带上 loopNum 和 iterations + TaskNodeExecuteMessage resumeMessage = new TaskNodeExecuteMessage(); + BeanUtil.copyProperties(pendingNode.getRootMessage(), resumeMessage); + resumeMessage.setLoopNum(pendingNode.getLoopIteration()); // 保留中断时的循环次数 + resumeMessage.setIterations(new ArrayList<>(pendingNode.getIterations())); // 保留中断时的路径 + flowItemExecutor.executeNode( - pendingNode.getGraph(), pendingNode.getNode(), - pendingNode.getStartNodeId(), pendingNode.getStartInputDefs(), - pendingNode.getRootMessage(), latch::countDown, - pendingNode.getLoopIteration() + pendingNode.getGraph(), + pendingNode.getNode(), + pendingNode.getStartNodeId(), + pendingNode.getStartInputDefs(), + resumeMessage, + latch::countDown, + pendingNode.getIterations() ); try { @@ -152,11 +165,19 @@ public class FlowControlService { } else { // 普通节点直接恢复 executor.execute(() -> { + TaskNodeExecuteMessage resumeMessage = new TaskNodeExecuteMessage(); + BeanUtil.copyProperties(pendingNode.getRootMessage(), resumeMessage); + resumeMessage.setLoopNum(pendingNode.getLoopIteration()); // 保留循环次数 + resumeMessage.setIterations(new ArrayList<>(pendingNode.getIterations())); // 保留路径 + flowItemExecutor.executeNode( - pendingNode.getGraph(), pendingNode.getNode(), - pendingNode.getStartNodeId(), pendingNode.getStartInputDefs(), - pendingNode.getRootMessage(), pendingNode.getOnFinished(), - pendingNode.getLoopIteration() + pendingNode.getGraph(), + pendingNode.getNode(), + pendingNode.getStartNodeId(), + pendingNode.getStartInputDefs(), + resumeMessage, + pendingNode.getOnFinished(), + pendingNode.getIterations() ); }); } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowLoopNodeHandler.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowLoopNodeHandler.java index 5bb9fcf..24e8838 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowLoopNodeHandler.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowLoopNodeHandler.java @@ -1,5 +1,6 @@ package com.cmvr.test.flow.runtime.dispatcher; +import cn.hutool.core.bean.BeanUtil; import com.cmvr.common.utils.spring.SpringUtils; import com.cmvr.test.flow.builder.FlowGraph; import com.cmvr.test.flow.builder.FlowNodeWrapper; @@ -11,6 +12,8 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.CountDownLatch; @Slf4j @@ -25,7 +28,9 @@ public class FlowLoopNodeHandler implements FlowNodeTypeHandler { String instId = message.getInstId(); FlowGraph graph = message.getGraph(); // 当前从第几次循环开始 - int loopNum = message.getLoopNum() < 0 ? 1 : message.getLoopNum(); + int loopNum = Math.max(message.getLoopNum(), 1); + // 父级迭代路径 + List parentIterations = message.getIterations(); String nodeId = message.getNodeId(); FlowNodeWrapper nodeWrapper = graph.getNode(nodeId); @@ -50,7 +55,28 @@ public class FlowLoopNodeHandler implements FlowNodeTypeHandler { taskThreadRegistry.registerLatch(instId, nodeId, latch); FlowItemExecutor flowItemExecutor = SpringUtils.getBean(FlowItemExecutor.class); - flowItemExecutor.executeSubGraph(subGraph, message, latch::countDown, i); + + // 构造新的迭代路径 +// List newIterations = new ArrayList<>(parentIterations); +// newIterations.add(i); +// +// // 克隆 message，避免覆盖父级 +// TaskNodeExecuteMessage subMessage = new TaskNodeExecuteMessage(); +// BeanUtil.copyProperties(message, subMessage); +//// subMessage.setLoopNum(i); // 当前层循环次数 +// subMessage.setIterations(newIterations); // 完整迭代路径 + + // 构造新的迭代路径 + List newIterations = new ArrayList<>(parentIterations); + newIterations.add(i); + + // 克隆 message，并明确设置 loopNum + TaskNodeExecuteMessage subMessage = new TaskNodeExecuteMessage(); + BeanUtil.copyProperties(message, subMessage); +// subMessage.setLoopNum(i); // 当前 loop 第 i 次 + subMessage.setIterations(newIterations); // 完整路径 + + flowItemExecutor.executeSubGraph(subGraph, subMessage, latch::countDown, newIterations); try { latch.await(); // 等待子图执行完毕后再进行下一轮 diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java index 2fd4838..b797aa3 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java @@ -53,12 +53,11 @@ public class FlowItemExecutor { String startNodeId = graph.findStartNodeId(); List startInputDefs = graph.getStartInputParamsDefs(); - - NodeExecutor executor = node -> - executeNode(graph, node, startNodeId, startInputDefs, rootMessage, onFinished, -1); - // 执行 start 节点（同步执行） - executeNode(graph, graph.getStartNode(), startNodeId, startInputDefs, rootMessage, onFinished, -1); + NodeExecutor executor = node -> + executeNode(graph, node, startNodeId, startInputDefs, rootMessage, onFinished, new ArrayList<>()); + + executeNode(graph, graph.getStartNode(), startNodeId, startInputDefs, rootMessage, onFinished, new ArrayList<>()); // 调度后续节点（并发） flowTaskScheduler.start(graph, taskInstHolder.getContext(rootMessage.getInstId()), executor, onFinished); @@ -70,7 +69,7 @@ public class FlowItemExecutor { public void executeSubGraph(FlowGraph subGraph, TaskNodeExecuteMessage rootMessage, Runnable onFinished, - int loopIteration) { + List iterations) { // 主图中开始节点的 inputParams List startInputDefs = subGraph.getStartInputParamsDefs(); @@ -81,15 +80,16 @@ public class FlowItemExecutor { NodeExecutor executor = node -> executeNode( subGraph, node, subStartNodeId, startInputDefs, - rootMessage, onFinished, loopIteration + rootMessage, onFinished, iterations ); // 执行子图起始节点 executeNode(subGraph, subStartNode, subStartNodeId, startInputDefs, - rootMessage, onFinished, loopIteration); + rootMessage, onFinished, iterations); // 启动子图调度 - flowTaskScheduler.subStart(subGraph, taskInstHolder.getContext(rootMessage.getInstId()), executor, onFinished, loopIteration); + flowTaskScheduler.subStart(subGraph, taskInstHolder.getContext(rootMessage.getInstId()), + executor, onFinished, iterations); } /** @@ -101,20 +101,22 @@ public class FlowItemExecutor { List startInputDefs, TaskNodeExecuteMessage rootMessage, Runnable onFinished, - int loopIteration) { + List iterations) { String instId = rootMessage.getInstId(); String itemId = rootMessage.getItemId(); String nodeId = node.getNodeId(); + log.info("当前执行的是 {} 节点，迭代路径={}", node.getNodeName(), iterations); //注册当前线程 - registerThread(graph, node, startNodeId, startInputDefs, rootMessage, onFinished, loopIteration, instId, itemId, nodeId); + registerThread(graph, node, startNodeId, startInputDefs, rootMessage, onFinished, iterations, instId, itemId, nodeId); + try { // 准备输入参数 JSONObject inputParams = paramPreparer.prepare(graph, node, startNodeId, startInputDefs, rootMessage); - // 创建带上下文的执行消息（关键：注入 graph） - TaskNodeExecuteMessage message = buildTaskNodeExecuteMsg(graph, node, rootMessage, nodeId, inputParams, loopIteration); + // 创建带上下文的执行消息（关键：注入 graph + iterations） + TaskNodeExecuteMessage message = buildTaskNodeExecuteMsg(graph, node, rootMessage, nodeId, inputParams, iterations); // 执行责任链 TaskNodeExecuteResult result = chainBuilder.build(interceptors, dispatchNodeHandler).apply(message); @@ -126,13 +128,14 @@ public class FlowItemExecutor { // 分支节点处理 if (node.getNodeType() == NodeTypeEnum.BRANCH) { - postHandleBranchNode(graph, node, startNodeId, startInputDefs, rootMessage, onFinished, loopIteration, nodeId, inputParams, message); + postHandleBranchNode(graph, node, startNodeId, startInputDefs, rootMessage, onFinished, iterations, nodeId, inputParams, message); return; } // END 节点 or 子图结束 if (node.getNodeType() == NodeTypeEnum.END || node.getNodeType() == NodeTypeEnum.SUB_END) { try { + log.info("{} 节点执行，释放线程", node.getNodeType()); onFinished.run(); } catch (Exception e) { log.error("onFinished 回调异常", e); @@ -140,22 +143,24 @@ public class FlowItemExecutor { } // 通知调度器继续调度后继节点 - flowTaskScheduler.markCompleted(message.getInstId(), nodeId, graph, n -> - executeNode(graph, n, startNodeId, startInputDefs, rootMessage, onFinished, loopIteration), loopIteration); + flowTaskScheduler.markCompleted( + message.getInstId(), nodeId, graph, + n -> executeNode(graph, n, startNodeId, startInputDefs, rootMessage, onFinished, iterations), + iterations + ); } finally { // 如果不是暂停全中断 TaskContext ctx = taskInstHolder.getContext(instId); if (ctx == null || !ctx.isPaused()) { taskThreadRegistry.unregisterNodeContext(instId, itemId, nodeId); - if (node.getNodeType().equals(NodeTypeEnum.SUB_START)) { - log.info("subStart注銷了"); - } } } } @NotNull - private static TaskNodeExecuteMessage buildTaskNodeExecuteMsg(FlowGraph graph, FlowNodeWrapper node, TaskNodeExecuteMessage rootMessage, String nodeId, JSONObject inputParams, int loopIteration) { + private static TaskNodeExecuteMessage buildTaskNodeExecuteMsg(FlowGraph graph, FlowNodeWrapper node, + TaskNodeExecuteMessage rootMessage, String nodeId, + JSONObject inputParams, List iterations) { TaskNodeExecuteMessage message = new TaskNodeExecuteMessage(); BeanUtil.copyProperties(rootMessage, message); message.setNodeId(nodeId); @@ -163,11 +168,18 @@ public class FlowItemExecutor { message.setAction(node.getAction()); message.setInputParams(inputParams); message.setGraph(graph); - message.setLoopNum(loopIteration); + // 只设置 iterations + message.setIterations(new ArrayList<>(iterations)); + + // loopNum 不在这里计算，而是由 FlowLoopNodeHandler / resume 显式写入 + // 如果 rootMessage 里已经带了 loopNum，就沿用它 + message.setLoopNum(rootMessage.getLoopNum()); return message; } - private void postHandleBranchNode(FlowGraph graph, FlowNodeWrapper node, String startNodeId, List startInputDefs, TaskNodeExecuteMessage rootMessage, Runnable onFinished, int loopIteration, String nodeId, JSONObject inputParams, TaskNodeExecuteMessage message) { + private void postHandleBranchNode(FlowGraph graph, FlowNodeWrapper node, String startNodeId, + List startInputDefs, TaskNodeExecuteMessage rootMessage, Runnable onFinished, + List iterations, String nodeId, JSONObject inputParams, TaskNodeExecuteMessage message) { String matchedAnchorId = FlowBranchEvaluator.evaluate(nodeId, node.getBranchConditions(), inputParams); FlowEdge matchedEdge = graph.getEdgeByFromAnchorId(matchedAnchorId); @@ -190,12 +202,13 @@ public class FlowItemExecutor { message.getInstId(), matchedEdge.getTo(), graph, - n -> executeNode(graph, n, startNodeId, startInputDefs, rootMessage, onFinished, loopIteration), - loopIteration + n -> executeNode(graph, n, startNodeId, startInputDefs, rootMessage, onFinished, iterations), + iterations ); } - private void registerThread(FlowGraph graph, FlowNodeWrapper node, String startNodeId, List startInputDefs, TaskNodeExecuteMessage rootMessage, Runnable onFinished, int loopIteration, String instId, String itemId, String nodeId) { + private void registerThread(FlowGraph graph, FlowNodeWrapper node, String startNodeId, List startInputDefs, + TaskNodeExecuteMessage rootMessage, Runnable onFinished, List iterations, String instId, String itemId, String nodeId) { TaskNodeExecuteContext nodeExecuteContext = new TaskNodeExecuteContext(); nodeExecuteContext.setThread(Thread.currentThread()); nodeExecuteContext.setGraph(graph); @@ -204,7 +217,8 @@ public class FlowItemExecutor { nodeExecuteContext.setNode(node); nodeExecuteContext.setOnFinished(onFinished); nodeExecuteContext.setRootMessage(rootMessage); - nodeExecuteContext.setLoopIteration(loopIteration); +// nodeExecuteContext.setLoopIteration(loopIteration); + nodeExecuteContext.setIterations(new ArrayList<>(iterations)); taskThreadRegistry.registerNodeContext(instId, itemId, nodeId, nodeExecuteContext); } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskScheduler.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskScheduler.java index 32fe23c..56e5a6b 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskScheduler.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskScheduler.java @@ -11,6 +11,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; import javax.annotation.Resource; +import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Set; @@ -32,7 +33,6 @@ public class FlowTaskScheduler { TaskContext context, FlowItemExecutor.NodeExecutor executorFunc, Runnable onFinished) { - String instId = context.getInstId(); String startNodeId = graph.findStartNodeId(); List nextNodes = graph.getNextNodes(startNodeId); @@ -44,7 +44,7 @@ public class FlowTaskScheduler { } for (String nextId : nextNodes) { - scheduleNode(instId, nextId, graph, executorFunc, -1); + scheduleNode(instId, nextId, graph, executorFunc, new ArrayList<>()); // 空 iterations } } @@ -55,8 +55,7 @@ public class FlowTaskScheduler { TaskContext context, FlowItemExecutor.NodeExecutor executorFunc, Runnable onFinished, - int loopIteration) { - + List iterations) { String instId = context.getInstId(); FlowNodeWrapper startNode = graph.findSubStartNodeId(); List nextNodes = graph.getNextNodes(startNode.getNodeId()); @@ -67,16 +66,15 @@ public class FlowTaskScheduler { return; } - // 并发调度后继节点 for (String nextId : nextNodes) { - scheduleNode(instId, nextId, graph, executorFunc, loopIteration); + scheduleNode(instId, nextId, graph, executorFunc, iterations); } } /** * 某节点执行完成后，判断并调度其所有后继节点 */ - public void markCompleted(String instId, +/* public void markCompleted(String instId, String nodeId, FlowGraph graph, FlowItemExecutor.NodeExecutor executorFunc, @@ -99,12 +97,38 @@ public class FlowTaskScheduler { scheduleNode(instId, nextId, graph, executorFunc, loopIteration); } } + }*/ + /** + * 某节点执行完成后，判断并调度其所有后继节点 + */ + public void markCompleted(String instId, + String nodeId, + FlowGraph graph, + FlowItemExecutor.NodeExecutor executorFunc, + List iterations) { + List nextNodes = graph.getNextNodes(nodeId); + + for (String nextId : nextNodes) { + String key = FlowSchedulerCache.buildKey(instId, nextId, iterations); + + // 使用 Set 来记录已完成的前驱节点 + Set completedPreNodes = schedulerCache.getOrInitCompletedPre(key); + // 添加当前节点作为已完成的前驱节点 + completedPreNodes.add(nodeId); + + int preCount = graph.getPredecessors().getOrDefault(nextId, Collections.emptyList()).size(); + + // 如果当前节点的所有前驱节点都已完成，则调度后继节点 + if (completedPreNodes.size() == preCount) { + scheduleNode(instId, nextId, graph, executorFunc, iterations); + } + } } /** * 提交节点执行任务（只调度一次） */ - private void scheduleNode(String instId, +/* private void scheduleNode(String instId, String nodeId, FlowGraph graph, FlowItemExecutor.NodeExecutor executorFunc, @@ -112,6 +136,31 @@ public class FlowTaskScheduler { String key = FlowSchedulerCache.buildKey(instId, nodeId, loopIteration); + // 如果已经调度过，则不再重复调度 + if (schedulerCache.trySchedule(key)) { + executor.submit(() -> { + Thread.currentThread().setName("node-thread-" + nodeId); + try { + executorFunc.execute(graph.getNode(nodeId)); + } catch (Exception e) { + if (ExceptionUtil.getRootCause(e) instanceof InterruptedException) { + return; + } + log.error("节点执行异常：instId={}, nodeId={}, err={}", instId, nodeId, e.getMessage(), e); + } + }); + } + }*/ + /** + * 提交节点执行任务（除循环只调度一次） + */ + private void scheduleNode(String instId, + String nodeId, + FlowGraph graph, + FlowItemExecutor.NodeExecutor executorFunc, + List iterations) { + String key = FlowSchedulerCache.buildKey(instId, nodeId, iterations); + // 如果已经调度过，则不再重复调度 if (schedulerCache.trySchedule(key)) { executor.submit(() -> { @@ -132,9 +181,9 @@ public class FlowTaskScheduler { String nodeId, FlowGraph graph, FlowItemExecutor.NodeExecutor executorFunc, - int loopIteration) { + List iterations) { - String key = FlowSchedulerCache.buildKey(instId, nodeId, loopIteration); + String key = FlowSchedulerCache.buildKey(instId, nodeId, iterations); if (schedulerCache.trySchedule(key)) { executor.submit(() -> { try { diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowSchedulerCache.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowSchedulerCache.java index 340f0fb..4b08217 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowSchedulerCache.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowSchedulerCache.java @@ -3,6 +3,7 @@ package com.cmvr.test.flow.runtime.engine.support; import cn.hutool.core.util.StrUtil; import org.springframework.stereotype.Component; +import java.util.List; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -34,4 +35,12 @@ public class FlowSchedulerCache { ? StrUtil.format("{}_{}_{}", instId, nodeId, loopIteration) : StrUtil.format("{}_{}", instId, nodeId); } + + public static String buildKey(String instId, String nodeId, List iterations) { + StringBuilder sb = new StringBuilder(instId).append(":").append(nodeId); + for (int it : iterations) { + sb.append(":").append(it); + } + return sb.toString(); + } } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowLoggingInterceptor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowLoggingInterceptor.java index f71eb9a..f568a19 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowLoggingInterceptor.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowLoggingInterceptor.java @@ -30,7 +30,8 @@ public class FlowLoggingInterceptor implements FlowMsgPreInterceptor { // todo 判断一下是否为LOOP重入 if (!(message.getNodeType().equalsIgnoreCase(NodeTypeEnum.LOOP.getCode()) && message.getLoopNum() > 0)) { taskInstHolder.markRunning(message.getInstId(), message.getTaskId(), message.getItemId(), message.getNodeId(), - message.getNodeType(), message.getAction().getOperate(), message.getAction().getAction(), message.getInputParams().toJSONString()); + message.getNodeType(), message.getAction().getOperate(), message.getAction().getAction(), + message.getInputParams().toJSONString(), message.getIterations()); } TaskNodeExecuteResult result = next.apply(message); @@ -43,7 +44,7 @@ public class FlowLoggingInterceptor implements FlowMsgPreInterceptor { return null; } - taskInstHolder.markSuccess(message.getInstId(), message.getItemId(), message.getNodeId(), message.getPendingItemCount(),message.getNodeType(), + taskInstHolder.markSuccess(message.getInstId(), message.getItemId(), message.getNodeId(), message.getPendingItemCount(), message.getNodeType(), result.getOutputParams().toJSONString(), StrUtil.format("[{}] 执行成功", message.getAction())); long end = System.currentTimeMillis(); diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/message/TaskNodeExecuteContext.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/message/TaskNodeExecuteContext.java index 0e56a6c..5c2f57c 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/message/TaskNodeExecuteContext.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/message/TaskNodeExecuteContext.java @@ -5,6 +5,7 @@ import com.cmvr.test.flow.builder.FlowNodeWrapper; import com.cmvr.test.flow.builder.FlowParamDef; import lombok.Data; +import java.util.ArrayList; import java.util.List; /** @@ -21,4 +22,5 @@ public class TaskNodeExecuteContext { private TaskNodeExecuteMessage rootMessage; private Runnable onFinished; private int loopIteration; + private List iterations = new ArrayList<>(); } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/message/TaskNodeExecuteMessage.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/message/TaskNodeExecuteMessage.java index 5d61754..1d81d69 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/message/TaskNodeExecuteMessage.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/message/TaskNodeExecuteMessage.java @@ -5,25 +5,38 @@ import com.cmvr.test.enums.ActionEnum; import com.cmvr.test.flow.builder.FlowGraph; import lombok.Data; +import java.util.ArrayList; +import java.util.List; + /** * 单个节点的执行输入消息 */ @Data public class TaskNodeExecuteMessage { - /** 实例ID */ + /** + * 实例ID + */ private String instId; - /** 任务ID */ + /** + * 任务ID + */ private String taskId; - /** 检测项ID（用于定位流程） */ + /** + * 检测项ID（用于定位流程） + */ private String itemId; - /** 节点ID */ + /** + * 节点ID + */ private String nodeId; - /** 待执行检测项数 */ + /** + * 待执行检测项数 + */ private int pendingItemCount; /** @@ -37,25 +50,44 @@ public class TaskNodeExecuteMessage { */ private transient int loopNum = 0; - /** 节点名称 */ + /** + * 全局迭代路径（嵌套循环时用） + */ + private List iterations = new ArrayList<>(); + + /** + * 节点名称 + */ private String nodeName; - /** 节点类型（START、FUNCTION、END 等） */ + /** + * 节点类型（START、FUNCTION、END 等） + */ private String nodeType; - /** 节点行为（动作标识符） */ + /** + * 节点行为（动作标识符） + */ private ActionEnum action; - /** 是否试运行 */ + /** + * 是否试运行 + */ private boolean trial; - /** 所属终端ID */ + /** + * 所属终端ID + */ private String terminalId; - /** 节点入参 */ + /** + * 节点入参 + */ private JSONObject inputParams; - /** 上游输出参数（来自上一个节点的输出） */ + /** + * 上游输出参数（来自上一个节点的输出） + */ private JSONObject upStreamOutput; } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeCameraOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeCameraOperateService.java index 09b6964..fb3f3d1 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeCameraOperateService.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeCameraOperateService.java @@ -64,8 +64,8 @@ public class EdgeCameraOperateService implements EdgeOperateService { // // 驾驶偏好 // imageUrl = "http://192.168.1.100:9000/cmvr-iot/FILE/20250822/1755849654731.jpg"; // } -// String imageUrl = "http://192.168.1.100:9000/cmvr-iot/IMAGE/20250901/1756718594725.jpg"; - String imageUrl = edgeCameraService.getRGBDImages(terminalId, deviceId); + String imageUrl = "http://192.168.1.100:9000/cmvr-iot/IMAGE/20250901/1756718594725.jpg"; +// String imageUrl = edgeCameraService.getRGBDImages(terminalId, deviceId); JSONArray imageUrls = new JSONArray(); if (ObjUtil.isNotEmpty(upstreamOutput)) { diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/model/domain/TeNodeInst.java b/cmvr-iot-test/src/main/java/com/cmvr/test/model/domain/TeNodeInst.java index 1be4ad1..7b8034b 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/model/domain/TeNodeInst.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/model/domain/TeNodeInst.java @@ -47,6 +47,9 @@ public class TeNodeInst { @ApiModelProperty("执行日志/备注") private String message; + @ApiModelProperty("循环节点的层级") + private String iteration; + @ApiModelProperty("开始时间戳（ms）") private Long startTime; diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/service/ITeNodeInstService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/service/ITeNodeInstService.java index 3117b0c..5ac529a 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/service/ITeNodeInstService.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/service/ITeNodeInstService.java @@ -23,18 +23,19 @@ public interface ITeNodeInstService extends IService { /** * 记录节点进入运行状态 * - * @param instId 任务实例 ID - * @param taskId 任务定义 ID - * @param itemId 检测项 ID - * @param nodeId 节点 ID - * @param nodeType 节点类型 - * @param operate 执行来源 - * @param action 动作类型 - * @param paramsIn 输入参数（已序列化为 JSON 字符串） + * @param instId 任务实例 ID + * @param taskId 任务定义 ID + * @param itemId 检测项 ID + * @param nodeId 节点 ID + * @param nodeType 节点类型 + * @param operate 执行来源 + * @param action 动作类型 + * @param paramsIn 输入参数（已序列化为 JSON 字符串） + * @param iterations 迭代次数 */ void logRunning(String instId, String taskId, String itemId, String nodeId, String nodeType, - String operate, String action, String paramsIn); + String operate, String action, String paramsIn, List iterations); /** * 节点执行后更新输出参数等信息/记录节点成功完成 @@ -52,7 +53,7 @@ public interface ITeNodeInstService extends IService { /** * 删除节点 */ - void delete(String instId, String nodeId); + void delete(String instId, String nodeId, List iterations); /** * 更新节点 diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/service/impl/ITeNodeInstServiceImpl.java b/cmvr-iot-test/src/main/java/com/cmvr/test/service/impl/ITeNodeInstServiceImpl.java index f88e52b..28c8d2d 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/service/impl/ITeNodeInstServiceImpl.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/service/impl/ITeNodeInstServiceImpl.java @@ -12,6 +12,7 @@ import com.cmvr.test.model.vo.TeFlowViewVO; import com.cmvr.test.service.ITeNodeInstService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.jetbrains.annotations.Nullable; import org.springframework.stereotype.Service; import java.util.List; @@ -33,7 +34,7 @@ public class ITeNodeInstServiceImpl extends ServiceImpl iterations) { TeNodeInst record = new TeNodeInst(); record.setInstId(instId); @@ -47,6 +48,7 @@ public class ITeNodeInstServiceImpl extends ServiceImpl iterations) { + String iteration = getIteration(iterations); LambdaQueryWrapper wrapper = Wrappers.lambdaQuery(); wrapper.eq(TeNodeInst::getInstId, instId) - .eq(TeNodeInst::getNodeId, nodeId); + .eq(TeNodeInst::getNodeId, nodeId) + .eq(TeNodeInst::getIteration, iteration); this.baseMapper.delete(wrapper); } + @Nullable + private static String getIteration(List iterations) { + return StrUtil.join(StrUtil.DASHED, iterations); + } + @Override public void update(String instId, String nodeId, TaskStatusEnum nodeStatus) { LambdaUpdateWrapper wrapper = Wrappers.lambdaUpdate(); wrapper.eq(TeNodeInst::getInstId, instId) .eq(TeNodeInst::getNodeId, nodeId) - .set(TeNodeInst::getStatus,nodeStatus.name()); + .set(TeNodeInst::getStatus, nodeStatus.name()); this.update(wrapper); } }