Conductor 任务定义Task Definition创建与更新完全指南从 UI、CLI、API 到重试与限流配置【免费下载链接】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 工作流引擎中Task Definition任务定义的创建、更新与核心参数配置。任务定义是 Conductor 中所有SIMPLEWorker任务得以在 workflow 中执行的前置注册条件它统一描述了任务的超时策略、重试逻辑、限流与并发上限、输入输出键以及默认输入模板。读完本文你将掌握通过 Conductor UI、CLI、REST API 与各语言 SDK 注册/更新任务定义的方法并能结合源码深入理解retryLogic、timeoutPolicy、rateLimitPerFrequency、concurrentExecLimit、inputTemplate等关键字段的真实行为直接用于生产环境的任务治理。一、任务定义Conductor 中任务行为的总契约在 Conductor 中一个 任务定义 规定了任务的一般性实现细节它作用于该任务在工作流中的所有实例主要包括超时策略Timeout policy重试逻辑Retry logic速率限制与并发执行限制Rate limit and execution limit输入/输出键Input/output keys输入模板Input template注意Task Definition 与 Workflow Definition 中的 Task Configurations 是两个概念——后者属于工作流定义的一部分定义在 workflow 的tasks属性中是某个工作流里如何使用这个任务的局部配置前者则是跨工作流复用的全局任务类型注册。从源码结构看任务定义的 Java 模型对应 common/src/main/java/com/netflix/conductor/common/metadata/tasks/TaskDef.java其中明确给出了TimeoutPolicyRETRY、TIME_OUT_WF、ALERT_ONLY与RetryLogicFIXED、EXPONENTIAL_BACKOFF、LINEAR_BACKOFF两个枚举以及name、retryCount、timeoutSeconds、responseTimeoutSeconds、concurrentExecLimit、rateLimitPerFrequency、rateLimitFrequencyInSeconds、ownerEmail、pollTimeoutSeconds、backoffScaleFactor、maxRetryDelaySeconds、backoffJitterMs、totalTimeoutSeconds等完整字段——本文后续对参数行为的描述均与该实现一致。什么场景必须注册任务定义Worker 任务SIMPLE所有 worker 任务在使用前必须在 Conductor server 上注册为任务定义否则无法在工作流中执行。系统任务System tasks系统任务本身不需要任务定义但你可以用同名注册一个任务定义来定制其重试、超时与速率限制行为。二、任务定义字段全解析Schema任务的完整机器可读字段契约定义在 schemas/TaskDef.json其必填字段只有name而人类可读的字段说明见 Task Definitions 参考文档。下面按参考文档整理完整字段表字段类型说明备注namestring任务名需与任务功能语义相符必须唯一descriptionstring任务描述可选retryCountnumber任务失败后重试的次数默认 3最大上限 10retryLogicstring (enum)重试间隔的计算机制见下文 Retry LogicretryDelaySecondsnumber首次重试前的基础延迟具体含义随retryLogic变化默认 60 秒maxRetryDelaySecondsnumber重试间隔的最大值秒用于封顶EXPONENTIAL_BACKOFF与LINEAR_BACKOFF计算出的延迟0表示不封顶默认 0不封顶见 Retry LogicbackoffJitterMsnumber每次重试延迟追加最多该毫秒数的随机抖动将同时发生的重试分散到不同时刻防止惊群效应0表示无抖动默认 0无抖动见 Retry LogictotalTimeoutSecondsnumber所有重试尝试合计的最大墙钟时间秒一旦超出任务立即失败且不再重试即使retryCount未耗尽0表示不设限默认 0无限制见 Timeout 场景timeoutPolicystring (enum)任务超时后执行何种策略默认TIME_OUT_WF见 Timeout PolicytimeoutSecondsnumber任务首次进入IN_PROGRESS后若未在指定秒数内到达终态则被标记为TIMED_OUT为 0 表示无超时responseTimeoutSecondsnumber若大于 0任务在该时间内未更新状态则被重新调度心跳机制常用于 worker 已 poll 任务但因错误/网络故障未完成默认 600pollTimeoutSecondsnumber任务若在指定秒数内未被 worker poll则被标记为TIMED_OUT为 0 表示无超时inputKeysarray of string任务期望的输入键数组用于文档化任务输入可选见 inputKeys 与 outputKeysoutputKeysarray of string任务期望的输出键数组用于文档化任务输出可选见 inputKeys 与 outputKeysinputTemplateobject定义默认输入值可选见 inputTemplateconcurrentExecLimitnumber任意时刻可并发执行的任务数可选rateLimitFrequencyInSecondsnumber设置速率限制的频次窗口可选见 Task Rate limitsrateLimitPerFrequencynumber在频次窗口内可以发给 worker 的最大任务数可选见 Task Rate limitsownerEmailstring拥有该任务的团队邮箱必填补充说明来自 TaskDef.java 的实现细节name上标注了NotEmpty(message TaskDef name cannot be null or empty)ownerEmail上有OwnerEmailMandatoryConstraint校验responseTimeoutSeconds最小值为 1 秒retryCount、pollTimeoutSeconds、maxRetryDelaySeconds、backoffJitterMs、totalTimeoutSeconds均要求 0backoffScaleFactor要求 1默认 1。类上标注了TaskTimeoutConstraint用于跨字段校验超时配置的合法性。该模型还包含isolationGroupId任务执行隔离组、executionNameSpace执行命名空间、runtimeMetadata任务在 poll 时需要注入的密钥/环境变量名列表以及可选的inputSchema/outputSchema/enforceSchema配合 Schema Registry 使用等进阶字段。三、重试逻辑深入三种策略与延迟计算公式retryLogic字段控制重试之间延迟的计算方式。最终应用的实际延迟为delay clamp(computedDelay, 0, maxRetryDelaySeconds) random(0, backoffJitterMs) ms其中clamp仅在maxRetryDelaySeconds 0时生效。值延迟公式说明FIXEDretryDelaySeconds每次重试使用恒定延迟EXPONENTIAL_BACKOFFretryDelaySeconds × 2^attemptNumber每次尝试延迟翻倍建议配合maxRetryDelaySeconds封顶避免延迟失控LINEAR_BACKOFFretryDelaySeconds × backoffScaleFactor × attemptNumber线性增长backoffScaleFactor默认 1maxRetryDelaySeconds 封顶示例以EXPONENTIAL_BACKOFF、retryDelaySeconds1、maxRetryDelaySeconds3为例尝试次数原始延迟封顶后延迟01s1s12s2s24s3s38s3s对应到源码注释TaskDef.java20 次重试、初始 1 秒、封顶 600 秒时退避序列为 1, 2, 4, 8, …, 600, 600, 600, …而不是无限增长。backoffJitterMs 抖动示例backoffJitterMs会在最终延迟上追加[0, backoffJitterMs]毫秒内的均匀随机值将多个失败 worker 的重试分散到不同时间点防止惊群效应thundering herd。例如retryDelaySeconds2、backoffJitterMs1000则每次重试会在失败后 2 000 ms3 000 ms 之间触发。四、超时策略与超时字段Timeout Policy超时策略RETRY超时后再次重试该任务。TIME_OUT_WF工作流被标记为TIMED_OUT并终止。这是默认值。ALERT_ONLY仅注册一个计数器task_timeout任务超时只告警、不重试不终止工作流。超时相关字段的职责边界timeoutSeconds任务进入IN_PROGRESS后允许的最大执行时间超出即TIMED_OUT。responseTimeoutSeconds作为心跳机制——worker poll 到任务但迟迟不更新状态时超过该时间任务被重新排队requeue可避免 worker 因网络/进程故障占着茅坑。pollTimeoutSeconds任务在队列中等待 worker poll 的最长时间超出即TIMED_OUT。totalTimeoutSeconds所有重试尝试合计的最大墙钟时间预算。即使retryCount尚未耗尽只要总预算被消耗完任务立即失败且不再重试详见 tasklifecycle。五、并发执行上限与速率限制concurrentExecLimit并发执行限制concurrentExecLimit限制任意时刻**同时处于执行中IN_PROGRESS**的任务数量上限。文档中的经典例子队列中有 1000 个任务执行等待同时有 1000 个 worker 在 poll 该队列但若将concurrentExecLimit设为 10则只有 10 个任务会被交给 worker其余会出现饥饿一旦某个 worker 完成执行才会从队列中取出新任务同时将当前执行数保持为 10。从实现看concurrencyLimit()方法在concurrentExecLimit null时返回 0不限制。Task Rate Limits速率限制rateLimitFrequencyInSeconds与rateLimitPerFrequency必须成对使用。rateLimitFrequencyInSeconds设置频次窗口即每秒事件数中的 duration例如 1s、5s、60s、300s。rateLimitPerFrequency定义在每个频次窗口内可以交给 worker 的任务数设为 0 表示不限制。示例设rateLimitFrequencyInSeconds 5、rateLimitPerFrequency 12即频次窗口为 5 秒每个窗口 Conductor 只给 worker 12 个任务。因此每分钟最多给出 12 × (60/5) 144个任务无论有多少 worker 在 poll。需要注意的是与concurrentExecLimit不同速率限制不会考虑已经在执行中或处于终态的任务——即使之前的任务在 1 秒内全部执行完或者要跑好几天新任务仍然按配置的频次发放如上例每分钟 144 个。六、inputKeys / outputKeys 与 inputTemplate使用 inputKeys 与 outputKeysinputKeys和outputKeys可视为任务的参数与返回值把任务定义想象成一个接口(value1, value2 .. valueN) someTaskDefinition(key1, key2 .. keyN);。但当前这些参数并非严格强制校验——两者目前主要充当任务复用的文档说明工作流中的任务不必覆盖任务定义里的全部键。从发展角度看未来可以扩展为类似编程语言接口的严格模板约束。使用 inputTemplateinputTemplate允许定义默认输入值这些值可以被工作流中提供的值覆盖。例如在任务定义中inputTemplate: { url: https://some_url:7004 }然后在工作流定义中使用该任务时既可以沿用默认的url也可以在任务的inputParameters中覆盖inputParameters: { url: ${workflow.input.some_new_url} }七、创建与更新任务定义的三种途径7.1 使用 Conductor UI创建任务定义在左侧导航中打开Definitions选择Task。点击Define task。在Task表单中配置任务或打开Code标签页直接编辑 JSON。完整参数参考 Task Definitions。点击Save保存。更新任务定义在左侧导航中打开Definitions选择Task然后选中要更新的任务定义。在Task表单或Code标签页中修改任务。完整参数参考 Task Definitions。点击Save保存。7.2 使用 CLI将任务定义保存到 JSON 文件后执行conductor task create taskdef.json文件内容可以是单个任务定义对象或对象数组。要更新已有定义编辑文件后执行conductor task update taskdef.json完整参数参考 Task Definitions。7.3 使用 REST API创建与更新端点位于元数据 REST 控制器 MetadataResource.java其源码签名清晰地体现了两个端点的差异POST /api/metadata/taskdefs→registerTaskDef(RequestBody ListTaskDef taskDefs)接收任务定义数组支持批量创建。PUT /api/metadata/taskdefs→registerTaskDef(RequestBody TaskDef taskDef)接收单个任务定义一次只能更新一个。创建任务定义cURL 示例curl http://localhost:8080/api/metadata/taskdefs \ -H accept: */* \ -H content-type: application/json \ --data-raw [{name:sample_task_name_1,description:This is a sample task for demo,responseTimeoutSeconds:10,timeoutSeconds:30,inputKeys:[],outputKeys:[],timeoutPolicy:TIME_OUT_WF,retryCount:3,retryLogic:FIXED,retryDelaySeconds:5,inputTemplate:{},rateLimitPerFrequency:0,rateLimitFrequencyInSeconds:1}]更新任务定义cURL 示例curl http://localhost:8080/api/metadata/taskdefs \ -X PUT \ -H accept: */* \ -H content-type: application/json \ --data-raw {name:sample_task_name_1,description:This is a sample task for demo,responseTimeoutSeconds:10,timeoutSeconds:30,inputKeys:[],outputKeys:[],timeoutPolicy:TIME_OUT_WF,retryCount:3,retryLogic:FIXED,retryDelaySeconds:5,inputTemplate:{},rateLimitPerFrequency:0,rateLimitFrequencyInSeconds:1}另外同控制器还提供GET /api/metadata/taskdefs获取全部任务定义、GET /api/metadata/taskdefs/{tasktype}获取单个任务定义与DELETE /api/metadata/taskdefs/{tasktype}删除任务定义等端点便于查看与维护已注册的定义。7.4 使用各语言 SDK每个客户端 SDK 都内置了 metadata-client 方法内部调用与上面相同的 create / update 端点。当任务注册逻辑应该归属于应用或部署代码而非手工步骤时推荐使用 SDK——例如在应用启动时自动注册全部任务定义保证环境一致性。八、任务复用与多租户队列隔离任务定义一旦注册就可以被多次复用同一工作流内以不同的任务引用名task reference name使用同一个任务定义。跨工作流任意工作流都可以引用任意已注册的任务定义。在多租户系统中复用任务时需要注意默认情况下分配给某个任务的所有工作都会进入同一个队列。如果出现吵闹邻居noisy neighbor导致 poll 延迟你可以横向扩容 worker 数量或使用 task-to-domain任务域路由将任务负载路由到独立队列中。九、生产实战完整任务定义与典型重试配置示例9.1 完整示例含全部主要字段以下是一个覆盖输入输出键、并发限制、速率限制、超时与重试的完整任务定义{ name: encode_task, retryCount: 3, retryLogic: EXPONENTIAL_BACKOFF, retryDelaySeconds: 10, maxRetryDelaySeconds: 120, backoffJitterMs: 5000, totalTimeoutSeconds: 600, timeoutSeconds: 1200, timeoutPolicy: TIME_OUT_WF, responseTimeoutSeconds: 3600, pollTimeoutSeconds: 3600, inputKeys: [ sourceRequestId, qcElementType ], outputKeys: [ state, skipped, result ], concurrentExecLimit: 100, rateLimitFrequencyInSeconds: 60, rateLimitPerFrequency: 50, ownerEmail: foobar.com, description: Sample Encoding task }字段解读该任务最多重试 3 次、指数退避从 10 秒起步且封顶 120 秒、每次重试附加最多 5 秒随机抖动整体含重试在 600 秒内必须完成单任务执行超时 1200 秒、心跳超时 3600 秒、poll 超时 3600 秒同时刻最多 100 个并发执行每分钟最多发放 50 个任务。9.2 重试一个不稳定的外部 API 调用{ name: call_payment_api, retryCount: 5, retryLogic: EXPONENTIAL_BACKOFF, retryDelaySeconds: 2, maxRetryDelaySeconds: 60, backoffJitterMs: 2000, responseTimeoutSeconds: 30, timeoutSeconds: 300, timeoutPolicy: RETRY, ownerEmail: paymentsexample.com }最多重试 5 次延迟序列为 2s、4s、8s、16s、32s——封顶 60s——且每次尝试附带最多 2 秒随机抖动。该配置可避免对已经降级的支付服务造成过度冲击。9.3 用 totalTimeoutSeconds 约束总重试预算{ name: process_order, retryCount: 10, retryLogic: FIXED, retryDelaySeconds: 5, totalTimeoutSeconds: 120, timeoutPolicy: TIME_OUT_WF, ownerEmail: ordersexample.com }每 5 秒重试一次但整个序列所有尝试合计必须在 2 分钟内完成。即使retryCount没有耗尽一旦 2 分钟预算用完任务即失败。9.4 高吞吐 worker 的抖动配置{ name: send_notification, retryCount: 3, retryLogic: FIXED, retryDelaySeconds: 1, backoffJitterMs: 3000, concurrentExecLimit: 500, ownerEmail: notificationsexample.com }当数千条通知同时失败例如下游服务宕机时抖动会将重试分散在 3 秒窗口内而不是让所有失败任务在同一瞬间再次冲击下游服务。十、总结任务定义是 Conductor 中任务行为的总契约SIMPLE任务必须注册后才能执行系统任务则可通过同名注册来定制重试/超时/限流行为。创建与更新任务定义有 UI、CLI、REST APIPOST/PUT /api/metadata/taskdefs与 SDK 四种途径其中 API 的批量创建与单条更新语义在 MetadataResource.java 中清晰可见。配置层面retryLogic与maxRetryDelaySeconds/backoffJitterMs共同决定了重试节奏timeoutPolicy决定超时后果concurrentExecLimit与rateLimitPerFrequency控制负载洪峰inputTemplate提供可被工作流覆盖的默认值——理解并组合运用这些参数是构建健壮、可治理的 Conductor 工作流的关键一步。【免费下载链接】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），仅供参考