此方案是我工作中遇到的一个标准运维的一个需求
可以在web页面编辑流程节点 ，节点类型，支持拖拽等等，节点可以选择脚本，插件等等功能
然后通过建任务记录（record表），关联某个模板 （task_template表）
只做简单实现记录，代码是执行不了的，方便以后面试的时候问起来，省得记不起来

LogicFlow是什么
官网：http://logicflow.cn/tutorial/about
LogicFlow是一款流程图编辑框架，提供了一系列流程图交互、编辑所必需的功能和灵活的节点自定义、插件等拓展机制。 LogicFlow支持前端自定义开发各种逻辑编排场景，如流程图、ER图、BPMN流程等。在工作审批流配置、机器人逻辑编排、无代码平台流程配置都有较好的应用。
特性
可视化模型：通过 LogicFlow 提供的直观可视化界面，用户可以轻松创建、编辑和管理复杂的逻辑流程图。
高可定制性：用户可以根据自己的需要定制节点、连接器和样式，创建符合特定用例的定制逻辑流程图。
自执行引擎：执行引擎支持浏览器端执行流程图逻辑，为无代码执行提供新思路。

//任务模板json，此json通过前端logicflow可以图形化

{
    "nodes": [
        {
            "id": "start-node",
            "type": "start",
            "x": 150,
            "y": 100,
            "text": "开始",
            "properties": {}
        },
        {
            "id": "task-node-01",
            "type": "task",
            "x": 150,
            "y": 250,
            "text": "数据库备份",
            "properties": {
                "type": "script",
                "scriptId": "9cf678b4-9108-454e-8b21-67b53a7b6dc0",
                "scriptName": "ssh测试ping",
                "scriptOrigin": 2,
                "scriptVersion": "1.0.0",
                "scriptType": 2,  
                "scriptTypeDesc": "用于管理开源应用的脚本",  
                "inputParameters": [
                    {
                        "parameterName": "env",
                        "parameterValue": "SIT",
                        "isRequired": true
                    }
                ],
                "timeout": 1800,
                "status": "fail"
            }
        },
        {
            "id": "task-node-02",
            "type": "task",
            "x": 400,
            "y": 250,
            "text": "发送失败通知",
            "properties": {
                "type": "email",
                "pluginCode": "1001",
                "pluginName": "通知",
                "pluginVersion": "v1.0",
                "pluginDescription": "用于发送流程节点执行结果通知",
                "nodeName": "发送通知",
                "status": "success",
                "noticeMethod": "email,meixin",
                "noticeUsers": "ex_xuyy15,zhangly97",
                "noticeSubject": "通知主题【告警】数据库备份任务执行失败",
                "noticeContent": "您好！失败了啊"
            }
        },
        {
            "id": "end-node",
            "type": "end",
            "x": 275,
            "y": 400,
            "text": "流程结束",
            "properties": {
                "description": "接收备份成功分支或通知节点流转，流程最终终止"
            }
        }
    ],
    "edges": [
        {
            "id": "edge-start-to-backup",
            "sourceNodeId": "start-node",
            "targetNodeId": "task-node-01",
            "text": "",
            "properties": {
                "description": "流程开始后，执行数据库备份任务"
            }
        },
        {
            "id": "edge-backup-to-end-success",
            "sourceNodeId": "task-node-01",
            "targetNodeId": "end-node",
            "text": "备份成功",
            "properties": {
                "condition": "${taskResult_task-node-01 === 'true'}",
                "description": "备份成功时，直接流向唯一结束节点"
            }
        },
        {
            "id": "edge-backup-to-notify",
            "sourceNodeId": "task-node-01",
            "targetNodeId": "task-node-02",
            "text": "备份失败",
            "properties": {
                "condition": "${taskResult_task-node-01 === 'false'}",
                "description": "备份失败时，先触发发送失败通知"
            }
        },
        {
            "id": "edge-notify-to-end",
            "sourceNodeId": "task-node-02",
            "targetNodeId": "end-node",
            "text": "",
            "properties": {
                "description": "失败通知发送完成后，流向唯一结束节点"
            }
        }
    ],
    "properties": {
        "flowId": "db-backup-flow-001",
        "flowName": "数据库自动备份流程（单结束节点版）",
        "version": "1.0.1",
        "description": "每日凌晨2点执行数据库备份：成功则直接结束，失败则先发通知再结束，所有分支最终汇聚到同一个结束节点"
    }
}

/**
 * 流程边模型
 * 定义节点之间的连接关系，支持条件判断
 */
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class Edge {
    private String id;                 // 边ID
    private String sourceNodeId;       // 源节点ID（从哪个节点出发）
    private String targetNodeId;       // 目标节点ID（到哪个节点）
    private String text;               // 边上显示的文本
    private Map<String, Object> properties;  // 边的属性，可包含condition条件表达式
}

