Cloudflare Workflows 开发实战基于 cloudflare-deploy 技能构建持久化多步骤应用【免费下载链接】skillsSkills Catalog for Codex项目地址: https://gitcode.com/GitHub_Trending/skills4/skills本文以 cloudflare-deploy 技能 中的 Workflows 参考文档为骨架系统讲解 Cloudflare Workflows 的核心概念、配置方式、Step API 与常用编排模式。读完你将掌握如何用WorkflowEntrypoint编写具备自动重试、状态持久化、长时间睡眠与外部事件等待能力的多步骤任务并能通过 Wrangler CLI 与 REST API 完成实例的创建、管理与监控。这套能力适用于用户生命周期提醒、数据处理管线、人工审批等分钟到周量级的后台任务场景。一、Workflows 是什么从 Worker 到持久化工作流在 SKILL.md 的产品决策树中Long-running multi-step jobs长时多步骤任务明确指向references/workflows/目录。Cloudflare Workflows 是构建在 Workers 之上的持久化多步骤应用运行环境其核心能力包括链式执行步骤并内置自动重试逻辑在步骤之间持久化状态分钟级到周级失败时不丢失已完成的进度等待外部事件 / 人工审批睡眠不消耗资源处于 waiting 状态的实例不占用并发额度。根据 README.mdWorkflows 在Free 与 Paid Workers 套餐中均可用区别仅在于配额详见下文限额与定价。与同为有状态方案的 Durable Objects 相比Workflows 更偏长时任务编排与消息驱动的 Queues 相比Workflows 内置步骤状态机与重试而 Workers 则是承载 Workflow 实例的入口环境。二、核心概念Workflow、Instance、Steps 与 State参考文档定义了四个必须理解的基础概念概念说明Workflow一个继承WorkflowEntrypoint并实现run方法的类Instance一次独立的执行拥有唯一 ID 与独立的运行状态Steps通过step.do()定义的、可独立重试的单元——可以是 API 调用、数据库查询、AI 调用等State由步骤返回值持久化而来步骤名即缓存键同名步骤在同一实例中只会执行一次理解步骤名即缓存键至关重要Workflows 把每个步骤的返回值按步骤名持久化重放或重试时直接复用成功结果这正是失败步骤不重跑已成功步骤durability的底层机制。三、快速开始一个 7 天提醒工作流参考文档给出的最小示例是一个查询用户 → 睡眠 7 天 → 发送提醒的工作流import { WorkflowEntrypoint, WorkflowStep, WorkflowEvent } from cloudflare:workers; type Env { MY_WORKFLOW: Workflow; DB: D1Database }; type Params { userId: string }; export class MyWorkflow extends WorkflowEntrypointEnv, Params { async run(event: WorkflowEventParams, step: WorkflowStep) { const user await step.do(fetch user, async () { return await this.env.DB.prepare(SELECT * FROM users WHERE id ?) .bind(event.params.userId).first(); }); await step.sleep(wait 7 days, 7 days); await step.do(send reminder, async () { await sendEmail(user.email, Reminder!); }); } }要点WorkflowEntrypointEnv, Params是两个泛型参数Env声明可访问的绑定Params声明实例入参类型event携带params、instanceId、timestamp等信息step提供do()/sleep()/sleepUntil()/waitForEvent()四个核心 API。四、配置 Workflowwrangler.jsonc 与 Step 配置4.1 wrangler.jsonc 基础配置参考 configuration.md在wrangler.jsonc中注册工作流{ name: my-worker, main: src/index.ts, compatibility_date: 2025-01-01, // 新项目使用当前日期 observability: { enabled: true // 开启 Workflows 仪表盘与结构化日志 }, workflows: [ { name: my-workflow, // Workflow 名称 binding: MY_WORKFLOW, // 注入到 env 的绑定名 class_name: MyWorkflow // TS 类名 // script_name: other-worker // 跨脚本调用时填写 } ], limits: { cpu_ms: 300000 // CPU 上限 5 分钟默认 30 秒 } }各字段含义observability.enabled开启后可在 Cloudflare 仪表盘查看工作流运行情况与结构化日志workflows[]可注册一个或多个工作流每项包含name工作流名、bindingenv中的绑定标识、class_name对应 TS 类名limits.cpu_ms单步 CPU 时间上限默认 30000ms30 秒最大可配 300000ms5 分钟。注意这里衡量的是活跃计算时间网络请求、数据库查询与睡眠不计入 CPU 时间详见常见错误一节。4.2 Step 配置重试与超时step.do()的第二个参数可以传入重试与超时配置// 基础步骤 const data await step.do(step name, async () ({ result: value })); // 带重试配置的步骤 await step.do(api call, { retries: { limit: 10, // 默认 5或 Infinity delay: 10 seconds, // 默认 10000ms backoff: exponential // constant | linear | exponential }, timeout: 30 minutes // 单次尝试超时默认 10 分钟 }, async () { const res await fetch(https://api.example.com/data); if (!res.ok) throw new Error(Failed); return res.json(); });重试参数取值backoff支持constant固定间隔、linear线性递增、exponential指数退避推荐用于易抖动的外部 API。4.3 并行、条件与动态步骤并行步骤用Promise.all并发执行多个step.doconst [user, settings] await Promise.all([ step.do(fetch user, async () this.env.KV.get(user:${id})), step.do(fetch settings, async () this.env.KV.get(settings:${id})) ]);条件步骤——关键原则是条件必须基于确定性的输入步骤输出或事件参数const config await step.do(fetch config, async () this.env.KV.get(flags, { type: json }) ); // ✅ 正确基于步骤输出的确定性分支 if (config.enableEmail) { await step.do(send email, async () sendEmail()); } // ❌ 错误在步骤外使用非确定性 Date.now() if (Date.now() deadline) { /* BAD */ }动态步骤循环——可基于前一步输出动态生成步骤名const files await step.do(list files, async () this.env.BUCKET.list() ); for (const file of files.objects) { await step.do(process ${file.key}, async () { const obj await this.env.BUCKET.get(file.key); return processData(await obj.arrayBuffer()); }); }4.4 多工作流与跨脚本绑定注册多个工作流时每个类各自继承WorkflowEntrypoint并拥有独立的Params类型{ workflows: [ {name: user-onboarding, binding: USER_ONBOARDING, class_name: UserOnboarding}, {name: data-processing, binding: DATA_PROCESSING, class_name: DataProcessing} ] }跨脚本绑定Worker A 定义工作流Worker B 通过script_name引用并调用// Worker B调用方 { workflows: [{ name: billing-workflow, binding: BILLING, script_name: billing-worker // 指向 Worker A }] }4.5 在步骤中使用 Cloudflare 绑定工作流内通过this.env访问 KV、D1、R2、Workers AI、Vectorize 等全部绑定type Env { MY_WORKFLOW: Workflow; KV: KVNamespace; DB: D1Database; BUCKET: R2Bucket; AI: Ai; VECTORIZE: VectorizeIndex; }; await step.do(use bindings, async () { const kv await this.env.KV.get(key); const db await this.env.DB.prepare(SELECT * FROM users).first(); const file await this.env.BUCKET.get(file.txt); const ai await this.env.AI.run(cf/meta/llama-2-7b-chat-int8, { prompt: Hi }); });4.6 从 Pages Functions 触发Pages Functions 可通过 service binding 触发工作流并在wrangler.jsonc的service_bindings中配置// functions/_middleware.ts export const onRequest: PagesFunctionEnv async ({ env, request }) { const instance await env.MY_WORKFLOW.create({ params: { url: request.url } }); return new Response(Started ${instance.id}); };五、Step API 详解do / sleep / sleepUntil / waitForEvent参考 api.md四个 API 的签名与语义如下// step.do() —— 执行一个可重试步骤返回值为持久化状态 const result await step.do(step name, async () { /* logic */ }); const result await step.do(step name, { retries, timeout }, async () {}); // step.sleep() —— 相对时间睡眠字符串或毫秒 await step.sleep(description, 1 hour); await step.sleep(description, 5000); // ms // step.sleepUntil() —— 绝对时间点唤醒 await step.sleepUntil(description, Date.parse(2024-12-31)); // step.waitForEvent() —— 等待外部事件webhook / 审批 const data await step.waitForEventPayloadType(wait, {event: webhook-type, timeout: 24h}); // 默认 24h最大 365d try { const event await step.waitForEvent(wait, { event: approval, timeout: 1h }); } catch (e) { /* 超时处理 */ }时间单位支持second、minute、hour、day、week、month、year最大 365 天睡眠中的实例不计入并发限制。waitForEvent默认超时 24 小时、最大 365 天超时会抛出异常务必用try-catch包裹以优雅降级。六、实例管理创建、查询、控制与事件6.1 创建实例// 单实例id 可选省略则自动生成 const instance await env.MY_WORKFLOW.create({id: crypto.randomUUID(), params: { userId: user123 }}); // 自定义保留期默认免费 3 天 / 付费 30 天 const instance await env.MY_WORKFLOW.create({ id: crypto.randomUUID(), params: { userId: user123 }, retention: 30 days // 覆盖默认保留期 }); // 批量创建最多 100 个幂等跳过已存在的 ID const instances await env.MY_WORKFLOW.createBatch([ {id: user1, params: {name: John}}, {id: user2, params: {name: Jane}} ]);6.2 查询与控制const instance await env.MY_WORKFLOW.get(instance-id); // 状态查询queued | running | paused | errored | terminated | complete | waiting | waitingForPause | unknown const status await instance.status(); // {status, error?, output?} // 控制操作 await instance.pause(); await instance.resume(); await instance.terminate(); await instance.restart(); // 发送事件type 必须与 waitForEvent 的 event 匹配 await instance.sendEvent({type: approval, payload: { approved: true }});七、触发工作流的五种方式从 api.md 可以看到工作流可从多种入口触发// 1. 从 Worker 的 fetch 处理器触发 export default { async fetch(req, env) { const instance await env.MY_WORKFLOW.create({id: crypto.randomUUID(), params: { userId: user123 }}); return Response.json({ id: instance.id }); }}; // 2. 从 Queue 消费触发 export default { async queue(batch, env) { for (const msg of batch.messages) { await env.MY_WORKFLOW.create({id: job-${msg.id}, params: msg.body}); } }}; // 3. 从 Cron 定时触发 export default { async scheduled(event, env) { await env.CLEANUP_WORKFLOW.create({id: cleanup-${Date.now()}, params: { timestamp: event.scheduledTime }}); }}; // 4. 从另一个工作流触发非阻塞 export class ParentWorkflow extends WorkflowEntrypointEnv, Params { async run(event, step) { const child await step.do(start child, async () await this.env.CHILD_WORKFLOW.create({id: child-${event.instanceId}, params: {}})); } }第 5 种即上一节提到的 Pages Functions service binding 触发。八、错误处理、幂等性与类型约束8.1 NonRetryableError 与普通异常参考文档提供了区分可重试与不可重试错误的范式import { NonRetryableError } from cloudflare:workers; await step.do(validate, async () { if (!event.params.paymentMethod) throw new NonRetryableError(Payment method required); const res await fetch(https://api.example.com/charge, { method: POST }); if (res.status 401) throw new NonRetryableError(Invalid credentials); // 不重试 if (!res.ok) throw new Error(Retryable failure); // 会重试 return res.json(); }); // 捕获错误后继续后续步骤如清理 try { await step.do(risky op, async () { throw new NonRetryableError(Failed); }); } catch (e) { await step.do(cleanup, async () {}); }幂等性Check-then-Execute由于失败步骤会重试重复执行可能造成重复扣款等副作用标准做法是先检查再执行await step.do(charge, async () { const sub await fetch(https://api/subscriptions/${id}).then(r r.json()); if (sub.charged) return sub; // 已完成则直接返回 return await fetch(https://api/subscriptions/${id}, {method: POST, body: JSON.stringify({ amount: 10.0 })}).then(r r.json()); });8.2 类型约束Rpc.SerializableParams与步骤返回值必须是Rpc.SerializableT可序列化类型// ✅ 合法类型 type ValidParams { userId: string; count: number; tags: string[]; metadata: Recordstring, unknown; }; // ❌ 非法类型 type InvalidParams { callback: () void; // 函数不可序列化 symbol: symbol; // Symbol 不可序列化 circular: any; // 循环引用不允许 }; // 步骤返回值遵循同样规则 const result await step.do(fetch, async () { return { userId: 123, data: [1, 2, 3] }; // ✅ 纯对象 });8.3 睡眠与调度// 相对时间 await step.sleep(wait 1 hour, 1 hour); await step.sleep(wait 30 days, 30 days); await step.sleep(wait 5s, 5000); // 毫秒 // 绝对时间 await step.sleepUntil(launch date, Date.parse(24 Oct 2024 13:00:00 UTC)); await step.sleepUntil(deadline, new Date(2024-12-31T23:59:59Z));8.4 参数传递// 创建时传入参数 const instance await env.MY_WORKFLOW.create({ id: crypto.randomUUID(), params: { userId: user123, email: userexample.com } }); // 在 run 中读取参数与实例元信息 async run(event: WorkflowEventParams, step: WorkflowStep) { const userId event.params.userId; const instanceId event.instanceId; const createdAt event.timestamp; }命令行触发方式npx wrangler workflows trigger my-workflow {userId:user123}。九、Wrangler CLI 与 REST API9.1 创建项目并部署npm create cloudflarelatest my-workflow -- --template cloudflare/workflows-starter npx wrangler deploy9.2 实例管理命令npx wrangler workflows list npx wrangler workflows trigger my-workflow {userId:user123} npx wrangler workflows instances list my-workflow npx wrangler workflows instances describe my-workflow instance-id npx wrangler workflows instances pause/resume/terminate my-workflow instance-id9.3 REST API# 创建实例 curl -X POST https://api.cloudflare.com/client/v4/accounts/{account_id}/workflows/{workflow_name}/instances -H Authorization: Bearer {token} -d {id:custom-id,params:{userId:user123}} # 查询状态 curl https://api.cloudflare.com/client/v4/accounts/{account_id}/workflows/{workflow_name}/instances/{instance_id}/status -H Authorization: Bearer {token} # 发送事件 curl -X POST https://api.cloudflare.com/client/v4/accounts/{account_id}/workflows/{workflow_name}/instances/{instance_id}/events -H Authorization: Bearer {token} -d {type:approval,payload:{approved:true}}十、典型工作流模式patterns.md参考 patterns.md以下是四个可直接套用的真实场景。10.1 图像处理管线AI 人工审批export class ImageProcessingWorkflow extends WorkflowEntrypointEnv, Params { async run(event, step) { const imageData await step.do(fetch, async () (await this.env.BUCKET.get(event.params.imageKey)).arrayBuffer()); const description await step.do(generate description, async () await this.env.AI.run(cf/llava-hf/llava-1.5-7b-hf, {image: Array.from(new Uint8Array(imageData)), prompt: Describe this image, max_tokens: 50}) ); await step.waitForEvent(await approval, { event: approved, timeout: 24h }); await step.do(publish, async () await this.env.BUCKET.put(public/${event.params.imageKey}, imageData)); } }10.2 用户生命周期试用期提醒export class UserLifecycleWorkflow extends WorkflowEntrypointEnv, Params { async run(event, step) { await step.do(welcome email, async () await sendEmail(event.params.email, Welcome!)); await step.sleep(trial period, 7 days); const hasConverted await step.do(check conversion, async () { const user await this.env.DB.prepare(SELECT subscription_status FROM users WHERE id ?) .bind(event.params.userId).first(); return user.subscription_status active; }); if (!hasConverted) await step.do(trial expiration email, async () await sendEmail(event.params.email, Trial ending)); } }10.3 数据管线抽取-转换-存储-装载export class DataPipelineWorkflow extends WorkflowEntrypointEnv, Params { async run(event, step) { const rawData await step.do(extract, {retries: { limit: 10, delay: 30s, backoff: exponential }}, async () { const res await fetch(event.params.sourceUrl); if (!res.ok) throw new Error(Fetch failed); return res.json(); }); const transformed await step.do(transform, async () rawData.map(item ({ id: item.id, normalized: normalizeData(item) })) ); const dataRef await step.do(store, async () { const key processed/${Date.now()}.json; await this.env.BUCKET.put(key, JSON.stringify(transformed)); return { key }; }); await step.do(load, async () { const data await (await this.env.BUCKET.get(dataRef.key)).json(); for (let i 0; i data.length; i 100) { await this.env.DB.batch(data.slice(i, i 100).map(item this.env.DB.prepare(INSERT INTO records VALUES (?, ?)).bind(item.id, item.normalized) )); } }); } }注意store步骤把大体积数据写入 R2仅返回引用{ key }正是大于 1 MiB 的数据存外部存储、只返回引用这一最佳实践的体现。10.4 人工审批Human-in-the-Loopexport class ApprovalWorkflow extends WorkflowEntrypointEnv, Params { async run(event, step) { await step.do(create approval, async () await this.env.DB.prepare(INSERT INTO approvals (id, user_id, status) VALUES (?, ?, ?)) .bind(event.instanceId, event.params.userId, pending).run()); try { const approval await step.waitForEvent{ approved: boolean }(wait for approval, { event: approval-response, timeout: 48h }); if (approval.approved) { await step.do(process approval, async () {}); } else { await step.do(handle rejection, async () {}); } } catch (e) { await step.do(auto reject, async () await this.env.DB.prepare(UPDATE approvals SET status ? WHERE id ?) .bind(auto-rejected, event.instanceId).run()); } } }10.5 编排模式Fan-Out / 父子 / 竞速 / 定时链// Fan-Out并行处理 const files await step.do(list, async () this.env.BUCKET.list()); await Promise.all(files.objects.map((file, i) step.do(process ${i}, async () processFile(await (await this.env.BUCKET.get(file.key)).arrayBuffer())))); // 父-子工作流 const child await step.do(start child, async () await this.env.CHILD_WORKFLOW.create({id: child-${event.instanceId}, params: { data: result.data }})); // 竞速模式 const winner await Promise.race([ step.do(option A, async () slowOperation()), step.do(option B, async () fastOperation()) ]); // 定时工作流链Cron 触发 7 天后二次跟进 export default { async scheduled(event, env) { await env.DAILY_WORKFLOW.create({id: daily-${event.scheduledTime}, params: { timestamp: event.scheduledTime }}); }}; export class DailyWorkflow extends WorkflowEntrypointEnv, Params { async run(event, step) { await step.do(daily task, async () {}); await step.sleep(wait 7 days, 7 days); await step.do(weekly followup, async () {}); } }十一、工作流测试vitest 与 Introspection API11.1 测试环境配置// vitest.config.ts import { defineWorkersConfig } from cloudflare/vitest-pool-workers/config; export default defineWorkersConfig({ test: { poolOptions: { workers: { wrangler: { configPath: ./wrangler.jsonc } } } } });11.2 Introspection API 控制步骤执行import { introspectWorkflowInstance } from cloudflare:test; const instance await env.MY_WORKFLOW.create({ params: { userId: 123 } }); const introspector await introspectWorkflowInstance(env.MY_WORKFLOW, instance.id); // 等待指定步骤完成 const result await introspector.waitForStepResult({ name: fetch user, index: 0 }); // 模拟步骤行为跳过真实副作用 await introspector.modify(async (m) { await m.mockStepResult({ name: api call }, { mocked: true }); });十二、最佳实践DO 与 DONT参考 patterns.md 的总结✅ 应该做粒度化步骤每个 API 调用一个步骤除非能证明幂等幂等性先检查再执行使用幂等键确定性步骤名使用静态名或基于步骤输出的名字通过返回值持久化状态而非依赖外部变量始终await step.do()避免悬空的 Promise确定性条件分支基于event.payload或步骤输出判断大数据存外部超过 1 MiB 的数据放入 R2/KV只返回引用批量创建需要多个实例时使用createBatch()。❌ 不要做一个巨型步骤破坏持久化与重试控制在步骤外保存状态休眠后会丢失修改 event事件不可变应返回新状态在步骤外使用非确定性逻辑Math.random()、Date.now()必须放进步骤内在步骤外做副作用重启时可能重复执行使用非确定性步骤名阻止结果缓存忽略超时waitForEvent会抛异常需 try-catch复用实例 ID保留期内必须唯一。十三、常见错误与排查gotchas.md参考 gotchas.md以下是高频问题速查错误现象原因解决方案Step Timeout步骤执行超过默认 10 分钟或自定义超时设置timeout: 30 minutes或在 wrangler.jsonc 提高 CPU 上限最大 5 分钟waitForEvent Timeout超时周期内未收到事件默认 24h最大 365d用 try-catch 包裹超时后走默认行为Non-Deterministic Step Names步骤名中使用Date.now()等动态值导致重放异常使用event.instanceId等确定性值命名State Lost in Variables用模块级/局部变量存状态休眠后丢失从step.do()返回值取状态自动持久化Non-Deterministic Conditionals在步骤外使用非确定性逻辑做判断把非确定性操作移入步骤内Large Step Returns Exceeding Limit步骤返回超过 1 MiB 数据存 R2只返回引用{ key: r2-object-key }CPU 限制触发但运行 30s混淆 CPU 时间与墙钟时间网络请求、DB 查询、睡眠不计入 CPU30s 指活跃计算时间Idempotency Violation操作非幂等导致重试重复扣款/动作执行前检查是否已完成如是否已扣款Instance ID Collision复用实例 ID 引发冲突使用带时间戳的唯一 ID${userId}-${Date.now()}完成后实例数据消失完成/报错实例在保留期后被自动删除免费 3 天 / 付费 30 天在完成前把关键数据导出到 KV/R2/D1漏掉await忘记await step.do()导致 fire-and-forget始终await step.do(task, ...)十四、限额与定价14.1 配额限制限制项FreePaid说明单步 CPU10ms30s默认5min最大通过 wrangler.jsonc 的limits.cpu_ms设置步骤状态1 MiB1 MiB单步返回值大小实例状态100 MB1 GB单个工作流实例总状态每工作流步骤数1,0241,024step.sleep()不计入每日执行数100k无限制日执行上限并发实例2510k最大并发waiting 状态不计入排队实例100k1M最大排队实例数单步子请求501,000每步最大出站请求数状态保留期3 天30 天已完成实例的保留时长Step 默认超时10 min10 min单次尝试waitForEvent 默认超时24h24h最大 365 天waitForEvent 最大超时365 天365 天最长等待时间关键提示处于waiting状态来自step.sleep或step.waitForEvent的实例不计入并发实例上限因此可以同时存在数百万个睡眠中的工作流。14.2 定价指标FreePaid说明请求100k/天10M/月 $0.30/M工作流调用次数CPU 时间10ms/次30M CPU-ms/月 $0.02/M CPU-ms实际 CPU 用量存储1 GB1 GB/月 $0.20/GB-月所有实例运行/报错/睡眠/完成十五、参考文档与阅读顺序阅读顺序入门路线configuration.md配置→ api.mdAPI→ patterns.md模式排查路线gotchas.md。本参考的全部文档文档覆盖内容configuration.mdwrangler.jsonc 配置、Step 配置、绑定api.mdStep API、实例管理、睡眠与参数patterns.md常见工作流、测试、编排gotchas.md超时、限额、调试策略相关方案对比durable-objects另一种有状态实现方案适合实时协调/强一致状态queues消息驱动的工作流workers工作流实例的入口环境。选择建议当任务本质是若干可重试步骤的编排、需要跨分钟到周的状态保持时Workflows 是最贴合的工具若需要细粒度单实体状态与实时同步可考虑 Durable Objects若以消息解耦为主则 Queues 更合适。这些参考文档均位于 cloudflare-deploy 技能目录 下供深入查阅。【免费下载链接】skillsSkills Catalog for Codex项目地址: https://gitcode.com/GitHub_Trending/skills4/skills创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考