Conductor 任务定义（Task Definition）创建与更新完全指南：从 UI、CLI、API 到重试与限流配置

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 中所有SIMPLE（Worker）任务得以在 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，其中明确给出了TimeoutPolicy（RETRY、TIME_OUT_WF、ALERT_ONLY）与RetryLogic（FIXED、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 参考文档。下面按参考文档整理完整字段表：

补充说明（来自 TaskDef.java 的实现细节）：

• name上标注了@NotEmpty(message = "TaskDef name cannot be null or empty")，ownerEmail上有@OwnerEmailMandatoryConstraint校验，responseTimeoutSeconds最小值为 1 秒，retryCount、pollTimeoutSeconds、maxRetryDelaySeconds、backoffJitterMs、totalTimeoutSeconds均要求>= 0，backoffScaleFactor要求>= 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时生效。

maxRetryDelaySeconds 封顶示例

以EXPONENTIAL_BACKOFF、retryDelaySeconds=1、maxRetryDelaySeconds=3为例：

对应到源码注释（TaskDef.java）：20 次重试、初始 1 秒、封顶 600 秒时，退避序列为 1, 2, 4, 8, …, 600, 600, 600, …，而不是无限增长。

backoffJitterMs 抖动示例

backoffJitterMs会在最终延迟上追加[0, backoffJitterMs]毫秒内的均匀随机值，将多个失败 worker 的重试分散到不同时间点，防止惊群效应（thundering herd）。例如retryDelaySeconds=2、backoffJitterMs=1000，则每次重试会在失败后 2 000 ms～3 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 与 outputKeys

• inputKeys和outputKeys可视为任务的参数与返回值：把任务定义想象成一个接口(value1, value2 .. valueN) someTaskDefinition(key1, key2 .. keyN);。
• 但当前这些参数并非严格强制校验——两者目前主要充当任务复用的文档说明，工作流中的任务不必覆盖任务定义里的全部键。
• 从发展角度看，未来可以扩展为类似编程语言接口的严格模板约束。

使用 inputTemplate

inputTemplate允许定义默认输入值，这些值可以被工作流中提供的值覆盖。例如在任务定义中：

"inputTemplate": { "url": "https://some_url:7004" }

然后在工作流定义中使用该任务时，既可以沿用默认的url，也可以在任务的inputParameters中覆盖：

"inputParameters": { "url": "${workflow.input.some_new_url}" }

七、创建与更新任务定义的三种途径

7.1 使用 Conductor UI

创建任务定义：

1. 在左侧导航中打开Definitions，选择Task。
2. 点击Define task。
3. 在Task表单中配置任务，或打开Code标签页直接编辑 JSON。完整参数参考 Task Definitions。
4. 点击Save保存。

更新任务定义：

1. 在左侧导航中打开Definitions，选择Task，然后选中要更新的任务定义。
2. 在Task表单或Code标签页中修改任务。完整参数参考 Task Definitions。
3. 点击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 List<TaskDef> 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": "foo@bar.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": "payments@example.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": "orders@example.com" }

每 5 秒重试一次，但整个序列（所有尝试合计）必须在 2 分钟内完成。即使retryCount没有耗尽，一旦 2 分钟预算用完任务即失败。

9.4 高吞吐 worker 的抖动配置

{ "name": "send_notification", "retryCount": 3, "retryLogic": "FIXED", "retryDelaySeconds": 1, "backoffJitterMs": 3000, "concurrentExecLimit": 500, "ownerEmail": "notifications@example.com" }

当数千条通知同时失败（例如下游服务宕机）时，抖动会将重试分散在 3 秒窗口内，而不是让所有失败任务在同一瞬间再次冲击下游服务。

十、总结

任务定义是 Conductor 中任务行为的总契约：SIMPLE任务必须注册后才能执行，系统任务则可通过同名注册来定制重试/超时/限流行为。创建与更新任务定义有 UI、CLI、REST API（POST/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