概览支持灵活的条件分支与并行处理可复用子工作流实现流程嵌套通过 Webhook、消息队列等方式与外部系统无缝集成一、条件分支与并行网关1. 条件分支根据工单严重程度自动路由到不同的处理流程Configuration public class TicketWorkflowConfig { Bean public WorkflowDefinition ticketWorkflow(DiagnosisWorkflowNode diagnosisNode, SolutionWorkflowNode solutionNode, ReportWorkflowNode reportNode) { WorkflowDefinition flow WorkflowDefinition.builder() .workflowName(运维工单处理流程) .build(); return flow.addNode(new TicketValidationNode()) .addNode(diagnosisNode) .addNode(new SeverityRoutingNode()) .addNode(solutionNode) // 人工审批流程暂停 .addNode(new HumanApprovalNode()) .addNode(new VerificationNode()) .addNode(reportNode) // 条件分支severity-routing 根据 context.route 分流 // NORMAL含 HIGH/NORMAL走 AI 方案生成再人工审批 .addConditionalTransition(severity-routing, NORMAL, ai-solution) // EMERGENCYCRITICAL跳过 AI 方案直接升级到人工审批 .addConditionalTransition(severity-routing, EMERGENCY, human-approval); } }工作流定义说明/** * 工作流定义——描述流程的结构 * p * 节点之间通过 transitions 建立连接 * ul * li默认转换key fromNodeIdvalue [toNodeId, ...]/li * li条件分支key fromNodeId:routevalue [toNodeId, ...]/li * /ul * addNode 会自动与上一个节点建立默认线性转换保持线性流程兼容。 */ Data Builder public class WorkflowDefinition { private final String workflowName; private final ListWorkflowNode nodes new ArrayList(); // 节点ID → 下一个节点ID列表含默认转换与条件分支 private final MapString, ListString transitions new HashMap(); /** * 添加节点并自动与上一个节点建立默认线性转换 */ public WorkflowDefinition addNode(WorkflowNode node) { // 与上一个节点建立默认转换保持线性流程兼容 if (!CollectionUtils.isEmpty(this.nodes)) { String prevNodeId this.nodes.getLast().getNodeId(); this.transitions.computeIfAbsent(prevNodeId, k - new ArrayList()).add(node.getNodeId()); } this.nodes.add(node); this.transitions.putIfAbsent(node.getNodeId(), new ArrayList()); return this; } /** * 添加默认转换fromNodeId → toNodeId */ public WorkflowDefinition addTransition(String fromNodeId, String toNodeId) { this.transitions.computeIfAbsent(fromNodeId, k - new ArrayList()).add(toNodeId); return this; } /** * 添加条件分支转换fromNodeId route → toNodeId * 解析时 key 形如 fromNodeId:route与 context.getString(route) 匹配 */ public WorkflowDefinition addConditionalTransition(String fromNodeId, String route, String toNodeId) { String key fromNodeId : route; this.transitions.computeIfAbsent(key, k - new ArrayList()).add(toNodeId); return this; } /** * 起始节点ID */ public String getStartNodeId() { return this.nodes.isEmpty() ? null : this.nodes.get(0).getNodeId(); } /** * 按节点ID查找节点 */ public WorkflowNode getNode(String nodeId) { for (WorkflowNode node : this.nodes) { if (node.getNodeId().equals(nodeId)) { return node; } } return null; } }工作流执行引擎/** * 工作流引擎 */ Component public class WorkflowEngine { private static final Logger log LoggerFactory.getLogger(WorkflowEngine.class); Resource private WorkflowTracingAdvisor tracingAdvisor; Resource private WorkflowMetrics metrics; /** * 执行工作流从头开始 */ public WorkflowContext execute(WorkflowDefinition definition, WorkflowContext context) { log.info(【工作流启动】{} - 工单ID: {}, definition.getWorkflowName(), context.getWorkflowId()); return executeFrom(definition, context, definition.getStartNodeId()); } /** * 从指定节点开始执行支持条件分支 * 通过 transitions 图遍历由 resolveNextNode 决定下一个节点 */ private WorkflowContext executeFrom(WorkflowDefinition definition, WorkflowContext context, String startNodeId) { String currentNodeId startNodeId; while (currentNodeId ! null) { WorkflowNode node definition.getNode(currentNodeId); if (node null) { log.error(【工作流异常】未找到节点: {}, currentNodeId); context.setStatus(WorkflowStatus.FAILED); break; } context.setCurrentNodeId(currentNodeId); // 检查是否需要等待人工介入 if (context.getStatus() WorkflowStatus.WAITING_HUMAN) { log.info(【工作流暂停】等待人工审批 - 节点: {}, currentNodeId); // 暂停等人工操作后恢复 return context; } // 流程已结束 if (context.getStatus() ! WorkflowStatus.RUNNING) { break; } try { log.info(【工作流执行】节点: {} - {}, node.getNodeId(), node.getNodeName()); // 通过 TracingAdvisor 包装执行记录 trace 耗时并写入执行日志 WorkflowResult result tracingAdvisor.traceExecution(node, context, () - node.execute(context)); log.info(【工作流执行结果】节点: {} - {}, node.getNodeId(), result.getDetail()); if (!result.isSuccess()) { context.setStatus(WorkflowStatus.FAILED); log.error(【工作流失败】节点: {} - {}, node.getNodeId(), result.getDetail()); break; } } catch (Exception e) { // traceExecution 异常时仅记录日志并重抛不写执行日志这里补记 context.logExecution(node.getNodeId(), FAILED, e.getMessage()); context.setStatus(WorkflowStatus.FAILED); log.error(【工作流异常】节点: {}, node.getNodeId(), e); break; } // 查找下一个节点支持条件分支 currentNodeId resolveNextNode(definition, currentNodeId, context); } if (context.getStatus() WorkflowStatus.RUNNING) { context.setStatus(WorkflowStatus.COMPLETED); } // 上报工作流完成指标COMPLETED/FAILED/CANCELLEDWAITING_HUMAN 已提前 return不计为完成 metrics.recordWorkflowCompletion(definition.getWorkflowName(), context.getStatus().name()); log.info(【工作流结束】状态: {}, context.getStatus()); return context; } /** * 解析下一个节点先检查条件分支再走默认转换 * 条件分支 key 形如 currentNodeId:routeroute 取自 context.getString(route) */ private String resolveNextNode(WorkflowDefinition definition, String currentNodeId, WorkflowContext context) { // 先检查条件分支 String route context.getString(route); if (route ! null) { String conditionalKey currentNodeId : route; ListString targets definition.getTransitions().get(conditionalKey); if (!CollectionUtils.isEmpty(targets)) { return targets.getFirst(); } } // 默认走普通转换 ListString targets definition.getTransitions().get(currentNodeId); return CollectionUtils.isEmpty(targets) ? null : targets.getFirst(); } /** * 从暂停点恢复执行人工审批后调用 * 通过 context.getCurrentNodeId() 定位恢复点仅执行该节点及后续节点 */ public WorkflowContext resume(WorkflowDefinition definition, WorkflowContext context) { log.info(【工作流恢复】工单ID: {}, context.getWorkflowId()); context.setStatus(WorkflowStatus.RUNNING); String resumeFromNodeId context.getCurrentNodeId(); if (resumeFromNodeId null) { log.warn(【工作流恢复】未记录恢复点从头开始执行); return execute(definition, context); } log.info(【工作流恢复】从节点: {} 继续, resumeFromNodeId); return executeFrom(definition, context, resumeFromNodeId); } }2. 并行网关实现多个步骤并行执行如同时检查多个服务/** * 并行网关节点——同时执行多个子节点全部完成后继续 */ public class ParallelGatewayNode extends WorkflowNode { private final ListWorkflowNode parallelNodes; public ParallelGatewayNode(String nodeId, String nodeName, ListWorkflowNode parallelNodes) { super(nodeId, nodeName); this.parallelNodes parallelNodes; } Override public WorkflowResult execute(WorkflowContext context) { ListCompletableFutureWorkflowResult futures parallelNodes.stream() .map(node - CompletableFuture.supplyAsync(() - node.execute(context) ).orTimeout(30, TimeUnit.SECONDS)) .toList(); // 等待所有并行节点完成 ListWorkflowResult results futures.stream() .map(f - { try { return f.get(); } catch (Exception e) { return WorkflowResult.failure(并行节点执行失败 e.getMessage()); } }) .toList(); boolean allSuccess results.stream().allMatch(WorkflowResult::isSuccess); String detail results.stream() .map(WorkflowResult::getDetail) .collect(Collectors.joining(\n)); return allSuccess ? WorkflowResult.success(detail) : WorkflowResult.failure(detail); } }二、子工作流——嵌套流程1. 子工作流节点实现通过子工作流节点主流程可调用嵌套子流程/** * 子工作流节点——调用另一个工作流作为子流程 */ public class SubWorkflowNode extends WorkflowNode { private final WorkflowEngine engine; private final WorkflowDefinition subWorkflow; public SubWorkflowNode(String nodeId, String nodeName, WorkflowEngine engine, WorkflowDefinition subWorkflow) { super(nodeId, nodeName); this.engine engine; this.subWorkflow subWorkflow; } Override public WorkflowResult execute(WorkflowContext parentContext) { // 创建子工作流上下文继承父上下文的数据 String subWorkflowId parentContext.getWorkflowId() -sub; WorkflowContext subContext new WorkflowContext(subWorkflowId); subContext.getData().putAll(parentContext.getData()); // 执行子工作流 engine.execute(subWorkflow, subContext); // 将子工作流的结果写回父上下文 parentContext.put(subWorkflowResult, subContext.getData()); parentContext.logExecution(getNodeId(), subContext.getStatus().name(), 子工作流执行完成); return subContext.getStatus() WorkflowStatus.COMPLETED ? WorkflowResult.success(子工作流执行成功) : WorkflowResult.failure(子工作流执行失败 subContext.getStatus()); } }2. 应用场景举例主工作流工单处理流程子工作流通知流程如生成通知 → 发送钉钉 → 发送短信 → 确认送达通过子工作流节点主流程可灵活嵌入通知等复用流程。三、事件驱动工作流与外部系统交互1. Webhook 回调节点支持人工审批后由外部系统通过 Webhook 通知工作流继续PostMapping(/webhook/approve/{workflowId}) public ResponseEntityString webhookApprove(PathVariable String workflowId, RequestBody ApprovalCallback callback) { WorkflowContext context runningWorkflows.get(workflowId); if (context null) { return ResponseEntity.notFound().build(); } if (callback.approved()) { context.put(approved, true); context.put(approver, callback.approver()); context.put(comment, callback.comment()); engine.resume(ticketWorkflow, context); } else { context.setStatus(WorkflowStatus.CANCELLED); } runningWorkflows.put(workflowId, context); return ResponseEntity.ok(OK); }public record ApprovalCallback(boolean approved, String approver, String comment) {}2. 消息队列触发监控系统通过 RocketMQ 的告警消息自动触发工作流依赖引入dependency groupIdorg.apache.rocketmq/groupId artifactIdrocketmq-spring-boot-starter/artifactId version2.3.5/version /dependency监听报警消息自动创建工单并启动工作流Component RocketMQMessageListener(topic ops-alert-topic, consumerGroup workflow-consumer-group) public class AlertMessageListener implements RocketMQListenerAlertMessage { Resource private WorkflowEngine engine; Resource private WorkflowDefinition ticketWorkflow; Override public void onMessage(AlertMessage message) { // 自动创建工单并启动工作流 String workflowId AUTO- System.currentTimeMillis(); WorkflowContext context new WorkflowContext(workflowId); context.put(ticketContent, message.content()); context.put(serviceName, message.serviceName()); context.put(submitter, SYSTEM); context.put(autoCreated, true); engine.execute(ticketWorkflow, context); } }public record AlertMessage(String content, String serviceName) {}