/**
 * 流程节点模型
 * 支持所有节点类型：开始、任务、结束、并行网关、条件并行网关、分支网关、汇聚网关
 */
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class Node {
    private String id;         // 节点ID
    private String type;       // 节点类型：start/task/end/parallelGateway/conditionalParallelGateway/branchGateway/convergeGateway
    private Integer x;         // x坐标
    private Integer y;         // y坐标
    private String text;       // 节点显示文本
    private Map<String, Object> properties;  // 节点自定义属性

    // 节点类型常量定义
    public static final String TYPE_START = "start";  // 开始节点
    public static final String TYPE_TASK = "task";    // 任务节点
    public static final String TYPE_END = "end";      // 结束节点

    // 网关类型
    public static final String TYPE_PARALLEL_GATEWAY = "parallelGateway";  // 并行网关
    public static final String TYPE_CONDITIONAL_PARALLEL_GATEWAY = "conditionalParallelGateway";  // 条件并行网关
    public static final String TYPE_BRANCH_GATEWAY = "branchGateway";      // 分支网关
    public static final String TYPE_CONVERGE_GATEWAY = "convergeGateway";  // 汇聚网关
}

/**
 * 流程上下文
 * 用于在节点之间传递数据、存储中间结果和跟踪流程状态
 */
@Data
public class WorkflowContext {
    // 存储流程变量（节点间共享的数据）
    private Map<String, Object> variables = new HashMap<>();
    // 标记流程是否已结束
    private boolean processEnded = false;
    // 存储等待中的网关（用于汇聚网关）
    private Set<String> pendingGateways = new HashSet<>();
    // 脚本要执行的目标ip
    private List<String> executeIpList = new ArrayList<>();
    // 用于关联记录表和执行情况和ip详细执行情况
    private Long opsTaskExecutionInfoId;
    // 任务id
    private String taskId;
    // 任务记录
    private OpsTaskRecordPO opsTaskRecordPO;
    // 新增：并行分支完成状态跟踪（key=分支标识，value=是否完成）
    // 用ConcurrentHashMap保证多线程并行分支操作安全
    private Map<String, Boolean> branchCompletionStatus = new ConcurrentHashMap<>();

    // 线程安全的分支完成状态存储（汇聚网关依赖此判断）
    private final Map<String, Boolean> branchCompletedMap = new ConcurrentHashMap<>();


    /**
     * 设置变量
     */
    public void setVariable(String key, Object value) {
        variables.put(key, value);
    }

    /**
     * 获取变量
     */
    public Object getVariable(String key) {
        return variables.get(key);
    }

    /**
     * 添加等待中的网关
     */
    public void addPendingGateway(String gatewayId) {
        pendingGateways.add(gatewayId);
    }

    /**
     * 创建上下文副本（用于并行网关的分支）
     */
    public WorkflowContext copy() {
        WorkflowContext copy = new WorkflowContext();
        copy.setVariables(new HashMap<>(this.variables));  // 浅拷贝变量
        copy.setProcessEnded(this.processEnded);
        return copy;
    }

    /**
     * 标记某个分支已完成（并行分支执行完后调用）
     * @param branchKey 分支标识（用并行网关出口边的sourceNodeId，确保唯一）
     */
    public void markBranchCompleted(String branchKey) {
        branchCompletedMap.put(branchKey, Boolean.TRUE);
    }

    /**
     * 判断某个分支是否完成
     */
    public boolean isBranchCompleted(String branchKey) {
        return Boolean.TRUE.equals(branchCompletedMap.get(branchKey));
    }

}

/**
 * 流程定义模型
 * 包含整个流程的所有节点、边和全局属性
 */
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class WorkflowDefinition {
    private List<Node> nodes;          // 所有节点
    private List<Edge> edges;          // 所有边
    private Map<String, Object> properties;  // 流程全局属性（如flowId、flowName等）
}

// 任务表
@Data
@TableName(value = "ops_task_record")
public class OpsTaskRecordPO {

    /**
     * 周期任务名称
     */
    @TableField(value = "task_name")
    private String taskName;

    /**
     * 任务模板主键id
     */
    @TableField(value = "ops_task_template_id")
    private Long opsTaskTemplateId;

    /**
     * 任务模板名称
     */
    @TableField(value = "ops_task_template_name")
    private String opsTaskTemplateName;

    /**
     * 任务模板配置（JSON格式）
     */
    @TableField(value = "ops_task_template_flow_json")
    private String opsTaskTemplateFlowJson;

    /**
     * 任务类型 周期任务，定时任务
     */
    @TableField(value = "ops_task_type")
    private String opsTaskType;

    /**
     * 执行类型：审批后立即执行/周期执行/定时执行
     */
    @TableField(value = "execute_type")
    private String executeType;

    /**
     * 执行开始时间（yyyy-MM-dd HH:mm:ss）
     */
    @TableField(value = "execute_start_time")
    @JsonFormat(shape = JsonFormat.Shape.STRING, pattern="yyyy-MM-dd HH:mm:ss" ,timezone = "GMT+8" )
    @JsonDeserialize(using = DateDeserializers.DateDeserializer.class)
    private Date executeStartTime;

