Conductor 工作流调度实战指南：Cron 调度配方、时区语义与调度器原理

Conductor 工作流调度实战指南Cron 调度配方、时区语义与调度器原理【免费下载链接】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/conductorConductor 是一个事件驱动的智能体工作流引擎其内置调度器Scheduler允许你以 6 段式 Spring Cron 表达式按时间槽位触发工作流执行。本文以官方 cookbook 文档docs/devguide/cookbook/workflow-scheduling.md为骨架逐条拆解仓库scheduler/examples/下全部可运行的调度配方每分钟触发、命名时区工作日、停机补跑Catchup、时间窗口约束、调度元数据注入与重叠执行演示。读完你将能够独立完成注册工作流 → 创建调度 → 预览执行时刻 → 查看执行历史 → 暂停/恢复/删除的完整生命周期并理解调度器在时区、并发与输入注入上的底层行为。先决条件调度器何时可用在动手之前需要确认运行环境满足以下条件依据 scheduler/examples/README.md 与 docs/devguide/how-tos/Workflows/scheduling-workflows.mdConductor 正在本地 8080 端口运行且使用支持调度器的持久化后端如conductor-scheduler-postgres-persistence、conductor-scheduler-mysql-persistence等参见 scheduler 目录 下的各 persistence 子模块服务端配置conductor.scheduler.enabledtrue默认开启调度器 controller 仅在该开关下挂载于/api/scheduler目标工作流定义已注册到元数据服务目标工作流依赖的 Worker 正在运行配方中使用内置于服务的HTTP任务无需额外 Worker。选择依据调度器是时钟拥有决策权的触发方式如果触发决策权来自消息/事件应改用事件编排docs/devguide/how-tos/event-bus.md。调度模型一个 Schedule 对象包含什么调度器的核心数据模型是WorkflowSchedule对应 scheduler/core 模块其字段语义在 docs/documentation/api/scheduler.md 中有完整定义字段类型必填运行期默认/行为namestring是唯一键用于创建或更新upsertcronExpressionstring两种 Cron 形式需有一种传统单表达式zoneIdstring否UTCcronSchedulesarray两种 Cron 形式需有一种非空数组时优先于cronExpression/zoneId数组内每条目的zoneId默认UTCstartWorkflowRequestobject是标准的工作流启动请求name、version、input 等runCatchupScheduleInstancesboolean否falsepausedboolean否falsepausedReasonstring否由暂停操作写入scheduleStartTimelong否下界epoch 毫秒含边界scheduleEndTimelong否上界epoch 毫秒含边界descriptionstring否用户描述createTime、updatedTime、createdBy、updatedBy、nextRunTime服务端字段否由服务端填充nextRunTime非空表示调度已保存生效Cron 表达式6 段式 Spring CronConductor 使用 Spring 的 6 段式 cron秒级精度比传统 5 段式多出秒位┌─────────────── second (0-59) │ ┌───────────── minute (0-59) │ │ ┌─────────── hour (0-23) │ │ │ ┌───────── day of month (1-31) │ │ │ │ ┌─────── month (1-12 or JAN-DEC) │ │ │ │ │ ┌───── day of week (0-7 or MON-SUN) │ │ │ │ │ │ * * * * * *表达式含义0 * * * * *每分钟0 0 9 * * MON-FRI工作日上午 9:000 0 0 1 * *每月 1 日零点0 0/30 9-17 * * MON-FRI工作日 9-17 点每 30 分钟Spring 解析器同时接受daily等宏。多表达式调度一个 Schedule 覆盖多个时区当日志、报表需要按区域时区分别触发时使用cronSchedules数组非空时它优先于cronExpression/zoneId{ name: regional-report, cronSchedules: [ {cronExpression: 0 0 9 * * MON-FRI, zoneId: America/New_York}, {cronExpression: 0 0 9 * * MON-FRI, zoneId: Europe/London} ], startWorkflowRequest: { name: daily_report_workflow, version: 1 } }Cron 求值遵循所选 IANA 时区含夏令时切换春令时前跳期间不存在的本地时间会被 cron 引擎跳过重复的本地时间按引擎的下一时刻计算。对业务敏感的调度务必在 DST 边界附近做验证。配方一每分钟触发一个简单工作流规范 fixture scheduler/examples/every-minute-schedule.json 每分钟在 UTC 触发一次daily_report_workflow{ name: every-minute-demo-schedule, cronExpression: 0 * * * * *, zoneId: UTC, startWorkflowRequest: { name: daily_report_workflow, version: 1, input: {} }, runCatchupScheduleInstances: false, paused: false }对应的目标工作流 scheduler/examples/daily-report-workflow.json 只含一个 HTTP 任务拉取示例数据集并输出状态码与条目数{ name: daily_report_workflow, description: Fetches a sample dataset on a schedule. Used as a scheduler demo., version: 1, tasks: [ { name: fetch_report_data, taskReferenceName: fetch_report_data_ref, type: HTTP, inputParameters: { http_request: { uri: https://jsonplaceholder.typicode.com/todos?userId1, method: GET, connectionTimeOut: 3000, readTimeOut: 3000 } } } ], outputParameters: { statusCode: ${fetch_report_data_ref.output.response.statusCode}, itemCount: ${fetch_report_data_ref.output.response.body.length()} }, schemaVersion: 2, restartable: true, ownerEmail: demoexample.com, timeoutPolicy: TIME_OUT_WF, timeoutSeconds: 120 }第一步注册工作流定义工作流必须先注册再创建调度curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json \ -d scheduler/examples/daily-report-workflow.json第二步创建调度使用 Conductor CLI简单 CRUD 场景conductor schedule create scheduler/examples/every-minute-schedule.json conductor schedule get every-minute-demo-schedule或使用完整 REST 接口Scheduler API 提供精确的请求体、查询参数与状态码契约curl -sS -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json \ --data-binary scheduler/examples/every-minute-schedule.json同一个POST按调度名执行创建或更新upsert成功返回200 OK及存储后的调度对象含计算字段nextRunTime。预期响应{ name: every-minute-demo-schedule, cronExpression: 0 * * * * *, zoneId: UTC, paused: false, nextRunTime: 1708300860000 }第三步预览未来执行时刻curl -s http://localhost:8080/api/scheduler/nextFewSchedules?cronExpression0*****limit5 \ | jq [.[] | (. / 1000 | todate)]注意预览端点不接受时区参数它在conductor.scheduler.schedulerTimeZone默认UTC下求值而不是调度自身的zoneId且最多返回 5 个时刻即使limit更大也会被截断。第四步查看执行历史等一两分钟后执行记录就会出现在历史中curl -s http://localhost:8080/api/scheduler/search/executions?freeTextevery-minute-demo-schedulesize5 \ | jq .results[] | {state, workflowId, scheduledTime}{ state: EXECUTED, workflowId: abc123..., scheduledTime: 1708300860000 } { state: EXECUTED, workflowId: def456..., scheduledTime: 1708300800000 }配方二命名时区中的工作日调度规范 fixture scheduler/examples/daily-report-schedule.json 在纽约时区America/New_York的工作日上午 9 点触发{ name: daily-report-schedule, cronExpression: 0 0 9 * * MON-FRI, zoneId: America/New_York, startWorkflowRequest: { name: daily_report_workflow, version: 1, input: {} }, scheduleStartTime: 0, scheduleEndTime: 0, runCatchupScheduleInstances: false, paused: false }两点关键语义官方文档明确说明IANA 时区跟随本地夏令时切换cron 引擎在该时区下求值DST 前跳期间不存在的本地时间会被跳过correlationId是字面量如果提供了startWorkflowRequest.correlationId调度器会原样复制不会插值${scheduledTime}等模板。如果每个执行都需要唯一关联 ID应在工作流内部用注入字段派生出每次运行的标识见配方五或在发起代码里构造。配方三补跑错过的 Cron 槽位Catchupfixture scheduler/examples/catchup-schedule.json 开启了runCatchupScheduleInstances: true{ name: catchup-demo-schedule, cronExpression: 0 * * * * *, zoneId: UTC, runCatchupScheduleInstances: true, paused: false, startWorkflowRequest: { name: catchup_demo_workflow, version: 1, input: {} } }行为差异与默认false对比runCatchupScheduleInstances: true调度器宕机 N 分钟后重启会逐槽位补跑每一个错过的 cron 槽不是直接跳到当前时间默认false调度器从当前时间继续推进不回放错过的槽位。观察方法先停掉 Conductor 几分钟再重启就能看到缺失槽位按顺序逐个触发执行。必须重视的副作用补跑可能在重启后产生突发burst因此目标工作流及其下游依赖必须是幂等且容量感知的。配对的 scheduler/examples/catchup-workflow.json 仅用一个 HTTP 任务记录当前时间戳用于直观观察每个槽位触发。配方四把调度约束到时间窗口模板 scheduler/examples/bounded-schedule-template.json 演示了scheduleStartTime/scheduleEndTime的用法{ name: bounded-demo-schedule, cronExpression: 0 * * * * *, zoneId: UTC, runCatchupScheduleInstances: false, scheduleStartTime: __START_MS__, scheduleEndTime: __END_MS__, startWorkflowRequest: { name: bounded_demo_workflow, version: 1, input: {} } }其中__START_MS__/__END_MS__是占位符必须替换为 epoch 毫秒数字后才能提交——模板本身故意不是合法的最终调度载荷。窗口是含边界的调度超出窗口后停止产生新执行但不会被自动删除。标准做法是先注册配对工作流再用sed填充占位符后通过管道提交curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d scheduler/examples/bounded-workflow.json NOW$(($(date %s) * 1000)) END$((NOW 300000)) # 5 分钟窗口 sed s/__START_MS__/$NOW/; s/__END_MS__/$END/ scheduler/examples/bounded-schedule-template.json | \ curl -sS -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json \ --data-binary -配对的 scheduler/examples/bounded-workflow.json 仅拉取世界时钟时间戳用于确认窗口内的槽位确实触发了。配方五在工作流内部读取调度元数据每次调度触发时调度器会将startWorkflowRequest.input原样复制后再追加五个注入字段docs/documentation/api/scheduler.md 与调度指南均有定义注入输入含义_startedByScheduler调度名称_scheduledTime预期的 cron 槽位epoch 毫秒_executedTime实际派发时刻epoch 毫秒_executionId唯一的调度执行记录 ID_schedulerCron产生本次执行的 cron 表达式与时区官方 cookbook 强调当下游系统需要每次运行唯一身份时使用${workflow.input._executionId}。注意startWorkflowRequest.correlationId是字面量复制调度器不会对其做模板插值。规范示例 scheduler/examples/input-param-schedule.json 附带静态输入reportOwner与alertThreshold这些静态键会保留{ name: input-param-demo-schedule, cronExpression: 0 * * * * *, zoneId: UTC, runCatchupScheduleInstances: false, startWorkflowRequest: { name: input_param_demo_workflow, version: 1, input: { reportOwner: platform-team, alertThreshold: 100 } } }配对工作流 scheduler/examples/input-param-workflow.json 用 INLINE JavaScript 任务从_scheduledTime与_executedTime计算 24 小时报表窗口{ name: input_param_demo_workflow, description: Demonstrates scheduler-injected workflow input. Uses _scheduledTime and _executedTime to compute a 24-hour reporting window ending at the scheduled time., version: 1, tasks: [ { name: compute_report_window, taskReferenceName: compute_report_window, type: INLINE, inputParameters: { scheduledTime: ${workflow.input._scheduledTime}, executionTime: ${workflow.input._executedTime}, evaluatorType: javascript, expression: function toISO(ms) { return new Date(ms).toISOString(); } ({ reportWindowStart: toISO($.scheduledTime - 86400000), reportWindowEnd: toISO($.scheduledTime), scheduledAt: toISO($.scheduledTime), triggeredAt: toISO($.executionTime) }) } } ], outputParameters: { reportWindowStart: ${compute_report_window.output.result.reportWindowStart}, reportWindowEnd: ${compute_report_window.output.result.reportWindowEnd}, scheduledAt: ${compute_report_window.output.result.scheduledAt}, triggeredAt: ${compute_report_window.output.result.triggeredAt} }, schemaVersion: 2, restartable: true, ownerEmail: demoexample.com, timeoutPolicy: ALERT_ONLY, timeoutSeconds: 30 }实测运行输出来自官方 walkthrough 记录scheduledAt: 2026-02-19T23:22:00.000Z ← 精确的 cron 槽位 triggeredAt: 2026-02-19T23:22:00.837Z ← 实际派发时间约 837ms 轮询开销 reportWindowStart: 2026-02-18T23:22:00.000Z reportWindowEnd: 2026-02-19T23:22:00.000ZtriggeredAt与scheduledAt的微小偏差来自调度器的轮询派发间隔这正是区分预期槽位与实际执行时刻的价值所在。配方六演示重叠执行无原生重叠策略Conductor 的调度器没有原生的重叠策略overlap policy。官方 cookbook 明确说明如果前一次工作流仍在运行下一个槽位照样会再启动一个执行。fixture 对 scheduler/examples/concurrent-schedule.json 与 scheduler/examples/concurrent-workflow.json 演示了这一行为——60 秒触发一次、工作流却要跑 90 秒{ name: concurrent-demo-schedule, cronExpression: 0 * * * * *, zoneId: UTC, runCatchupScheduleInstances: false, startWorkflowRequest: { name: concurrent_demo_workflow, version: 1, input: {} } }工作流核心部分HTTP 取开始时间 → WAIT 90 秒 → HTTP 取结束时间{ name: concurrent_demo_workflow, description: Scheduled workflow that intentionally takes longer (90s) than the firing interval (60s). Demonstrates that OSS Conductors scheduler does NOT prevent concurrent executions: each minute a new workflow starts even if the previous one is still running., version: 1, tasks: [ { name: fetch_start_time, taskReferenceName: fetch_start_time, type: HTTP, inputParameters: { http_request: { uri: https://timeapi.io/api/time/current/zone?timeZoneUTC, method: GET, connectionTimeOut: 5000, readTimeOut: 5000 } } }, { name: wait_90s, taskReferenceName: wait_90s, type: WAIT, inputParameters: { duration: 90s } }, { name: fetch_end_time, taskReferenceName: fetch_end_time, type: HTTP, inputParameters: { http_request: { uri: https://timeapi.io/api/time/current/zone?timeZoneUTC, method: GET, connectionTimeOut: 5000, readTimeOut: 5000 } } } ], outputParameters: { startedAt: ${fetch_start_time.output.response.body.dateTime}, finishedAt: ${fetch_end_time.output.response.body.dateTime} }, schemaVersion: 2, restartable: true, ownerEmail: demoexample.com, timeoutPolicy: ALERT_ONLY, timeoutSeconds: 300 }两个实测陷阱WAIT 任务时长必须是90s/2m/1h这类形式不能是 ISO-8601 的PT90S。Conductor 的DateTimeUtils.parseDuration使用自己的正则解析不是 Java 的Duration解析器由于实例会不断叠加必须自己设计并发与幂等策略例如在业务侧用_executionId做去重或用工作流内部逻辑限流。更多规范 fixture多步骤、重试与循环ischeduler/examples/目录下还有更多经过实测的配方注册顺序一律是先注册工作流再创建调度多步骤 FORK/JOINmultistep-workflow.json multistep-schedule.json两个并行 HTTP 调用UTC 与 America/New_YorkJOIN 后汇成一张输出 map。陷阱时区查询参数里要写字面量/America/New_York不要用%2F编码——Conductor 的 HTTP 任务会把百分号编码的斜杠原样传给远端 API导致时区无效失败场景retry-workflow.json retry-schedule.json工作流必然失败404但调度器不关心前一次结果——每个 tick 仍然产生一条新的调度执行记录即使工作流本身记录为FAILED。这验证了每次槽位独立触发的语义DO_WHILE 变体dowhile-workflow.json dowhile-schedule.json内部循环 3 次拉取当前时间。陷阱DO_WHILE 的输出按迭代号字符串键控1、2、3而不是按任务引用名键控取最后一次迭代输出要写${poll_loop.output.3.fetch_current_time.response.body.dateTime}。完整的调度运维 REST 接口所有调度接口挂载在/api/schedulerdocs/documentation/api/scheduler.md仅在conductor.scheduler.enabledtrue时存在成功均返回200 OK方法路径说明POST/api/scheduler/schedules创建或更新调度GET/api/scheduler/schedules列出全部可选?workflowName过滤GET/api/scheduler/schedules/search搜索调度按名称、工作流、paused 过滤GET/api/scheduler/schedules/{name}按名称获取单个调度DELETE/api/scheduler/schedules/{name}删除调度PUT/api/scheduler/schedules/{name}/pause?reason暂停reason可选PUT/api/scheduler/schedules/{name}/resume恢复GET/api/scheduler/nextFewSchedules预览未来 N 次执行?cronExpressionlimit5GET/api/scheduler/search/executions搜索执行历史?freeTextsize100PUT/api/scheduler/bulk/pause//api/scheduler/bulk/resume批量暂停/恢复请求体为调度名 JSON 数组GET/api/scheduler/admin/requeue/pause/resume调度器内部恢复/调试端点需访问控制搜索调度的查询参数workflowName、scheduleName、paused、freeText默认*、start默认0、size默认100、sort。执行历史搜索返回SearchResultWorkflowScheduleExecutionModel每条记录含调度执行 ID、计划与实际执行时间、工作流名/ID、状态及失败详情。CLI 与 REST 对应的常用运维操作conductor schedule list conductor schedule pause every-minute-demo-schedule conductor schedule resume every-minute-demo-schedule conductor schedule delete every-minute-demo-schedulecurl http://localhost:8080/api/scheduler/schedules/search?pausedfalsesize20 curl http://localhost:8080/api/scheduler/search/executions?freeTextevery-minute-demo-schedulesize20暂停后应核对存储的paused状态并确认下一个 cron 槽位后没有新执行产生恢复后再确认新执行出现并检查五个注入字段是否齐全。调度器服务端配置调度器行为由conductor.scheduler.*配置项控制官方 walkthrough 中的完整示例conductor: scheduler: enabled: true # 默认: true polling-interval: 1000 # 轮询间隔 ms默认: 100 polling-thread-count: 1 # 默认: 1 poll-batch-size: 5 # 每轮处理的调度数量默认: 5 scheduler-time-zone: UTC # 默认: UTC archival-max-records: 5 # 每个调度保留的历史记录条数默认: 5 archival-max-record-threshold: 10 # 超过该阈值触发清理默认: 10 jitter-max-ms: 0 # 每个调度的派发抖动 ms默认: 0关闭实践建议当大量调度在同一 cron 时刻齐发时把poll-batch-size提高到预期的扇出数量并给jitter-max-ms设一个较小值如 200以平滑数据库与执行器线程池上的突发负载。相关的并发/负载测试脚本如同槽位 N 个调度各触发一次的 thundering-herd 验证、双机并发注册的 UPSERT 正确性验证收录在scheduler/examples/对应../scripts/脚本目录中。已知限制与设计取舍官方文档明确列举的调度器限制设计生产方案前必须知晓无原生重叠策略前序工作流未结束时下一槽位仍会启动新执行并发/幂等策略需在工作流或下游自行实现无立即运行run-now与手动回填端点临时运行请直接启动目标工作流并在输入中显式传递预期窗口预览限制仅支持单条 cron、最多 5 个时刻、且使用服务端调度时区而非调度自身的zoneIdcorrelationId是字面量不做模板展开SDK 一致性Java、Python、TypeScript、Go SDK 可通过生成式或底层客户端调用 REST 面但本仓库没有在所有 SDK 间定义一致的高层调度 APIREST 是可移植的完整接口CLI 发布版也未完整暴露每个调度字段与操作。进一步阅读调度语义与操作指南docs/devguide/how-tos/Workflows/scheduling-workflows.mdScheduler REST 契约请求体、查询参数、状态码docs/documentation/api/scheduler.md完整本地演练8 个实测场景 负载测试脚本scheduler/examples/README.md调度器源码与各持久化实现scheduler/core、scheduler/postgres-persistence、scheduler/mysql-persistence、scheduler/redis-persistence、scheduler/cassandra-persistence、scheduler/sqlite-persistence【免费下载链接】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），仅供参考