SmartBench/CMVR-IOT
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 嵌套循环

Browse Source
This commit is contained in:
stream 2025-09-05 17:35:47 +08:00
parent d49c3bd0ab
commit fdc7f08929
14 changed files with 248 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowGraph.java
View File

@ -174,7 +174,6 @@ public class FlowGraph {
* 获取开始节点对象
*/
public FlowNodeWrapper getStartNode() {
System.out.println("这是什么线程：" + Thread.currentThread().getName());
String id = findStartNodeId();
return getNode(id);
}

6
cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskInstHolder.java
View File

@ -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<Integer> 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,

39
cmvr-iot-test/src/main/java/com/cmvr/test/flow/control/FlowControlService.java
View File

@ -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()
);
});
}

30
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowLoopNodeHandler.java
View File

@ -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<Integer> 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<Integer> newIterations = new ArrayList<>(parentIterations);
// newIterations.add(i);
//
// // 克隆 message，避免覆盖父级
// TaskNodeExecuteMessage subMessage = new TaskNodeExecuteMessage();
// BeanUtil.copyProperties(message, subMessage);
//// subMessage.setLoopNum(i); // 当前层循环次数
// subMessage.setIterations(newIterations); // 完整迭代路径
// 构造新的迭代路径
List<Integer> 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(); // 等待子图执行完毕后再进行下一轮

66
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java
View File

@ -53,12 +53,11 @@ public class FlowItemExecutor {
String startNodeId = graph.findStartNodeId();
List<FlowParamDef> 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<Integer> iterations) {
// 主图中开始节点的 inputParams
List<FlowParamDef> 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<FlowParamDef> startInputDefs,
TaskNodeExecuteMessage rootMessage,
Runnable onFinished,
int loopIteration) {
List<Integer> 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<Integer> 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<FlowParamDef> startInputDefs, TaskNodeExecuteMessage rootMessage, Runnable onFinished, int loopIteration, String nodeId, JSONObject inputParams, TaskNodeExecuteMessage message) {
private void postHandleBranchNode(FlowGraph graph, FlowNodeWrapper node, String startNodeId,
List<FlowParamDef> startInputDefs, TaskNodeExecuteMessage rootMessage, Runnable onFinished,
List<Integer> 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<FlowParamDef> startInputDefs, TaskNodeExecuteMessage rootMessage, Runnable onFinished, int loopIteration, String instId, String itemId, String nodeId) {
private void registerThread(FlowGraph graph, FlowNodeWrapper node, String startNodeId, List<FlowParamDef> startInputDefs,
TaskNodeExecuteMessage rootMessage, Runnable onFinished, List<Integer> 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);
}

69
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskScheduler.java
View File

@ -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<String> 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<Integer> iterations) {
String instId = context.getInstId();
FlowNodeWrapper startNode = graph.findSubStartNodeId();
List<String> 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<Integer> iterations) {
List<String> nextNodes = graph.getNextNodes(nodeId);
for (String nextId : nextNodes) {
String key = FlowSchedulerCache.buildKey(instId, nextId, iterations);
// 使用 Set 来记录已完成的前驱节点
Set<String> 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<Integer> 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<Integer> iterations) {
String key = FlowSchedulerCache.buildKey(instId, nodeId, loopIteration);
String key = FlowSchedulerCache.buildKey(instId, nodeId, iterations);
if (schedulerCache.trySchedule(key)) {
executor.submit(() -> {
try {

9
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowSchedulerCache.java
View File

@ -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<Integer> iterations) {
StringBuilder sb = new StringBuilder(instId).append(":").append(nodeId);
for (int it : iterations) {
sb.append(":").append(it);
}
return sb.toString();
}
}

5
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowLoggingInterceptor.java
View File

@ -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();

2
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/message/TaskNodeExecuteContext.java
View File

@ -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<Integer> iterations = new ArrayList<>();
}

56
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/message/TaskNodeExecuteMessage.java
View File

@ -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<Integer> 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;
}

4
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeCameraOperateService.java
View File

@ -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)) {

3
cmvr-iot-test/src/main/java/com/cmvr/test/model/domain/TeNodeInst.java
View File

@ -47,6 +47,9 @@ public class TeNodeInst {
@ApiModelProperty("执行日志/备注")
private String message;
@ApiModelProperty("循环节点的层级")
private String iteration;
@ApiModelProperty("开始时间戳（ms）")
private Long startTime;

21
cmvr-iot-test/src/main/java/com/cmvr/test/service/ITeNodeInstService.java
View File

@ -23,18 +23,19 @@ public interface ITeNodeInstService extends IService<TeNodeInst> {
/**
* 记录节点进入运行状态
*
* @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<Integer> iterations);
/**
* 节点执行后更新输出参数等信息/记录节点成功完成
@ -52,7 +53,7 @@ public interface ITeNodeInstService extends IService<TeNodeInst> {
/**
* 删除节点
*/
void delete(String instId, String nodeId);
void delete(String instId, String nodeId, List<Integer> iterations);
/**
* 更新节点

17
cmvr-iot-test/src/main/java/com/cmvr/test/service/impl/ITeNodeInstServiceImpl.java
View File

@ -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<TeNodeInstMapper, TeNode
@Override
public 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<Integer> iterations) {
TeNodeInst record = new TeNodeInst();
record.setInstId(instId);
@ -47,6 +48,7 @@ public class ITeNodeInstServiceImpl extends ServiceImpl<TeNodeInstMapper, TeNode
record.setParamsIn(paramsIn);
record.setParamsOut("{}");
record.setStartTime(System.currentTimeMillis());
record.setIteration(getIteration(iterations));
this.save(record);
}
@ -113,19 +115,26 @@ public class ITeNodeInstServiceImpl extends ServiceImpl<TeNodeInstMapper, TeNode
}
@Override
public void delete(String instId, String nodeId) {
public void delete(String instId, String nodeId, List<Integer> iterations) {
String iteration = getIteration(iterations);
LambdaQueryWrapper<TeNodeInst> 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<Integer> iterations) {
return StrUtil.join(StrUtil.DASHED, iterations);
}
@Override
public void update(String instId, String nodeId, TaskStatusEnum nodeStatus) {
LambdaUpdateWrapper<TeNodeInst> 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);
}
}