    /**
     * 执行结束时间（yyyy-MM-dd HH:mm:ss）
     */
    @TableField(value = "execute_end_time")
    @JsonFormat(shape = JsonFormat.Shape.STRING, pattern="yyyy-MM-dd HH:mm:ss" ,timezone = "GMT+8" )
    @JsonDeserialize(using = DateDeserializers.DateDeserializer.class)
    private Date executeEndTime;

    /**
     * 任务状态（0-未启用 1-运行中 2-已暂停 3-已结束）
     */
    @TableField(value = "execute_status")
    private OpsTaskStatusEnum executeStatus;

    /**
     * 管理员
     */
    @TableField(value = "administrator")
    private String administrator;

    /**
     * 首次开始执行时间（yyyy-MM-dd HH:mm:ss）
     */
    @TableField(value = "first_execute_time")
    @JsonFormat(shape = JsonFormat.Shape.STRING, pattern="yyyy-MM-dd HH:mm:ss" ,timezone = "GMT+8" )
    @JsonDeserialize(using = DateDeserializers.DateDeserializer.class)
    private Date firstExecuteTime;

    /**
     * 最后一次执行开始时间（yyyy-MM-dd HH:mm:ss）
     */
    @TableField(value = "last_execute_time")
    @JsonFormat(shape = JsonFormat.Shape.STRING, pattern="yyyy-MM-dd HH:mm:ss" ,timezone = "GMT+8" )
    @JsonDeserialize(using = DateDeserializers.DateDeserializer.class)
    private Date lastExecuteTime;

    /**
     * 任务周期（单位：天/周/月，如：1D/2W/1M）
     */
    @TableField(value = "task_cycle")
    private String taskCycle;

    /**
     * 任务周期 cron
     */
    @TableField(value = "task_cycle_cron")
    private String taskCycleCron;

    /**
     * 已执行周期数
     */
    @TableField(value = "executed_cycles")
    private Integer executedCycles;
}

/**
 * 任务处理器接口
 * 不同类型的任务节点实现此接口
 */
public interface TaskProcessor {
    /**
     * 执行任务
     * @param taskNode 任务节点
     * @param context 流程上下文
     * @return 任务执行结果
     */
    TaskResult execute(Node taskNode, WorkflowContext context) throws Exception;
}

/**
 * 通知任务处理器
 * 处理"发送通知"类型的任务节点
 */
@Component
public class EmailProcessor implements TaskProcessor {
 	@Override
    public TaskResult execute(Node taskNode, WorkflowContext context) throws Exception {

	}

}

/**
 * 脚本任务处理器
 * 
 */
@Component
public class ScriptTaskProcessor implements TaskProcessor {
	@Override
    public TaskResult execute(Node taskNode, WorkflowContext context) throws Exception {

	}

}

/**
 * 流程引擎核心类
 * 处理所有节点类型：开始、任务、结束、各种网关
 */
@Component
public class WorkflowEngine {

    private static final Logger log = LoggerFactory.getLogger(WorkflowEngine.class);

    private static final long PARALLEL_BRANCH_TIMEOUT = 30; // 并行分支超时时间（分钟）
    private static final TimeUnit PARALLEL_TIMEOUT_UNIT = TimeUnit.MINUTES;

    @Autowired
    private TaskProcessorFactory taskProcessorFactory;

    @Resource(name = "opsTaskExecutor")
    ThreadPoolExecutor opsTaskExecutor;

