Conductor 微服务编排实战基于 HTTP 任务的串行链、条件分支与 Fork/Join 并行编排【免费下载链接】conductorConductor is an event driven agentic workflow engine providing durable and highly resilient execution engine for applications and AI Agents项目地址: https://gitcode.com/GitHub_Trending/co/conductor导读本篇文章以 Conductor 官方 Cookbook 中的微服务编排配方microservice orchestration为主线讲解如何不编写任何自定义 Worker仅靠内置的 HTTP 系统任务把散落在多个服务中的接口编排成可靠的工作流。你将掌握三种核心编排模式——HTTP 串行服务链、基于 SWITCH 的条件分支、基于 Fork/Join 的并行调用并能在本地用 curl 完成工作流注册与运行同时从源码层面理解 HTTP 任务的状态判定、SWITCH 表达式求值与 JOIN 的汇合机制。一、为什么微服务编排可以零代码微服务编排最常见的诉求是把校验订单 → 扣款 → 预留库存 → 发送确认这类跨服务调用串起来。在 Conductor 中这不需要你为每个接口写一个 Worker 进程因为引擎内置了 HTTP 系统任务HTTPtask它本身就能发起 REST 调用并在调用结束后把响应写回任务输出。HTTP 任务的实现位于 HttpTask.java。从源码可以看到它继承自WorkflowSystemTask通过Component(TASK_TYPE_HTTP)注册任务类型名为HTTP见 TaskType.TASK_TYPE_HTTP 相关定义任务输入约定使用http_request作为键名源码中REQUEST_PARAMETER_NAME http_request值是一个Input对象如果输入里没有http_request键则会直接把整个inputData当作请求参数解析start()方法校验uri与method是否缺失缺失则直接置为FAILED并写入reasonForIncompletion调用成功后HttpResponse被写入response输出键结构包含body、headers、statusCode、reasonPhrase四个字段HttpResponse.asMap()isAsync()返回true属于异步系统任务。Input类定义了 HTTP 任务的完整入参其默认值同样可以在源码中确认参数类型默认值说明uriString无必填目标地址methodString无必填如PUT/POST/GET/DELETE/OPTIONS/HEADheadersMap空请求头值为非 null 才会被添加bodyObjectnull请求体acceptListStringapplication/json接受列表兼容单个字符串写法contentTypeStringapplication/json请求 MIME 类型connectionTimeOutInteger3000毫秒连接超时readTimeOutInteger3000毫秒读取超时vipAddressStringnull旧版服务发现预留字段关于状态判定源码逻辑是httpCall()返回的statusCode落在199 statusCode 300区间时任务置为COMPLETED否则置为FAILED并把响应体写入reasonForIncompletion任何RestClientException异常都会让任务失败。对应测试可参考 HttpTaskTest.java覆盖了 GET 超时、非 2xx 失败、缺失uri/method等场景。理解了 HTTP 任务的入参与输出契约下面三种模式就可以直接用 JSON 定义工作流了。二、模式一HTTP 服务链串行编排场景依次调用一组 HTTP 接口后一步需要用到前一步的输出。这是最常见的管道式编排。工作流定义order_processing完整示例如下{ name: order_processing, description: Validate order, charge payment, reserve inventory, send confirmation, version: 1, schemaVersion: 2, inputParameters: [orderId, customerId, amount, items], tasks: [ { name: validate_order, taskReferenceName: validate, type: HTTP, inputParameters: { http_request: { uri: https://api.example.com/orders/${workflow.input.orderId}/validate, method: POST, body: { customerId: ${workflow.input.customerId}, items: ${workflow.input.items} }, connectionTimeOut: 5000, readTimeOut: 5000 } } }, { name: charge_payment, taskReferenceName: payment, type: HTTP, inputParameters: { http_request: { uri: https://api.example.com/payments/charge, method: POST, body: { orderId: ${workflow.input.orderId}, amount: ${workflow.input.amount}, customerId: ${workflow.input.customerId} }, connectionTimeOut: 10000, readTimeOut: 10000 } } }, { name: reserve_inventory, taskReferenceName: inventory, type: HTTP, inputParameters: { http_request: { uri: https://api.example.com/inventory/reserve, method: POST, body: { orderId: ${workflow.input.orderId}, items: ${workflow.input.items}, paymentId: ${payment.output.response.body.paymentId} }, connectionTimeOut: 5000, readTimeOut: 5000 } } }, { name: send_confirmation, taskReferenceName: notify, type: HTTP, inputParameters: { http_request: { uri: https://api.example.com/notifications/send, method: POST, body: { customerId: ${workflow.input.customerId}, orderId: ${workflow.input.orderId}, paymentId: ${payment.output.response.body.paymentId}, reservationId: ${inventory.output.response.body.reservationId} } } } } ], outputParameters: { paymentId: ${payment.output.response.body.paymentId}, reservationId: ${inventory.output.response.body.reservationId} }, failureWorkflow: order_compensation, timeoutPolicy: TIME_OUT_WF, timeoutSeconds: 120 }数据在任务间传递的关键每个任务通过taskReferenceName引用上游任务的输出配合 HTTP 任务的response输出结构表达式形如${taskReferenceName.output.response.body.field}${workflow.input.xxx}读取工作流启动时传入的全局输入${payment.output.response.body.paymentId}读取charge_payment任务reference 名为paymentHTTP 响应体中的paymentId字段由于HttpResponse固定包含body/headers/statusCode/reasonPhrase即使响应体不是 JSON 也能通过response.body拿到原始内容。失败与超时兜底重试如果某个 HTTP 任务失败Conductor 会按任务定义配置的策略自动重试默认可配无需在 JSON 中额外声明补偿工作流级配置failureWorkflow: order_compensation指定补偿工作流当主流程失败时自动触发。其底层实现对应 WorkflowExecutor.terminateWorkflow(...) 中的failureWorkflow参数处理整体超时timeoutPolicy: TIME_OUT_WF配合timeoutSeconds: 120意味着整个订单流程最多执行 120 秒超时则终止工作流避免扣款成功但确认邮件长期挂起。注册并运行将上述 JSON 保存为order_processing.json后执行curl -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json \ -d order_processing.json curl -X POST http://localhost:8080/api/workflow/order_processing \ -H Content-Type: application/json \ -d {orderId: ORD-123, customerId: CUST-456, amount: 99.99, items: [SKU-A, SKU-B]}第一条命令把工作流定义注册到 Conductor 元数据服务/api/metadata/workflow第二条命令用订单参数启动一个工作流实例/api/workflow/order_processing。三、模式二HTTP 调用后的条件分支SWITCH场景根据上一个 HTTP 任务的响应结果决定后续走哪条分支例如按用户等级决定分配客户经理还是发送欢迎邮件。这里使用SWITCH系统任务它是已废弃DECISION任务的替代实现见 Switch.java。工作流定义user_onboarding完整示例如下{ name: user_onboarding, version: 1, schemaVersion: 2, inputParameters: [userId], tasks: [ { name: get_user_profile, taskReferenceName: profile, type: HTTP, inputParameters: { http_request: { uri: https://api.example.com/users/${workflow.input.userId}, method: GET } } }, { name: route_by_tier, taskReferenceName: tier_switch, type: SWITCH, evaluatorType: javascript, expression: $.tier enterprise ? enterprise : standard, inputParameters: { tier: ${profile.output.response.body.tier} }, decisionCases: { enterprise: [ { name: assign_account_manager, taskReferenceName: assign_am, type: HTTP, inputParameters: { http_request: { uri: https://api.example.com/account-managers/assign, method: POST, body: {userId: ${workflow.input.userId}} } } } ], standard: [ { name: send_welcome_email, taskReferenceName: welcome, type: HTTP, inputParameters: { http_request: { uri: https://api.example.com/emails/welcome, method: POST, body: {userId: ${workflow.input.userId}} } } } ] } } ] }SWITCH 是如何工作的从 SwitchTaskMapper.java 的源码可以看出完整调度链路根据evaluatorType从已注册的Evaluator集合中取出求值器示例用的是内置的javascript求值器对expression求值得到evalResult并把它写入 SWITCH 任务的输入case与输出evaluationResult、selectedCase用evalResult到decisionCases中查表命中则调度对应 case 下的任务列表没有命中时回退到defaultCase源码注释特别说明只有 case 键完全没匹配时才走默认分支显式空列表的分支不会被默认分支覆盖若evaluatorType没有注册对应的求值器工作流会被直接终止TerminateWorkflowException。示例中expression为$.tier enterprise ? enterprise : standard其中$.tier指向输入参数tier由profile任务的响应体传入求值结果要么是enterprise要么是standard与decisionCases中的两个 case 一一对应。这种先 HTTP 取数、再 SWITCH 路由的组合是动态路由类工作流的通用骨架。四、模式三并行 HTTP 调用Fork/Join场景多个 HTTP 调用彼此独立希望并发执行以缩短总耗时并在全部完成后继续后续流程。使用FORK_JOINJOIN组合。工作流定义enrich_customer_data完整示例如下{ name: enrich_customer_data, version: 1, schemaVersion: 2, inputParameters: [customerId], tasks: [ { name: parallel_enrichment, taskReferenceName: fork, type: FORK_JOIN, forkTasks: [ [ { name: get_credit_score, taskReferenceName: credit, type: HTTP, inputParameters: { http_request: { uri: https://api.example.com/credit/${workflow.input.customerId}, method: GET } } } ], [ { name: get_purchase_history, taskReferenceName: purchases, type: HTTP, inputParameters: { http_request: { uri: https://api.example.com/purchases/${workflow.input.customerId}, method: GET } } } ], [ { name: get_support_tickets, taskReferenceName: tickets, type: HTTP, inputParameters: { http_request: { uri: https://api.example.com/support/${workflow.input.customerId}, method: GET } } } ] ] }, { name: join_results, taskReferenceName: join, type: JOIN, joinOn: [credit, purchases, tickets] } ], outputParameters: { creditScore: ${credit.output.response.body}, purchases: ${purchases.output.response.body}, tickets: ${tickets.output.response.body} } }Fork/Join 的引擎侧实现Fork 侧ForkJoinTaskMapper.java 会把FORK_JOIN映射为一个立即COMPLETED的FORK系统任务再按forkTasks中每个分支列表逐个调度首任务同时它会校验 FORK 的下一个任务必须是JOIN否则抛出TerminateWorkflowExceptionFork task definition is not followed by a join task。forkTasks是一个二维数组每个内层数组是一条独立分支分支内可以有多个串行任务。Join 侧JoinTaskMapper.java 把joinOn写入 JOIN 任务输入Join.java 在执行时逐个检查joinOn引用的任务是否已进入终态TERMINAL只要还有分支未终态JOIN 返回false等待下一次调度轮询全部终态后JOIN 将各分支任务的输出以referenceName - output的形式汇总到自身输出然后置为COMPLETED若有非可选optionalfalse分支失败JOIN 会置为FAILED并汇总失败原因可选分支的失败则表现为COMPLETED_WITH_ERRORS默认采用带退避的异步求值getEvaluationOffset按轮询次数指数退避也支持joinMode: SYNC让 JOIN 立即参与每次调度判定。上面示例中三个 HTTP 调用会同时执行JOIN等待credit、purchases、tickets三个分支全部完成后工作流才继续随后即可在outputParameters中通过${credit.output.response.body}等表达式把三路数据汇总为工作流输出。五、三种模式的选择建议模式适用场景关键任务类型数据传递方式HTTP 服务链步骤有严格先后依赖、后一步消费前一步结果HTTP${taskRef.output.response.body.field}SWITCH 条件分支依据前序任务结果动态路由HTTPSWITCHevaluatorTypeexpressiondecisionCasesFork/Join 并行多个独立调用需并发、等待全部完成FORK_JOINJOINforkTasks二维数组 joinOn汇合三个模式可以自由嵌套组合例如 Fork 分支内再放 SWITCH从而在不写一行 Java 代码的情况下用纯 JSON 声明式地完成复杂的跨服务编排。工作流元数据与运行期的完整概念可进一步参阅 workflows、tasks 与 workers 文档若要深入了解 HTTP 任务的超时、状态码判定与响应解析细节可直接阅读 HttpTask.java 及对应单元测试 HttpTaskTest.java。【免费下载链接】conductorConductor is an event driven agentic workflow engine providing durable and highly resilient execution engine for applications and AI Agents项目地址: https://gitcode.com/GitHub_Trending/co/conductor创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考