企业官网建设流程全解析

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端口。

前置条件

在开始之前，请确认以下三点：

1. Conductor 已启动并接入支持调度的持久化后端。调度器有独立于主库的持久化模块，仓库中提供了五种实现：

   • scheduler/postgres-persistence（conductor-scheduler-postgres-persistence）
   • scheduler/mysql-persistence
   • scheduler/redis-persistence
   • scheduler/cassandra-persistence
   • scheduler/sqlite-persistence

2. 开启调度器开关：conductor.scheduler.enabled=true（默认即开启）。该开关由 SchedulerConditions.java 中的SchedulerEnabled条件注解控制，关闭后调度器相关 Bean 不会被装配。

3. HTTP 任务可用：示例中的工作流大量使用HTTP任务调用外部 API（timeapi.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.json

daily-report-workflow.json 定义了一个名为daily_report_workflow的工作流，通过 HTTP 任务抓取https://jsonplaceholder.typicode.com/todos?userId=1的示例 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 为例）：

POST /api/scheduler/schedules同时承担"创建"与"更新"两种语义：同名调度再次提交即为 UPSERT 更新（修改 cron、时区、目标工作流均可），这一行为在SchedulerService.createOrUpdateWorkflowSchedule中实现。

Step 3 — 预览未来的执行时刻

在真正生效前，可以用nextFewSchedules接口预览任意 cron 表达式的未来触发时间点，无需先创建调度：

curl -s "http://localhost:8080/api/scheduler/nextFewSchedules?cronExpression=0+*+*+*+*+*&limit=5" \ | jq '[.[] | (. / 1000 | todate)]'

接口返回的是 epoch 毫秒数组，这里除以 1000 后转为可读的 UTC 时间字符串。limit控制返回条数。这一能力对应的核心计算逻辑在 SchedulerService.java 的computeNextSchedule/computeNextScheduleWithZone方法中，它们基于当前系统时间与上次预期运行时间，结合CronSchedule模型推算下一个触发点。

Step 4 — 查看执行历史

等待一两分钟后（调度器默认轮询间隔 100ms，实际触发会有少量延迟），通过执行历史搜索接口查看该调度产生的执行记录：

curl -s "http://localhost:8080/api/scheduler/search/executions?freeText=every-minute-demo-schedule&size=5" \ | 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?reason=testing+pause"

reason为可选参数，用于记录暂停原因（如"发布窗口""故障排查"）。验证是否已暂停：

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?workflowName=daily_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前缀）：

Cron 表达式格式

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) │ │ │ │ │ │ * * * * * *

常用表达式速查：

关于 cron 字段的解析与下一个触发时刻计算，调度器核心实现在SchedulerService.computeNextScheduleWithZone，它接收调度、当前系统时间、上次预期运行时间三个输入，返回下一个运行时刻（NextScheduleResult），并受zoneId的时区语义约束——例如daily-report-schedule.json使用America/New_York时区，0 0 9 * * MON-FRI指的是纽约当地工作日上午 9 点，而非服务器本地时间。

配置项详解

调度器全部配置以conductor.scheduler为前缀，其默认值定义在 SchedulerProperties.java（@ConfigurationProperties("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：历史归档线程数，默认2；
• archival-poll-batch-size：归档轮询批量大小，默认5；
• archival-maintenance-interval-record-count：维护任务间隔记录数，默认5000；
• archival-maintenance-lock-seconds/archival-maintenance-lock-try-seconds：归档维护分布式锁的持锁与抢锁秒数，默认600/1；
• max-schedule-jitter-ms：允许的最大抖动毫秒数，默认1000（jitter-max-ms不得超过该上限）；
• initial-delay-ms：调度器启动后的初始延迟，默认15000，用于等待 Conductor 各项服务就绪；
• cache-enabled：是否启用外部SchedulerCacheDAO缓存热路径查询，默认false（对应SchedulerOssConfiguration中的CachingSchedulerDAO条件装配）。

生产建议（摘自原文档）：对于大量调度在同一个 cron 时刻同时触发的场景，应将poll-batch-size提高到预期的扇出数量，并将jitter-max-ms设为较小值（如 200ms），以削平数据库与执行线程池上的突发压力。默认poll-batch-size=5意味着每轮轮询最多处理 5 个到期的调度，其余顺延到下一轮。

八种实战场景

仓库scheduler/examples/下共有八组经过实测验证的场景，每组都包含一个工作流定义文件与一个调度定义文件：

1. 基础触发（every-minute-schedule.json+daily-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.json

2. 补跑模式（catchup-schedule.json+catchup-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.json+bounded-workflow.json）

通过scheduleStartTime/scheduleEndTime（epoch 毫秒）把调度限定在某个时间窗口内执行。模板文件用__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/JOIN（multistep-schedule.json+multistep-workflow.json）

multistep-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 中应写timeZone=America/New_York而非timeZone=America%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.json

5. 失败场景（retry-schedule.json+retry-workflow.json）

retry-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.json

6. 并发执行（concurrent-schedule.json+concurrent-workflow.json）

concurrent-workflow.json 模拟了一个 90 秒 WAIT 任务、每 60 秒触发一次的场景。OSS Conductor 调度器没有内置的并发执行保护，因此实例会叠加堆积——这个场景正是为了让使用者理解并自行设计防护（例如在应用层加分布式锁，或在工作流开头用隔离/去重逻辑）。

踩坑提示（原文档 Gotcha）：WAIT 任务的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.json

7. 输入参数注入（input-param-schedule.json+input-param-workflow.json）

这是理解调度器数据流的核心场景。每次触发时，调度器都会向工作流输入中注入五个下划线前缀的元数据字段（源码证据见 SchedulerService.java 第 1027-1031 行的swr.getInput().put(...)）：

• _startedByScheduler：调度名称；
• _scheduledTime：cron 槽位时间（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.000Z

8. DO_WHILE 循环变体（dowhile-schedule.json+dowhile-workflow.json）

dowhile-workflow.json 使用DO_WHILE任务内部循环 3 次（loopCondition为if ($.iteration < 3)），每次迭代通过 HTTP 请求 timeapi.io 获取当前时间，最后用 INLINE 任务汇总。

踩坑提示（原文档 Gotcha）：DO_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-size=5时每轮只有 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 模块的源码，可以将调度器的运行机制归纳为如下闭环：

1. 装配与启动：SchedulerOssConfiguration在conductor.scheduler.enabled生效时装配SchedulerService、SchedulerDAO、SchedulerArchivalDAO、SchedulerCacheDAO等 Bean；默认提供NoOpSchedulerCacheDAO与ScheduleChangeListenerStub，可通过conductor.scheduler.cache.enabled和conductor.schedule-change-listener.type切换为 Redis 缓存或真实事件监听实现。
2. 轮询触发：SchedulerServiceExecutorImpl按polling-interval周期轮询，每个周期取poll-batch-size个到期调度；触发时用startWorkflowRequest构造StartWorkflowRequest并注入五个_前缀元数据字段（_startedByScheduler、_scheduledTime、_executedTime、_executionId、_schedulerCron），随后交给WorkflowService提交实例，并把执行记录写入SchedulerArchivalDAO。
3. 时间计算：SchedulerTimeProvider.getUtcTime(zoneId)提供基于指定时区的当前时间；computeNextScheduleWithZone结合CronSchedule推算下一触发点，供nextFewSchedules预览与轮询调度使用。
4. 持久化：调度定义、执行历史与归档清理分别由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