    /**
     * 执行流程
     */
    public void execute(String workflowJson, WorkflowContext context) {
        // 校验入参
        if (workflowJson == null || context == null || context.getOpsTaskRecordPO() == null) {
            throw new IllegalArgumentException("流程JSON、上下文或任务记录不能为空");
        }

        OpsTaskRecordPO opsTaskRecordPO = context.getOpsTaskRecordPO();
        OpsTaskExecutionInfoPO opsTaskExecutionInfoPO = null;

        try {
            // 1. 解析流程定义
            WorkflowDefinition workflow = JSON.parseObject(workflowJson, WorkflowDefinition.class);
            if (workflow.getNodes() == null || workflow.getNodes().isEmpty()) {
                throw new RuntimeException("流程定义中无节点配置");
            }

            // 2. 查找开始节点
            Node startNode = workflow.getNodes().stream()
                    .filter(node -> Node.TYPE_START.equals(node.getType()))
                    .findFirst()
                    .orElseThrow(() -> new RuntimeException("流程定义中未找到开始节点"));

            // 3. 创建执行记录（首次执行）
            // 先有任务，再有执行记录，所以要根据recordId去查执行明细表
            Integer maxExecutedCycles = opsTaskExecutionInfoMapper.getMaxExecutedCycles(opsTaskRecordPO.getOpsTaskRecordId()); 

            opsTaskExecutionInfoPO = new OpsTaskExecutionInfoPO();
            // set执行明细表的字段值 省略。使用mybatisplus 的注解主键字段，保存成功后会填充主键id
            opsTaskExecutionInfoMapper.save(opsTaskExecutionInfoPO);
			//将执行明细的id传递到流程上下文
            context.setOpsTaskExecutionInfoId(opsTaskExecutionInfoPO.getId());

            // 4. 执行流程（核心：此处会等待所有并行分支完成）
            processNode(startNode.getId(), workflow, context);

            // 5. 流程正常完成：统计最终状态
            updateFinalExecutionStatus(opsTaskExecutionInfoPO, context);

        } catch (Exception e) {
            LogUtil.info(log, "WorkflowEngine-execute", "WorkflowEngine-execute流程执行失败：{}", e);
        } finally {
            if (opsTaskExecutionInfoPO != null && opsTaskExecutionInfoPO.getExecuteEndTime() == null) {
                opsTaskExecutionInfoPO.setExecuteEndTime(new Date());
                updateExecutionInfo(opsTaskExecutionInfoPO);
            }
        }

    }
    /**
     * 处理节点（递归核心，保持原有逻辑结构，优化状态传递）
     */
    private void processNode(String nodeId, WorkflowDefinition workflow, WorkflowContext context) {
        if (context.isProcessEnded()) {
            LogUtil.info(log, "processNode", "流程已结束，跳过节点：{}", nodeId);
            return;
        }

        // 获取当前节点
        Node currentNode = workflow.getNodes().stream()
                .filter(node -> node.getId().equals(nodeId))
                .findFirst()
                .orElseThrow(() -> new RuntimeException("未找到节点：" + nodeId));

        LogUtil.info(log, "processNode", "开始处理节点：{}({})，节点类型：{}",
                currentNode.getId(), currentNode.getText(), currentNode.getType());

        // 按节点类型处理
        switch (currentNode.getType()) {
            case Node.TYPE_START:
                processStartNode(currentNode, workflow, context);
                break;
            case Node.TYPE_TASK:
                processTaskNode(currentNode, workflow, context);
                break;
            case Node.TYPE_END:
                processEndNode(currentNode, context);
                break;
            case Node.TYPE_PARALLEL_GATEWAY://处理并行网关
                processParallelGateway(currentNode, workflow, context);
                break;
            case Node.TYPE_CONDITIONAL_PARALLEL_GATEWAY://处理条件并行网关
                processConditionalParallelGateway(currentNode, workflow, context);
                break;
            case Node.TYPE_BRANCH_GATEWAY://处理分支网关
                processBranchGateway(currentNode, workflow, context);
                break;
            case Node.TYPE_CONVERGE_GATEWAY://处理汇聚网关
                processConvergeGateway(currentNode, workflow, context);
                break;
            default:
                throw new RuntimeException("不支持的节点类型：" + currentNode.getType());
        }
    }

    /**
     * 处理开始节点（无逻辑变更，保持原有）
     */
    private void processStartNode(Node startNode, WorkflowDefinition workflow, WorkflowContext context) {
        List<Edge> outgoingEdges = getOutgoingEdges(startNode.getId(), workflow);
        if (outgoingEdges.size() != 1) {
            throw new RuntimeException("开始节点必须有且仅有一条出口边，当前：" + outgoingEdges.size());
        }
        processNode(outgoingEdges.get(0).getTargetNodeId(), workflow, context);
    }

    /**
     * 处理任务节点（优化：变量存储线程安全，异常信息更详细）
     */
    private void processTaskNode(Node taskNode, WorkflowDefinition workflow, WorkflowContext context) {
        String taskNodeId = taskNode.getId();
        try {
            // 获取任务处理器（策略模式，保持原有）
            TaskProcessor processor = taskProcessorFactory.getProcessor(taskNode);
            if (processor == null) {
                throw new RuntimeException("未找到任务节点[" + taskNodeId + "]的处理器");
            }

            // 执行任务
            LogUtil.info(log, "processTaskNode", "开始执行任务节点：{}", taskNodeId);
            TaskResult result = processor.execute(taskNode, context);
            LogUtil.info(log, "processTaskNode", "任务节点{}执行结果：{}", taskNodeId, result.isSuccess());

            // 线程安全存储任务结果（主上下文，无副本）
            context.setVariable("taskResult_" + taskNodeId, result.isSuccess());
            if (!result.isSuccess()) {
                context.setVariable("taskError_" + taskNodeId, result.getMessage());
            }

            // 继续下一个节点
            processNextNodes(taskNodeId, workflow, context);

        } catch (Exception e) {
            String errorMsg = "任务节点[" + taskNodeId + "]执行失败：" + e.getMessage();
            context.setVariable("taskError_" + taskNodeId, errorMsg);
            throw new RuntimeException(errorMsg, e);
        }
    }
/**
     * 处理结束节点（无逻辑变更，保持原有）
     */
    private void processEndNode(Node endNode, WorkflowContext context) {
        LogUtil.info(log, "processEndNode", "流程结束于节点：{}({})", endNode.getId(), endNode.getText());
        context.setProcessEnded(true);
    }

