Conductor Workflow Scheduler 快速上手基于 curl 的定时工作流全生命周期实战指南【免费下载链接】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 开源仓库中scheduler/examples/README.md为主干结合scheduler/模块的源码与示例文件完整演示从注册工作流、创建调度、预览触发时刻、查看执行历史、暂停/恢复/删除到八种真实业务场景每分钟触发、补跑 Catchup、时间窗限定、FORK/JOIN 并行、失败重试、并发叠加、参数注入、DO_WHILE 循环的端到端用法并深入讲解 6 字段 Spring Cron 语法、调度器全部配置项及其默认值与源码对应关系。Conductor 的 Workflow Scheduler 为工作流提供了一种事件驱动的定时触发能力你只需要定义一个WorkflowSchedule包含 cron 表达式、时区、目标工作流及注入参数调度器就会在每个触发时刻自动向 Conductor 提交一次工作流实例。本指南基于仓库中scheduler/examples/目录下全部 16 个可直接运行的示例文件使用curl走完调度器 API 的完整生命周期全程假设 Conductor 运行在本机8080端口。前置条件在开始之前请确认以下三点Conductor 已启动并接入支持调度的持久化后端。调度器有独立于主库的持久化模块仓库中提供了五种实现scheduler/postgres-persistenceconductor-scheduler-postgres-persistencescheduler/mysql-persistencescheduler/redis-persistencescheduler/cassandra-persistencescheduler/sqlite-persistence开启调度器开关conductor.scheduler.enabledtrue默认即开启。该开关由 SchedulerConditions.java 中的SchedulerEnabled条件注解控制关闭后调度器相关 Bean 不会被装配。HTTP 任务可用示例中的工作流大量使用HTTP任务调用外部 APItimeapi.io、jsonplaceholder 等。Conductor 内置了 HTTP 任务执行器http-task模块无需额外注册 worker若你的环境没有启用它可将示例中的type: HTTP替换为type: SIMPLE并自行注册对应 worker。部署提示仓库中的 scheduler/examples/seed.sh 展示了如何在容器化环境中自动完成注册工作流 创建调度两步它运行在 conductor-seed 容器中等待 Conductor 健康后执行可以作为 CI/CD 初始化的参考模板。Step 1 — 注册工作流调度器只负责到点触发它触发的工作流必须先注册到元数据服务。使用daily-report-workflow.json注册一个示例工作流curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json \ -d daily-report-workflow.jsondaily-report-workflow.json 定义了一个名为daily_report_workflow的工作流通过 HTTP 任务抓取https://jsonplaceholder.typicode.com/todos?userId1的示例 JSON 数据集并在outputParameters中暴露statusCode与itemCount用${fetch_report_data_ref.output.response.body.length()}计算数组长度同时设置了timeoutPolicy: TIME_OUT_WF、timeoutSeconds: 120即整个工作流 120 秒内未完成会被判定超时终止。这个工作流同时被every-minute-schedule.json和daily-report-schedule.json两个调度复用是验证环境是否就绪的最佳第一个测试对象。Step 2 — 创建调度调度Schedule是 cron 表达式与工作流之间的绑定关系。every-minute-schedule.json每分钟触发一次适合快速看到效果daily-report-schedule.json则是一个更贴近生产的工作日早上 9 点纽约时区的日报调度curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json \ -d every-minute-schedule.json | jq .成功后会返回完整的调度定义期望响应如下{ name: every-minute-demo-schedule, cronExpression: 0 * * * * *, zoneId: UTC, paused: false, nextRunTime: 1708300860000 }关键字段说明以 every-minute-schedule.json 为例字段值含义nameevery-minute-demo-schedule调度唯一名称后续暂停/恢复/删除/查询都以此定位cronExpression0 * * * * *6 字段 Spring Cron秒级精度zoneIdUTCcron 的解析时区决定本地时间归属startWorkflowRequest{name, version, input}触发时提交的工作流名称、版本与静态输入runCatchupScheduleInstancesfalse是否补跑错过的调度窗口pausedfalse创建后是否直接处于暂停态scheduleStartTime/scheduleEndTime可选epoch ms仅在该时间窗口内执行POST /api/scheduler/schedules同时承担创建与更新两种语义同名调度再次提交即为 UPSERT 更新修改 cron、时区、目标工作流均可这一行为在SchedulerService.createOrUpdateWorkflowSchedule中实现。Step 3 — 预览未来的执行时刻在真正生效前可以用nextFewSchedules接口预览任意 cron 表达式的未来触发时间点无需先创建调度curl -s http://localhost:8080/api/scheduler/nextFewSchedules?cronExpression0*****limit5 \ | jq [.[] | (. / 1000 | todate)]接口返回的是 epoch 毫秒数组这里除以 1000 后转为可读的 UTC 时间字符串。limit控制返回条数。这一能力对应的核心计算逻辑在 SchedulerService.java 的computeNextSchedule/computeNextScheduleWithZone方法中它们基于当前系统时间与上次预期运行时间结合CronSchedule模型推算下一个触发点。Step 4 — 查看执行历史等待一两分钟后调度器默认轮询间隔 100ms实际触发会有少量延迟通过执行历史搜索接口查看该调度产生的执行记录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 }每条记录对应一次调度触发scheduledTime是 cron 槽位时间workflowId是本次触发提交的工作流实例 ID。执行历史由SchedulerArchivalDAO各持久化实现如 PostgresSchedulerArchivalDAO落地支持按freeText全文检索也可通过SchedulerSearchQuery.parse支持更精细的字段过滤如scheduledTimeAfter、workflowName等。Step 5 — 暂停调度暂停会让调度器停止在后续 cron 槽位触发工作流但不会删除调度定义curl -s -X PUT http://localhost:8080/api/scheduler/schedules/every-minute-demo-schedule/pause?reasontestingpausereason为可选参数用于记录暂停原因如发布窗口故障排查。验证是否已暂停curl -s http://localhost:8080/api/scheduler/schedules/every-minute-demo-schedule | jq {paused, pausedReason}暂停语义在源码中有两层体现REST 层由 SchedulerResource.java 的pauseSchedule(name, reason)接收请求服务层SchedulerService.pauseSchedule(name, pausedReason)持久化paused与pausedReason字段此外SchedulerService还提供pauseScheduler(boolean)用于全局暂停所有调度。Step 6 — 恢复调度curl -s -X PUT http://localhost:8080/api/scheduler/schedules/every-minute-demo-schedule/resume恢复后调度器会基于当前时间重新计算nextRunTime下一个可用的 cron 槽位会照常触发。从源码看resumeSchedule与pauseSchedule都会通过ScheduleChangeListener广播onScheduleResumed/onSchedulePaused事件默认实现为 ScheduleChangeListenerStub.java不执行任何动作方便其他模块或事件总线监听。Step 7 — 列出全部调度查看所有调度及其关键状态curl -s http://localhost:8080/api/scheduler/schedules | jq [.[] | {name, cronExpression, paused, nextRunTime}]按工作流名称过滤curl -s http://localhost:8080/api/scheduler/schedules?workflowNamedaily_report_workflow | jq .此外GET /api/scheduler/schedules/search支持按名称、工作流、暂停状态等条件进行搜索。批量场景下仓库还提供了 SchedulerBulkResource.java 的PUT /api/scheduler/bulk/pause与PUT /api/scheduler/bulk/resume一次请求可对调度名列表执行批量暂停/恢复。Step 8 — 删除调度curl -s -X DELETE http://localhost:8080/api/scheduler/schedules/every-minute-demo-schedule删除后该调度的定义与后续触发全部移除执行历史记录由各持久化实现的deleteWorkflowSchedule联动清理。注意删除调度不会删除已触发的工作流实例已提交的工作流仍按自身状态机继续运行。API 参考仓库中 SchedulerResource.java 完整定义了以下 REST 端点/api/scheduler前缀MethodPathDescriptionPOST/api/scheduler/schedulesCreate or update a schedule同名即 UPSERTGET/api/scheduler/schedulesList all可选?workflowName过滤GET/api/scheduler/schedules/searchSearch schedules按名称、工作流、暂停状态过滤GET/api/scheduler/schedules/{name}Get a schedule by nameDELETE/api/scheduler/schedules/{name}Delete a schedulePUT/api/scheduler/schedules/{name}/pausePause可选?reasonPUT/api/scheduler/schedules/{name}/resumeResumeGET/api/scheduler/nextFewSchedulesPreview next N times?cronExpressionlimit5GET/api/scheduler/search/executionsSearch execution history?freeTextsize100PUT/api/scheduler/bulk/pause/.../resume批量暂停/恢复请求体为调度名列表见SchedulerBulkResourceCron 表达式格式Conductor 调度器使用6 字段 Spring Cron秒级精度而非 Linux crontab 的 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) │ │ │ │ │ │ * * * * * *常用表达式速查ExpressionMeaning0 * * * * *Every minute整分钟触发0 0 9 * * MON-FRIWeekdays at 9:00 AM工作日 9 点0 0 0 1 * *First day of every month每月 1 日零点0 0/30 9-17 * * MON-FRIEvery 30 min, business hours工作日每 30 分钟关于 cron 字段的解析与下一个触发时刻计算调度器核心实现在SchedulerService.computeNextScheduleWithZone它接收调度、当前系统时间、上次预期运行时间三个输入返回下一个运行时刻NextScheduleResult并受zoneId的时区语义约束——例如daily-report-schedule.json使用America/New_York时区0 0 9 * * MON-FRI指的是纽约当地工作日上午 9 点而非服务器本地时间。配置项详解调度器全部配置以conductor.scheduler为前缀其默认值定义在 SchedulerProperties.javaConfigurationProperties(conductor.scheduler)即 README 中的 YAML 配置与源码字段一一对应conductor: scheduler: enabled: true # 是否启用调度器默认: true由 SchedulerConditions 控制装配 polling-interval: 1000 # 轮询间隔毫秒源码默认: 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 # 每个调度的派发抖动上限源码默认: 0禁用对照源码还有几个 README 未列出但值得了解的默认参数archival-thread-count历史归档线程数默认2archival-poll-batch-size归档轮询批量大小默认5archival-maintenance-interval-record-count维护任务间隔记录数默认5000archival-maintenance-lock-seconds/archival-maintenance-lock-try-seconds归档维护分布式锁的持锁与抢锁秒数默认600/1max-schedule-jitter-ms允许的最大抖动毫秒数默认1000jitter-max-ms不得超过该上限initial-delay-ms调度器启动后的初始延迟默认15000用于等待 Conductor 各项服务就绪cache-enabled是否启用外部SchedulerCacheDAO缓存热路径查询默认false对应SchedulerOssConfiguration中的CachingSchedulerDAO条件装配。生产建议摘自原文档对于大量调度在同一个 cron 时刻同时触发的场景应将poll-batch-size提高到预期的扇出数量并将jitter-max-ms设为较小值如 200ms以削平数据库与执行线程池上的突发压力。默认poll-batch-size5意味着每轮轮询最多处理 5 个到期的调度其余顺延到下一轮。八种实战场景仓库scheduler/examples/下共有八组经过实测验证的场景每组都包含一个工作流定义文件与一个调度定义文件1. 基础触发every-minute-schedule.jsondaily-report-workflow.json每分钟触发一次通过 HTTP 抓取示例 JSON 数据集。这是环境搭建后的第一个验证用例curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d daily-report-workflow.json curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d every-minute-schedule.json2. 补跑模式catchup-schedule.jsoncatchup-workflow.json该场景将runCatchupScheduleInstances设为true。当调度器离线 N 分钟时重启后会逐个槽位补跑slot-by-slot而不是直接跳到当前时间curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d catchup-workflow.json curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d catchup-schedule.json观察方式先停止 Conductor 数分钟再重启即可看到错过的槽位按顺序依次触发。该行为由SchedulerService在重启恢复时基于lastExpectedRunTime与当前时间之间的所有 cron 槽位逐一补算实现。3. 时间窗限定bounded-schedule-template.jsonbounded-workflow.json通过scheduleStartTime/scheduleEndTimeepoch 毫秒把调度限定在某个时间窗口内执行。模板文件用__START_MS__/__END_MS__占位用sed填充后提交curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d bounded-workflow.json NOW$(($(date %s) * 1000)) END$((NOW 300000)) # 5-minute window sed s/__START_MS__/$NOW/; s/__END_MS__/$END/ bounded-schedule-template.json | \ curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d -窗口外的 cron 槽位会被跳过bounded-workflow.json中通过 HTTP 请求 timeapi.io 记录实际触发时间。4. 多步 FORK/JOINmultistep-schedule.jsonmultistep-workflow.jsonmultistep-workflow.json 演示 FORK_JOIN 并行fork_parallel_calls任务用forkTasks数组分成两个分支分别请求 UTC 与 America/New_York 两个时区的当前时间再由joinOn: [fetch_utc_time, fetch_ny_time]的 JOIN 任务汇合最终输出一个包含两个时区时间的 map。踩坑提示原文档 Gotcha时区查询参数请使用字面量/不要用%2F。Conductor 的 HTTP 任务会把百分号编码的斜杠原样传给远端 API导致 timeapi.io 将其解析为非法时区而报错。因此 URI 中应写timeZoneAmerica/New_York而非timeZoneAmerica%2FNew_York。curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d multistep-workflow.json curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d multistep-schedule.json5. 失败场景retry-schedule.jsonretry-workflow.jsonretry-workflow.json 故意调用一个不存在的 API 端点retryCount: 0请求一个全零 UUID 的 workflow 接口必然返回 404。该场景验证了调度器不因上次失败而跳过后续触发每个 cron 槽位照常产生一条新的执行记录工作流实例本身记录为FAILED但调度器历史中始终新增EXECUTED状态记录。curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d retry-workflow.json curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d retry-schedule.json6. 并发执行concurrent-schedule.jsonconcurrent-workflow.jsonconcurrent-workflow.json 模拟了一个 90 秒 WAIT 任务、每 60 秒触发一次的场景。OSS Conductor 调度器没有内置的并发执行保护因此实例会叠加堆积——这个场景正是为了让使用者理解并自行设计防护例如在应用层加分布式锁或在工作流开头用隔离/去重逻辑。踩坑提示原文档 GotchaWAIT 任务的duration必须写90s/2m/1h这类格式不能写 ISO-8601 的PT90S。原因在于 Conductor 的 DateTimeUtils.java 使用自己的正则DURATION_PATTERN解析支持d/day、h/hr/hour、m/min、s/sec组合而不是 Java 标准的Duration.parse传入PT90S会抛出IllegalArgumentException: Not valid duration。curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d concurrent-workflow.json curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d concurrent-schedule.json7. 输入参数注入input-param-schedule.jsoninput-param-workflow.json这是理解调度器数据流的核心场景。每次触发时调度器都会向工作流输入中注入五个下划线前缀的元数据字段源码证据见 SchedulerService.java 第 1027-1031 行的swr.getInput().put(...)_startedByScheduler调度名称_scheduledTimecron 槽位时间epoch 毫秒_executedTime实际派发时间epoch 毫秒_executionId本次执行记录 ID_schedulerCron调度使用的 cron 表达式。同时input-param-schedule.json 的startWorkflowRequest.input中静态声明的reportOwner、alertThreshold等键会被原样保留两者不冲突。input-param-workflow.json 用 INLINE JavaScript 任务基于_scheduledTime计算 24 小时报告窗口curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d input-param-workflow.json curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d input-param-schedule.json一次真实运行中观测到的输出scheduledTime为精确的 cron 槽位executedTime与实际派发时间相差约 837ms 的轮询开销scheduledAt: 2026-02-19T23:22:00.000Z ← exact cron slot triggeredAt: 2026-02-19T23:22:00.837Z ← actual dispatch (~837ms poll overhead) reportWindowStart: 2026-02-18T23:22:00.000Z reportWindowEnd: 2026-02-19T23:22:00.000Z8. DO_WHILE 循环变体dowhile-schedule.jsondowhile-workflow.jsondowhile-workflow.json 使用DO_WHILE任务内部循环 3 次loopCondition为if ($.iteration 3)每次迭代通过 HTTP 请求 timeapi.io 获取当前时间最后用 INLINE 任务汇总。踩坑提示原文档 GotchaDO_WHILE 的输出是按迭代序号字符串1、2、3作为键的而不是按任务引用名。要引用最后一次迭代的输出需写成${poll_loop.output.3.fetch_current_time.response.body.dateTime}而不是${poll_loop.output.fetch_current_time...}。curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d dowhile-workflow.json curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d dowhile-schedule.json并发与压测脚本原文档提到../scripts/目录包含四个来自真实并发测试的脚本需要curl、python3与运行中的 Conductor本仓库scheduler/examples/目录中未包含这些脚本文件命令与预期行为以原文档为准test-09-concurrent-write.sh — 并发注册两台机器在同一 epoch 秒对同一 Conductor 实例执行./scripts/test-09-concurrent-write.sh http://localhost:8080验证调度 UPSERT 在并发写入下的正确性。test-10-concurrent-resume.sh — 并发恢复./scripts/test-10-concurrent-resume.sh setup http://localhost:8080完成准备后两台机器同时执行 fire 命令验证一个被暂停的调度被并发恢复后恰好触发一次。test-11-thundering-herd.sh — 惊群效应./scripts/test-11-thundering-herd.sh 50 http://localhost:8080注册 N 个都在0 * * * * *触发的调度验证每个都恰好触发一次。注意需要poll-batch-size N或等待多个轮询周期默认poll-batch-size5时每轮只有 5 个调度被处理N 5 前务必先调大该参数。test-12-load-blast.py — 并发提交压测python3 scripts/test-12-load-blast.py --url http://localhost:8080 --count 25同时发起 N 个POST /api/workflow请求并输出延迟百分位两台机器可用--target $(($(date %s) 15))对齐到同一 epoch 秒后同时开跑。源码视角调度器如何工作结合 scheduler/core 模块的源码可以将调度器的运行机制归纳为如下闭环装配与启动SchedulerOssConfiguration在conductor.scheduler.enabled生效时装配SchedulerService、SchedulerDAO、SchedulerArchivalDAO、SchedulerCacheDAO等 Bean默认提供NoOpSchedulerCacheDAO与ScheduleChangeListenerStub可通过conductor.scheduler.cache.enabled和conductor.schedule-change-listener.type切换为 Redis 缓存或真实事件监听实现。轮询触发SchedulerServiceExecutorImpl按polling-interval周期轮询每个周期取poll-batch-size个到期调度触发时用startWorkflowRequest构造StartWorkflowRequest并注入五个_前缀元数据字段_startedByScheduler、_scheduledTime、_executedTime、_executionId、_schedulerCron随后交给WorkflowService提交实例并把执行记录写入SchedulerArchivalDAO。时间计算SchedulerTimeProvider.getUtcTime(zoneId)提供基于指定时区的当前时间computeNextScheduleWithZone结合CronSchedule推算下一触发点供nextFewSchedules预览与轮询调度使用。持久化调度定义、执行历史与归档清理分别由SchedulerDAO与SchedulerArchivalDAO的五个后端实现Postgres/MySQL/Redis/Cassandra/SQLite见上文的*-persistence目录完成archival-max-records与archival-max-record-threshold控制每个调度的历史保留条数与清理触发阈值。掌握了这层闭环再回头看上面的 API 与配置项就能理解为什么暂停/恢复只影响后续槽位失败不影响下次触发补跑按槽位逐个执行——这些行为全部由SchedulerService的轮询状态机决定与具体持久化后端无关。小结本文以scheduler/examples/下 16 个可直接运行的示例文件为教材走完了 Conductor 调度器的完整生命周期注册工作流 → 创建/更新调度 → 预览触发时刻 → 查看执行历史 → 暂停/恢复/删除 → 批量操作并逐一剖析了 8 种实战场景与 4 个已知踩坑点%2F时区编码、WAIT 时长格式、DO_WHILE 输出键、并发叠加行为。配合conductor.scheduler.*配置项默认值均可追溯至 SchedulerProperties.java与源码级触发链路读者应能在自己的 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），仅供参考 Conductor Workflow Scheduler 快速上手基于 curl 的定时工作流全生命周期实战指南【免费下载链接】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 开源仓库中scheduler/examples/README.md为主干结合scheduler/模块的源码与示例文件完整演示从注册工作流、创建调度、预览触发时刻、查看执行历史、暂停/恢复/删除到八种真实业务场景每分钟触发、补跑 Catchup、时间窗限定、FORK/JOIN 并行、失败重试、并发叠加、参数注入、DO_WHILE 循环的端到端用法并深入讲解 6 字段 Spring Cron 语法、调度器全部配置项及其默认值与源码对应关系。Conductor 的 Workflow Scheduler 为工作流提供了一种事件驱动的定时触发能力你只需要定义一个WorkflowSchedule包含 cron 表达式、时区、目标工作流及注入参数调度器就会在每个触发时刻自动向 Conductor 提交一次工作流实例。本指南基于仓库中scheduler/examples/目录下全部 16 个可直接运行的示例文件使用curl走完调度器 API 的完整生命周期全程假设 Conductor 运行在本机8080端口。前置条件在开始之前请确认以下三点Conductor 已启动并接入支持调度的持久化后端。调度器有独立于主库的持久化模块仓库中提供了五种实现scheduler/postgres-persistenceconductor-scheduler-postgres-persistencescheduler/mysql-persistencescheduler/redis-persistencescheduler/cassandra-persistencescheduler/sqlite-persistence开启调度器开关conductor.scheduler.enabledtrue默认即开启。该开关由 SchedulerConditions.java 中的SchedulerEnabled条件注解控制关闭后调度器相关 Bean 不会被装配。HTTP 任务可用示例中的工作流大量使用HTTP任务调用外部 APItimeapi.io、jsonplaceholder 等。Conductor 内置了 HTTP 任务执行器http-task模块无需额外注册 worker若你的环境没有启用它可将示例中的type: HTTP替换为type: SIMPLE并自行注册对应 worker。部署提示仓库中的 scheduler/examples/seed.sh 展示了如何在容器化环境中自动完成注册工作流 创建调度两步它运行在 conductor-seed 容器中等待 Conductor 健康后执行可以作为 CI/CD 初始化的参考模板。Step 1 — 注册工作流调度器只负责到点触发它触发的工作流必须先注册到元数据服务。使用daily-report-workflow.json注册一个示例工作流curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json \ -d daily-report-workflow.jsondaily-report-workflow.json 定义了一个名为daily_report_workflow的工作流通过 HTTP 任务抓取https://jsonplaceholder.typicode.com/todos?userId1的示例 JSON 数据集并在outputParameters中暴露statusCode与itemCount用${fetch_report_data_ref.output.response.body.length()}计算数组长度同时设置了timeoutPolicy: TIME_OUT_WF、timeoutSeconds: 120即整个工作流 120 秒内未完成会被判定超时终止。这个工作流同时被every-minute-schedule.json和daily-report-schedule.json两个调度复用是验证环境是否就绪的最佳第一个测试对象。Step 2 — 创建调度调度Schedule是 cron 表达式与工作流之间的绑定关系。every-minute-schedule.json每分钟触发一次适合快速看到效果daily-report-schedule.json则是一个更贴近生产的工作日早上 9 点纽约时区的日报调度curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json \ -d every-minute-schedule.json | jq .成功后会返回完整的调度定义期望响应如下{ name: every-minute-demo-schedule, cronExpression: 0 * * * * *, zoneId: UTC, paused: false, nextRunTime: 1708300860000 }关键字段说明以 every-minute-schedule.json 为例字段值含义nameevery-minute-demo-schedule调度唯一名称后续暂停/恢复/删除/查询都以此定位cronExpression0 * * * * *6 字段 Spring Cron秒级精度zoneIdUTCcron 的解析时区决定本地时间归属startWorkflowRequest{name, version, input}触发时提交的工作流名称、版本与静态输入runCatchupScheduleInstancesfalse是否补跑错过的调度窗口pausedfalse创建后是否直接处于暂停态scheduleStartTime/scheduleEndTime可选epoch ms仅在该时间窗口内执行POST /api/scheduler/schedules同时承担创建与更新两种语义同名调度再次提交即为 UPSERT 更新修改 cron、时区、目标工作流均可这一行为在SchedulerService.createOrUpdateWorkflowSchedule中实现。Step 3 — 预览未来的执行时刻在真正生效前可以用nextFewSchedules接口预览任意 cron 表达式的未来触发时间点无需先创建调度curl -s http://localhost:8080/api/scheduler/nextFewSchedules?cronExpression0*****limit5 \ | jq [.[] | (. / 1000 | todate)]接口返回的是 epoch 毫秒数组这里除以 1000 后转为可读的 UTC 时间字符串。limit控制返回条数。这一能力对应的核心计算逻辑在 SchedulerService.java 的computeNextSchedule/computeNextScheduleWithZone方法中它们基于当前系统时间与上次预期运行时间结合CronSchedule模型推算下一个触发点。Step 4 — 查看执行历史等待一两分钟后调度器默认轮询间隔 100ms实际触发会有少量延迟通过执行历史搜索接口查看该调度产生的执行记录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 }每条记录对应一次调度触发scheduledTime是 cron 槽位时间workflowId是本次触发提交的工作流实例 ID。执行历史由SchedulerArchivalDAO各持久化实现如 PostgresSchedulerArchivalDAO落地支持按freeText全文检索也可通过SchedulerSearchQuery.parse支持更精细的字段过滤如scheduledTimeAfter、workflowName等。Step 5 — 暂停调度暂停会让调度器停止在后续 cron 槽位触发工作流但不会删除调度定义curl -s -X PUT http://localhost:8080/api/scheduler/schedules/every-minute-demo-schedule/pause?reasontestingpausereason为可选参数用于记录暂停原因如发布窗口故障排查。验证是否已暂停curl -s http://localhost:8080/api/scheduler/schedules/every-minute-demo-schedule | jq {paused, pausedReason}暂停语义在源码中有两层体现REST 层由 SchedulerResource.java 的pauseSchedule(name, reason)接收请求服务层SchedulerService.pauseSchedule(name, pausedReason)持久化paused与pausedReason字段此外SchedulerService还提供pauseScheduler(boolean)用于全局暂停所有调度。Step 6 — 恢复调度curl -s -X PUT http://localhost:8080/api/scheduler/schedules/every-minute-demo-schedule/resume恢复后调度器会基于当前时间重新计算nextRunTime下一个可用的 cron 槽位会照常触发。从源码看resumeSchedule与pauseSchedule都会通过ScheduleChangeListener广播onScheduleResumed/onSchedulePaused事件默认实现为 ScheduleChangeListenerStub.java不执行任何动作方便其他模块或事件总线监听。Step 7 — 列出全部调度查看所有调度及其关键状态curl -s http://localhost:8080/api/scheduler/schedules | jq [.[] | {name, cronExpression, paused, nextRunTime}]按工作流名称过滤curl -s http://localhost:8080/api/scheduler/schedules?workflowNamedaily_report_workflow | jq .此外GET /api/scheduler/schedules/search支持按名称、工作流、暂停状态等条件进行搜索。批量场景下仓库还提供了 SchedulerBulkResource.java 的PUT /api/scheduler/bulk/pause与PUT /api/scheduler/bulk/resume一次请求可对调度名列表执行批量暂停/恢复。Step 8 — 删除调度curl -s -X DELETE http://localhost:8080/api/scheduler/schedules/every-minute-demo-schedule删除后该调度的定义与后续触发全部移除执行历史记录由各持久化实现的deleteWorkflowSchedule联动清理。注意删除调度不会删除已触发的工作流实例已提交的工作流仍按自身状态机继续运行。API 参考仓库中 SchedulerResource.java 完整定义了以下 REST 端点/api/scheduler前缀MethodPathDescriptionPOST/api/scheduler/schedulesCreate or update a schedule同名即 UPSERTGET/api/scheduler/schedulesList all可选?workflowName过滤GET/api/scheduler/schedules/searchSearch schedules按名称、工作流、暂停状态过滤GET/api/scheduler/schedules/{name}Get a schedule by nameDELETE/api/scheduler/schedules/{name}Delete a schedulePUT/api/scheduler/schedules/{name}/pausePause可选?reasonPUT/api/scheduler/schedules/{name}/resumeResumeGET/api/scheduler/nextFewSchedulesPreview next N times?cronExpressionlimit5GET/api/scheduler/search/executionsSearch execution history?freeTextsize100PUT/api/scheduler/bulk/pause/.../resume批量暂停/恢复请求体为调度名列表见SchedulerBulkResourceCron 表达式格式Conductor 调度器使用6 字段 Spring Cron秒级精度而非 Linux crontab 的 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) │ │ │ │ │ │ * * * * * *常用表达式速查ExpressionMeaning0 * * * * *Every minute整分钟触发0 0 9 * * MON-FRIWeekdays at 9:00 AM工作日 9 点0 0 0 1 * *First day of every month每月 1 日零点0 0/30 9-17 * * MON-FRIEvery 30 min, business hours工作日每 30 分钟关于 cron 字段的解析与下一个触发时刻计算调度器核心实现在SchedulerService.computeNextScheduleWithZone它接收调度、当前系统时间、上次预期运行时间三个输入返回下一个运行时刻NextScheduleResult并受zoneId的时区语义约束——例如daily-report-schedule.json使用America/New_York时区0 0 9 * * MON-FRI指的是纽约当地工作日上午 9 点而非服务器本地时间。配置项详解调度器全部配置以conductor.scheduler为前缀其默认值定义在 SchedulerProperties.javaConfigurationProperties(conductor.scheduler)即 README 中的 YAML 配置与源码字段一一对应conductor: scheduler: enabled: true # 是否启用调度器默认: true由 SchedulerConditions 控制装配 polling-interval: 1000 # 轮询间隔毫秒源码默认: 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 # 每个调度的派发抖动上限源码默认: 0禁用对照源码还有几个 README 未列出但值得了解的默认参数archival-thread-count历史归档线程数默认2archival-poll-batch-size归档轮询批量大小默认5archival-maintenance-interval-record-count维护任务间隔记录数默认5000archival-maintenance-lock-seconds/archival-maintenance-lock-try-seconds归档维护分布式锁的持锁与抢锁秒数默认600/1max-schedule-jitter-ms允许的最大抖动毫秒数默认1000jitter-max-ms不得超过该上限initial-delay-ms调度器启动后的初始延迟默认15000用于等待 Conductor 各项服务就绪cache-enabled是否启用外部SchedulerCacheDAO缓存热路径查询默认false对应SchedulerOssConfiguration中的CachingSchedulerDAO条件装配。生产建议摘自原文档对于大量调度在同一个 cron 时刻同时触发的场景应将poll-batch-size提高到预期的扇出数量并将jitter-max-ms设为较小值如 200ms以削平数据库与执行线程池上的突发压力。默认poll-batch-size5意味着每轮轮询最多处理 5 个到期的调度其余顺延到下一轮。八种实战场景仓库scheduler/examples/下共有八组经过实测验证的场景每组都包含一个工作流定义文件与一个调度定义文件1. 基础触发every-minute-schedule.jsondaily-report-workflow.json每分钟触发一次通过 HTTP 抓取示例 JSON 数据集。这是环境搭建后的第一个验证用例curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d daily-report-workflow.json curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d every-minute-schedule.json2. 补跑模式catchup-schedule.jsoncatchup-workflow.json该场景将runCatchupScheduleInstances设为true。当调度器离线 N 分钟时重启后会逐个槽位补跑slot-by-slot而不是直接跳到当前时间curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d catchup-workflow.json curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d catchup-schedule.json观察方式先停止 Conductor 数分钟再重启即可看到错过的槽位按顺序依次触发。该行为由SchedulerService在重启恢复时基于lastExpectedRunTime与当前时间之间的所有 cron 槽位逐一补算实现。3. 时间窗限定bounded-schedule-template.jsonbounded-workflow.json通过scheduleStartTime/scheduleEndTimeepoch 毫秒把调度限定在某个时间窗口内执行。模板文件用__START_MS__/__END_MS__占位用sed填充后提交curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d bounded-workflow.json NOW$(($(date %s) * 1000)) END$((NOW 300000)) # 5-minute window sed s/__START_MS__/$NOW/; s/__END_MS__/$END/ bounded-schedule-template.json | \ curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d -窗口外的 cron 槽位会被跳过bounded-workflow.json中通过 HTTP 请求 timeapi.io 记录实际触发时间。4. 多步 FORK/JOINmultistep-schedule.jsonmultistep-workflow.jsonmultistep-workflow.json 演示 FORK_JOIN 并行fork_parallel_calls任务用forkTasks数组分成两个分支分别请求 UTC 与 America/New_York 两个时区的当前时间再由joinOn: [fetch_utc_time, fetch_ny_time]的 JOIN 任务汇合最终输出一个包含两个时区时间的 map。踩坑提示原文档 Gotcha时区查询参数请使用字面量/不要用%2F。Conductor 的 HTTP 任务会把百分号编码的斜杠原样传给远端 API导致 timeapi.io 将其解析为非法时区而报错。因此 URI 中应写timeZoneAmerica/New_York而非timeZoneAmerica%2FNew_York。curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d multistep-workflow.json curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d multistep-schedule.json5. 失败场景retry-schedule.jsonretry-workflow.jsonretry-workflow.json 故意调用一个不存在的 API 端点retryCount: 0请求一个全零 UUID 的 workflow 接口必然返回 404。该场景验证了调度器不因上次失败而跳过后续触发每个 cron 槽位照常产生一条新的执行记录工作流实例本身记录为FAILED但调度器历史中始终新增EXECUTED状态记录。curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d retry-workflow.json curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d retry-schedule.json6. 并发执行concurrent-schedule.jsonconcurrent-workflow.jsonconcurrent-workflow.json 模拟了一个 90 秒 WAIT 任务、每 60 秒触发一次的场景。OSS Conductor 调度器没有内置的并发执行保护因此实例会叠加堆积——这个场景正是为了让使用者理解并自行设计防护例如在应用层加分布式锁或在工作流开头用隔离/去重逻辑。踩坑提示原文档 GotchaWAIT 任务的duration必须写90s/2m/1h这类格式不能写 ISO-8601 的PT90S。原因在于 Conductor 的 DateTimeUtils.java 使用自己的正则DURATION_PATTERN解析支持d/day、h/hr/hour、m/min、s/sec组合而不是 Java 标准的Duration.parse传入PT90S会抛出IllegalArgumentException: Not valid duration。curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d concurrent-workflow.json curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d concurrent-schedule.json7. 输入参数注入input-param-schedule.jsoninput-param-workflow.json这是理解调度器数据流的核心场景。每次触发时调度器都会向工作流输入中注入五个下划线前缀的元数据字段源码证据见 SchedulerService.java 第 1027-1031 行的swr.getInput().put(...)_startedByScheduler调度名称_scheduledTimecron 槽位时间epoch 毫秒_executedTime实际派发时间epoch 毫秒_executionId本次执行记录 ID_schedulerCron调度使用的 cron 表达式。同时input-param-schedule.json 的startWorkflowRequest.input中静态声明的reportOwner、alertThreshold等键会被原样保留两者不冲突。input-param-workflow.json 用 INLINE JavaScript 任务基于_scheduledTime计算 24 小时报告窗口curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d input-param-workflow.json curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d input-param-schedule.json一次真实运行中观测到的输出scheduledTime为精确的 cron 槽位executedTime与实际派发时间相差约 837ms 的轮询开销scheduledAt: 2026-02-19T23:22:00.000Z ← exact cron slot triggeredAt: 2026-02-19T23:22:00.837Z ← actual dispatch (~837ms poll overhead) reportWindowStart: 2026-02-18T23:22:00.000Z reportWindowEnd: 2026-02-19T23:22:00.000Z8. DO_WHILE 循环变体dowhile-schedule.jsondowhile-workflow.jsondowhile-workflow.json 使用DO_WHILE任务内部循环 3 次loopCondition为if ($.iteration 3)每次迭代通过 HTTP 请求 timeapi.io 获取当前时间最后用 INLINE 任务汇总。踩坑提示原文档 GotchaDO_WHILE 的输出是按迭代序号字符串1、2、3作为键的而不是按任务引用名。要引用最后一次迭代的输出需写成${poll_loop.output.3.fetch_current_time.response.body.dateTime}而不是${poll_loop.output.fetch_current_time...}。curl -s -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d dowhile-workflow.json curl -s -X POST http://localhost:8080/api/scheduler/schedules \ -H Content-Type: application/json -d dowhile-schedule.json并发与压测脚本原文档提到../scripts/目录包含四个来自真实并发测试的脚本需要curl、python3与运行中的 Conductor本仓库scheduler/examples/目录中未包含这些脚本文件命令与预期行为以原文档为准test-09-concurrent-write.sh — 并发注册两台机器在同一 epoch 秒对同一 Conductor 实例执行./scripts/test-09-concurrent-write.sh http://localhost:8080验证调度 UPSERT 在并发写入下的正确性。test-10-concurrent-resume.sh — 并发恢复./scripts/test-10-concurrent-resume.sh setup http://localhost:8080完成准备后两台机器同时执行 fire 命令验证一个被暂停的调度被并发恢复后恰好触发一次。test-11-thundering-herd.sh — 惊群效应./scripts/test-11-thundering-herd.sh 50 http://localhost:8080注册 N 个都在0 * * * * *触发的调度验证每个都恰好触发一次。注意需要poll-batch-size N或等待多个轮询周期默认poll-batch-size5时每轮只有 5 个调度被处理N 5 前务必先调大该参数。test-12-load-blast.py — 并发提交压测python3 scripts/test-12-load-blast.py --url http://localhost:8080 --count 25同时发起 N 个POST /api/workflow请求并输出延迟百分位两台机器可用--target $(($(date %s) 15))对齐到同一 epoch 秒后同时开跑。源码视角调度器如何工作结合 scheduler/core 模块的源码可以将调度器的运行机制归纳为如下闭环装配与启动SchedulerOssConfiguration在conductor.scheduler.enabled生效时装配SchedulerService、SchedulerDAO、SchedulerArchivalDAO、SchedulerCacheDAO等 Bean默认提供NoOpSchedulerCacheDAO与ScheduleChangeListenerStub可通过conductor.scheduler.cache.enabled和conductor.schedule-change-listener.type切换为 Redis 缓存或真实事件监听实现。轮询触发SchedulerServiceExecutorImpl按polling-interval周期轮询每个周期取poll-batch-size个到期调度触发时用startWorkflowRequest构造StartWorkflowRequest并注入五个_前缀元数据字段_startedByScheduler、_scheduledTime、_executedTime、_executionId、_schedulerCron随后交给WorkflowService提交实例并把执行记录写入SchedulerArchivalDAO。时间计算SchedulerTimeProvider.getUtcTime(zoneId)提供基于指定时区的当前时间computeNextScheduleWithZone结合CronSchedule推算下一触发点供nextFewSchedules预览与轮询调度使用。持久化调度定义、执行历史与归档清理分别由SchedulerDAO与SchedulerArchivalDAO的五个后端实现Postgres/MySQL/Redis/Cassandra/SQLite见上文的*-persistence目录完成archival-max-records与archival-max-record-threshold控制每个调度的历史保留条数与清理触发阈值。掌握了这层闭环再回头看上面的 API 与配置项就能理解为什么暂停/恢复只影响后续槽位失败不影响下次触发补跑按槽位逐个执行——这些行为全部由SchedulerService的轮询状态机决定与具体持久化后端无关。小结本文以scheduler/examples/下 16 个可直接运行的示例文件为教材走完了 Conductor 调度器的完整生命周期注册工作流 → 创建/更新调度 → 预览触发时刻 → 查看执行历史 → 暂停/恢复/删除 → 批量操作并逐一剖析了 8 种实战场景与 4 个已知踩坑点%2F时区编码、WAIT 时长格式、DO_WHILE 输出键、并发叠加行为。配合conductor.scheduler.*配置项默认值均可追溯至 SchedulerProperties.java与源码级触发链路读者应能在自己的 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），仅供参考