    /**
     * 处理并行网关（优化核心：用ExecutorService+Future等待分支，无上下文副本）
     */
    private void processParallelGateway(Node gatewayNode, WorkflowDefinition workflow, WorkflowContext context) {
        List<Edge> outgoingEdges = getOutgoingEdges(gatewayNode.getId(), workflow);
        if (outgoingEdges.isEmpty()) {
            throw new RuntimeException("并行网关[" + gatewayNode.getId() + "]必须至少有一条出口边");
        }

        String gatewayId = gatewayNode.getId();
        LogUtil.info(log, "processParallelGateway", "并行网关{}开始执行{}个分支", gatewayId, outgoingEdges.size());

        // 存储并行分支的Future（用于等待所有分支完成）
        List<Future<?>> branchFutures = new ArrayList<>(outgoingEdges.size());

        try {
            // 提交所有分支任务到线程池
            for (Edge edge : outgoingEdges) {
                String nextNodeId = edge.getTargetNodeId();
                String branchKey = edge.getSourceNodeId() + "_" + nextNodeId; // 唯一分支标识（网关ID+目标节点ID）

                // 提交任务（lambda捕获final变量，确保线程安全）
                Future<?> future = opsTaskExecutor.submit(() -> {
                    try {
                        LogUtil.info(log, "processParallelGateway", "并行网关{}分支{}开始执行", gatewayId, branchKey);
                        processNode(nextNodeId, workflow, context); // 直接用主上下文，状态同步
                        context.markBranchCompleted(branchKey); // 标记分支完成（主上下文）
                        LogUtil.info(log, "processParallelGateway", "并行网关{}分支{}执行完成", gatewayId, branchKey);
                    } catch (Exception e) {
                        String errorMsg = "并行网关[" + gatewayId + "]分支[" + branchKey + "]执行失败";
                        LogUtil.error(log, "processParallelGateway", errorMsg, e);
                        context.setVariable("parallelBranchError_" + branchKey, errorMsg);
                        throw new RuntimeException(errorMsg, e);
                    }
                });
                branchFutures.add(future);
            }

            // 等待所有分支完成（加超时，避免永久阻塞）
            for (Future<?> future : branchFutures) {
                future.get(PARALLEL_BRANCH_TIMEOUT, PARALLEL_TIMEOUT_UNIT); // 超时抛出TimeoutException
            }

            LogUtil.info(log, "processParallelGateway", "并行网关{}所有分支执行完成", gatewayId);
            processNextNodes(gatewayId, workflow, context); // 继续下一个节点（通常是汇聚网关）

        } catch (TimeoutException e) {
            // 并行分支超时：标记所有未完成分支为失败
            String errorMsg = "并行网关[" + gatewayId + "]分支执行超时（超过" + PARALLEL_BRANCH_TIMEOUT + "分钟）";
            LogUtil.error(log, "processParallelGateway", errorMsg, e);
            context.setVariable("parallelGatewayTimeout_" + gatewayId, errorMsg);
            throw new RuntimeException(errorMsg, e);

        } catch (InterruptedException | ExecutionException e) {
            // 分支执行异常：包装异常并抛出
            String errorMsg = "并行网关[" + gatewayId + "]分支执行异常";
            LogUtil.error(log, "processParallelGateway", errorMsg, e);
            throw new RuntimeException(errorMsg, e.getCause());
        }
    }

    /**
     * 处理条件并行网关（复用并行网关逻辑，先筛选条件分支）
     */
    private void processConditionalParallelGateway(Node gatewayNode, WorkflowDefinition workflow, WorkflowContext context) {
        List<Edge> outgoingEdges = getOutgoingEdges(gatewayNode.getId(), workflow);
        // 筛选满足条件的分支（用SpEL解析）
        List<Edge> qualifiedEdges = outgoingEdges.stream()
                .filter(edge -> evaluateCondition(edge.getProperties().getOrDefault("condition", "true").toString(), context))
                .collect(Collectors.toList());

        if (qualifiedEdges.isEmpty()) {
            throw new RuntimeException("条件并行网关[" + gatewayNode.getId() + "]无满足条件的分支");
        }

        String gatewayId = gatewayNode.getId();
        LogUtil.info(log, "processConditionalParallelGateway", "条件并行网关{}筛选出{}个满足条件的分支", gatewayId, qualifiedEdges.size());

        // 替换网关的出口边为筛选后的分支，复用并行网关逻辑
        workflow.setEdges(workflow.getEdges().stream()
                .filter(edge -> !edge.getSourceNodeId().equals(gatewayId) || qualifiedEdges.contains(edge))
                .collect(Collectors.toList()));

        // 调用并行网关处理逻辑（复用代码，减少冗余）
        processParallelGateway(gatewayNode, workflow, context);
    }
     /**
     * 处理分支网关（优化：用SpEL解析条件，逻辑更健壮）
     */
    private void processBranchGateway(Node gatewayNode, WorkflowDefinition workflow, WorkflowContext context) {
        List<Edge> outgoingEdges = getOutgoingEdges(gatewayNode.getId(), workflow);
        if (outgoingEdges.isEmpty()) {
            throw new RuntimeException("分支网关[" + gatewayNode.getId() + "]无出口边");
        }

        // 查找第一个满足条件的分支
        Edge qualifiedEdge = null;
        for (Edge edge : outgoingEdges) {
            String condition = edge.getProperties().getOrDefault("condition", "true").toString();
            if (evaluateCondition(condition, context)) {
                qualifiedEdge = edge;
                break;
            }
        }

        // 无满足条件分支时，查找默认分支（无condition或condition=true）
        if (qualifiedEdge == null) {
            qualifiedEdge = outgoingEdges.stream()
                    .filter(edge -> !edge.getProperties().containsKey("condition")
                            || "true".equals(edge.getProperties().get("condition").toString()))
                    .findFirst()
                    .orElseThrow(() -> new RuntimeException("分支网关[" + gatewayNode.getId() + "]无满足条件分支，也无默认分支"));
        }

        LogUtil.info(log, "processBranchGateway", "分支网关{}选择分支：{}（目标节点：{}）",
                gatewayNode.getId(), qualifiedEdge.getId(), qualifiedEdge.getTargetNodeId());
        processNode(qualifiedEdge.getTargetNodeId(), workflow, context);
    }

    /**
     * 处理汇聚网关（优化：依赖主上下文分支状态，判断逻辑正确）
     */
    private void processConvergeGateway(Node gatewayNode, WorkflowDefinition workflow, WorkflowContext context) {
        List<Edge> incomingEdges = getIncomingEdges(gatewayNode.getId(), workflow);
        if (incomingEdges.isEmpty()) {
            LogUtil.warn(log, "processConvergeGateway", "汇聚网关[" + gatewayNode.getId() + "]无入口边，默认继续执行");
            processNextNodes(gatewayNode.getId(), workflow, context);
            return;
        }

        // 检查所有入口边对应的分支是否完成
        boolean allBranchesCompleted = checkAllBranchesCompleted(incomingEdges, context);
        if (!allBranchesCompleted) {
            // 分支未完成：标记pending（实际项目可结合定时任务重试，此处简化）
            String pendingMsg = "汇聚网关[" + gatewayNode.getId() + "]等待所有分支完成，暂存状态";
            LogUtil.info(log, "processConvergeGateway", pendingMsg);
            context.setVariable("convergeGatewayPending_" + gatewayNode.getId(), pendingMsg);
            return;
        }

        // 所有分支完成：继续执行
        LogUtil.info(log, "processConvergeGateway", "汇聚网关{}所有分支已完成，继续执行", gatewayNode.getId());
        processNextNodes(gatewayNode.getId(), workflow, context);
    }
    // ==================== 工具方法（优化后）====================

    /**
     * 获取节点出口边（无变更）
     */
    private List<Edge> getOutgoingEdges(String nodeId, WorkflowDefinition workflow) {
        return workflow.getEdges().stream()
                .filter(edge -> edge.getSourceNodeId().equals(nodeId))
                .collect(Collectors.toList());
    }

    /**
     * 获取节点入口边（无变更）
     */
    private List<Edge> getIncomingEdges(String nodeId, WorkflowDefinition workflow) {
        return workflow.getEdges().stream()
                .filter(edge -> edge.getTargetNodeId().equals(nodeId))
                .collect(Collectors.toList());
    }

    /**
     * 处理下一个节点（优化：增加多出口边日志提醒）
     */
    private void processNextNodes(String currentNodeId, WorkflowDefinition workflow, WorkflowContext context) {
        if (context.isProcessEnded()) {
            return;
        }

        List<Edge> outgoingEdges = getOutgoingEdges(currentNodeId, workflow);
        if (outgoingEdges.size() > 1) {
            LogUtil.info(log, "processNextNodes", "节点{}存在{}条出口边，可能导致并行执行（非网关节点建议单出口）",
                    currentNodeId, outgoingEdges.size());
        }

        for (Edge edge : outgoingEdges) {
            // 解析边的条件（SpEL）
            if (edge.getProperties().containsKey("condition") && !evaluateCondition(edge.getProperties().get("condition").toString(), context)) {
                LogUtil.info(log, "processNextNodes", "节点{}出口边{}条件不满足，跳过", currentNodeId, edge.getId());
                continue;
            }
            processNode(edge.getTargetNodeId(), workflow, context);
        }
    }

    /**
     * 评估条件表达式（简化实现，实际可集成SpEL等表达式引擎）
     */
    private boolean evaluateCondition(String condition, WorkflowContext context) {
        // 示例：处理 ${backupResult === 'success'} 这样的表达式
        if (condition.startsWith("${") && condition.endsWith("}")) {
            String expr = condition.substring(2, condition.length() - 1);

            // 简单处理 backupResult === 'success' 这种形式
            if (expr.contains("===")) {
                String[] parts = expr.split("===");
                String varName = parts[0].trim();
                String expectedValue = parts[1].trim().replace("'", "");

                Object actualValue = context.getVariable(varName);
                return expectedValue.equals(actualValue != null ? actualValue.toString() : null);
            }
        }

        // 默认返回true（无条件或无法解析的条件）
        return true;
    }

    /**
     * 检查所有分支是否完成（优化：依赖主上下文的分支状态）
     */
    private boolean checkAllBranchesCompleted(List<Edge> incomingEdges, WorkflowContext context) {
        for (Edge edge : incomingEdges) {
            // 分支标识：入口边的「源节点ID+目标节点ID」（与并行分支标记的一致）
            String branchKey = edge.getSourceNodeId() + "_" + edge.getTargetNodeId();
            if (!context.isBranchCompleted(branchKey)) {
                LogUtil.info(log, "checkAllBranchesCompleted", "汇聚网关入口边{}对应分支{}未完成", edge.getId(), branchKey);
                return false;
            }
        }
        return true;
    }
    /**
     * 更新执行记录（抽取通用方法，减少冗余）
     */
    private void updateExecutionInfo(OpsTaskExecutionInfoPO executionInfoPO) {
    	//更新执行状态，执行时间等字段，根据主键id
        executionInfoPO.setUpdateTime(LocalDateTime.now());
        //opsTaskExecutionInfoMapper.updateById();
    }

    /**
     * 统计并更新最终执行状态（抽取通用方法）
     */
    private void updateFinalExecutionStatus(OpsTaskExecutionInfoPO executionInfoPO, WorkflowContext context) {
        Long executionInfoId = context.getOpsTaskExecutionInfoId();
		// executionInfoPO.setExecutionStatus(TicketExecutionStatusEnum.EXECUTED);
}

/**
 * 任务执行情况表 PO 类（与数据库表 ops_task_execution_info 映射）
 * @author: 
 * @since: 2025/9/8 15:16
 */
@Data
@TableName("ops_task_execution_info")
public class OpsTaskExecutionInfoPO {

    /**
     * 关联任务列表主键ID（对应表字段 ops_task_record_id）
     */
    @TableField(value = "ops_task_record_id")
    private Long opsTaskRecordId;

    /**
     * 执行次数（第几次执行）（对应表字段 execute_count）
     */
    @TableField(value = "execute_count")
    private Integer executeCount;

    /**
     * 执行开始时间（对应表字段 execute_start_time）
     * 备注：表中字段允许为 NULL，实体类用 Date 类型映射 datetime 字段
     */
    @TableField(value = "execute_start_time")
    private Date executeStartTime;

    /**
     * 执行结束时间（对应表字段 execute_end_time）
     */
    @TableField(value = "execute_end_time")
    private Date executeEndTime;

    /**
     * 执行状态（对应表字段 execution_status）
     */
    @TableField(value = "execution_status")
    private TicketExecutionStatusEnum executionStatus;
}

/**
 * @author: 
 * 任务模板
 * @since: 2025/9/1 13:41
 */
@Data
@TableName("ops_task_template")
public class OpsTaskTemplatePO extends BasePO {

    /** 任务模板名称（脚本名称） */
    @TableField("task_template_name")
    private String taskTemplateName;

    /** 任务模板描述（脚本描述） */
    @TableField("task_template_desc")
    private String taskTemplateDesc;

    /** 流程配置JSON（脚本的核心配置） */
    @TableField("flow_json")
    private String flowJson;

    /** 流程唯一标识Key */
    @TableField("diagram_key")
    private String diagramKey;


    @TableField("market_diagram_key")
    private String marketDiagramKey;
}

  /**
     * 测试调用LTS 提交任务, 并接收任务执行反馈结果
     * https://www.cnblogs.com/MrYuChen-Blog/p/14803475.html
     * @author 
     * @date 2025/9/15 11:07
     * @return Map<Object>
     */
    @GetMapping("test01")
    public Map<String, Object> test01() {
        Job job = new Job();
        job.setTaskId("task-AAAAAAAAAAAAAAA");
        job.setCronExpression("0/3 * * * * ?");
        //设置任务类型 区分不同的任务 执行不同的业务逻辑
        //job.setParam("type", "aType");
        job.setNeedFeedback(true);
        //任务触发时间 如果设置了 cron 则该设置无效
        //job.setTriggerTime(DateUtils.addDay(new Date(), 1).getTime());
        
        // job.setMaxRetryTimes(5);
        // 这个是 cron expression 和 quartz 一样，可选
        // job.setCronExpression(cronExpression);
        // 这个是指定执行时间，可选
        // job.setTriggerTime(new Date());
        // 当 cronExpression 和 triggerTime 都不设置的时候，默认是立即执行任务
        
        //任务执行节点组
        job.setTaskTrackerNodeGroup("test_TaskTracker");
        //当任务队列中存在这个任务的时候，是否替换更新
        job.setReplaceOnExist(true);
        Map<String, Object> submitResult = new HashMap<String, Object>(4);
        try {
            //任务提交返回值 response
            Response response = jobClient.submitJob(job);
            submitResult.put("success", response.isSuccess());
            submitResult.put("msg", response.getMsg());
            submitResult.put("code", response.getCode());
        } catch (Exception e) {
            log.error("提交任务失败", e);
            throw new RuntimeException("提交任务失败");
        }
        return submitResult;
    }

/**
 * 
 * 任务具体的业务逻辑,此类可以被Quartz，LTS，xxl-job 调用，我使用的是LTS定时任务框架
 * @since: 2025/9/2 9:24
 */
@Component
@JobRunner4TaskTracker
public class JobRunnerA implements JobRunner {

    private static final Logger log = LoggerFactory.getLogger(JobRunnerA.class);

    @Resource
    OpsTaskRecordMapper opsTaskRecordMapper;

    @Resource
    OpsTaskExecutionInfoMapper opsTaskExecutionInfoMapper;

    @Resource
    WorkflowEngine workflowEngine;

    @Resource
    JobClient jobClient;

    @Value("${lts.tasktracker.node-group}")
    private String taskTrackerNodeGroup;

    @Override
    public Result run(JobContext jobContext) throws Throwable {
        Job job = jobContext.getJob();
        //扩展参数
        Map<String, String> extParams = job.getExtParams();
        //任务id，也是record表的主键id
        String taskId = job.getTaskId();

        LogUtil.info(log, "JobRunnerA", "JobRunnerA-{},开始执行", taskId);

        Date date = new Date();
        // 周期任务的有效时间
        String executeStartTime = extParams.get("executeStartTime");
        String executeEndTime = extParams.get("executeEndTime");

        //周期性任务应该判断当前时间是否在周期任务的有效时间范围内，如果在则执行任务
        OpsTaskRecordPO recordPO = opsTaskRecordMapper.getById(Long.parseLong(taskId));
        if (date.after(DateUtil.parseDate(executeStartTime)) && date.before(DateUtil.parseDate(executeEndTime))) {
            try {
                WorkflowContext workflowContext = new WorkflowContext();
                //将任务id传入上下文中
                workflowContext.setTaskId(taskId);
                //将任务对象放入上下文
                workflowContext.setOpsTaskRecordPO(recordPO);
                // 执行任务完成后，如果任务执行成功，则更新任务记录表的状态，记录执行情况表，ip表
                workflowEngine.execute(recordPO.getOpsTaskTemplateFlowJson(), workflowContext);

                // ops_task_execution_info 查询当前已执行次数（max(executedCycles)，首次执行为0）
                Integer maxExecutedCycles = opsTaskExecutionInfoService.lambdaQuery()
                        .eq(OpsTaskExecutionInfoPO::getOpsTaskRecordId, recordPO.getId())
                        .select(OpsTaskExecutionInfoPO::getExecuteCount)
                        .orderByDesc(OpsTaskExecutionInfoPO::getExecuteCount)
                        .last("LIMIT 1") // 取最大的一条
                        .oneOpt() // 用oneOpt避免空指针，返回Optional
                        .map(OpsTaskExecutionInfoPO::getExecuteCount)
                        .orElse(0); // 首次执行时默认0


                //  ops_task_record 计算新的执行次数（+1）并更新任务记录
                int newExecutedCycles = maxExecutedCycles + 1;
                opsTaskRecordService.lambdaUpdate()
                        .eq(OpsTaskRecordPO::getId, recordPO.getId()) // 补全更新条件：根据ID定位
                        .set(OpsTaskRecordPO::getExecutedCycles, newExecutedCycles)
                        .set(OpsTaskRecordPO::getExecuteStatus, OpsTaskStatusEnum.PERIODIC_RUNNING) // 更新为执行中
                        .update();

            } catch (Exception e) {
                LogUtil.info(log, "JobRunnerA", "JobRunnerA-{},更新任务记录为失败状态:{}", e);
            }
        } else {
            // 如果不在范围内，则不执行任务，并且任务取消掉，更新记录表的状态
            jobClient.cancelJob(taskId, taskTrackerNodeGroup);
            recordPO.setExecuteStatus(OpsTaskStatusEnum.COMPLETED);
            opsTaskRecordService.updateById(recordPO);
        }

        LogUtil.info(log, "JobRunnerA", "JobRunnerA-{},执行完毕", taskId);
        return new Result(Action.EXECUTE_SUCCESS);
    }


}

  @Resource
    WorkflowEngine workflowEngine;

    @PostMapping("workflowEngineTest")
    public Object workflowEngineTest(@RequestBody WorkflowDefinition flowMapJson){
        workflowEngine.execute(JSONUtil.toJsonStr(flowMapJson), new WorkflowContext());
        return "成功过孤寡孤寡孤寡嘎嘎嘎嘎嘎嘎";
    }