跳转到内容
搜索文档
↵选择
↑↓导航
ESC关闭

触发 Workflows

最后更新 查看 MarkdownAgent 设置

你可以通过编程方式和 Workflows API 触发 Workflows，包括：

  1. 在 Workers 中通过 fetch 处理程序中的 HTTP 请求，或从 queue 或 scheduled 处理程序中的绑定（binding）
  2. 在 Wrangler 配置的 Workflow 绑定上定义 schedules，按 recurring 间隔运行
  3. 使用 Workflows REST API
  4. 在终端中通过 wrangler CLI

Workers API（绑定）

你可以通过创建 Workflow 绑定，从任何 Worker 脚本以编程方式与 Workflows 交互。一个 Worker 可以绑定多个 Workflows，包括账户内其他 Workers 项目（脚本）中定义的 Workflows。

你可以通过以下方式触发 Workflow：

  • 通过 fetch 处理程序直接通过 HTTP
  • 在 queue 处理程序中从 Queue consumer
  • 在 wrangler.jsonc 的 Workflow 绑定上定义 schedules，按 recurring 计划运行
  • 在 scheduled 处理程序中从 Cron Trigger
  • 在 Durable Objects 内

要从 Workers 代码绑定到 Workflow，你需要定义到特定 Workflow 的绑定（binding）。例如，要绑定到快速入门指南 中定义的 Workflow，你需要在 Wrangler 配置文件 中配置以下内容：

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "workflows-tutorial",
	"main": "src/index.ts",
	// Set this to today's date
	"compatibility_date": "2026-08-17",
	"workflows": [
		{
			// Workflow 名称
			"name": "workflows-tutorial",
			// 绑定名称，必须是有效的 JavaScript 变量名。这将是你从其他 Workers 处理程序或
			// 脚本调用（运行）Workflow 的方式。
			"binding": "MY_WORKFLOW",
			// 必须与代码中扩展 Workflow 类的 class 匹配
			"class_name": "MyWorkflow"
		}
	]
}
"$schema" = "./node_modules/wrangler/config-schema.json"
name = "workflows-tutorial"
main = "src/index.ts"
# Set this to today's date
compatibility_date = "2026-08-17"

[[workflows]]
name = "workflows-tutorial"
binding = "MY_WORKFLOW"
class_name = "MyWorkflow"

binding = "MY_WORKFLOW" 行定义了 Workflow 方法可访问的 JavaScript 变量，包括 create（触发新实例）或 get（返回现有实例的状态）。

直接调度 Workflow

如果希望按 recurring 间隔创建 Workflow 实例，在 Wrangler 配置的 Workflow 绑定中添加 schedules 数组（每个账户最多 100 个 cron 表达式）：

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "workflows-tutorial",
	"main": "src/index.ts",
	// Set this to today's date
	"compatibility_date": "2026-08-17",
	"workflows": [
		{
			"name": "workflows-tutorial",
			"binding": "MY_WORKFLOW",
			"class_name": "MyWorkflow",
			"schedules": ["0 * * * *"]
		}
	]
}
"$schema" = "./node_modules/wrangler/config-schema.json"
name = "workflows-tutorial"
main = "src/index.ts"
# Set this to today's date
compatibility_date = "2026-08-17"

[[workflows]]
name = "workflows-tutorial"
binding = "MY_WORKFLOW"
class_name = "MyWorkflow"
schedules = [ "0 * * * *" ]

每个匹配的 cron 表达式会自动创建新的 Workflow 实例。当你希望按计划运行 Workflow 而无需定义顶层 triggers.crons 和单独的 scheduled 处理程序时使用此方式。

调度的实例在 event.schedule 上包含匹配的 cron 表达式和调度触发时间：

export class MyWorkflow extends WorkflowEntrypoint<Env> {
	async run(event: WorkflowEvent<unknown>, step: WorkflowStep) {
		if (event.schedule) {
			console.log(event.schedule.cron);
			console.log(new Date(event.schedule.scheduledTime));
		}
	}
}

在 Workers Paid 上，由 schedules 创建的 Workflow 实例每次 cron 触发可运行最多一小时，而不消耗 Workflow 并发槽位。如果实例在该窗口后暂停或休眠，实例让出并在恢复时进入正常并发队列。当并发槽位可用时恢复。

配置 Workflow 调度时请使用最新 Wrangler 版本。如果本地 Wrangler schema 尚不识别 schedules，请在部署前更新 Wrangler。

以下示例展示如何在 Worker 内管理 Workflows，包括：

  • 通过 ID 检索现有 Workflow 实例的状态
  • 创建（触发）新的 Workflow 实例
  • 返回给定实例 ID 的状态
src/index.tsts
interface Env {
	MY_WORKFLOW: Workflow;
}

export default {
	async fetch(req: Request, env: Env) {
		// 从查询参数获取 instanceId
		const instanceId = new URL(req.url).searchParams.get("instanceId");

		// 如果提供了 ?instanceId=<id> 查询参数，则获取
		// 现有 Workflow 的状态。
		if (instanceId) {
			let instance = await env.MY_WORKFLOW.get(instanceId);
			return Response.json({
				status: await instance.status(),
			});
		}

		// 否则，创建 Workflow 的新实例，传递任何（可选）
		// 参数并返回 ID。
		const newId = crypto.randomUUID();
		let instance = await env.MY_WORKFLOW.create({ id: newId });
		return Response.json({
			id: instance.id,
			details: await instance.status(),
		});
	},
};

检查 Workflow 状态

你可以通过针对特定实例 ID 调用 status 来检查任何运行中 Workflow 实例的状态。这允许你以编程方式检查实例是排队（等待调度）、正在运行、已暂停还是出错。

let instance = await env.MY_WORKFLOW.get("abc-123");
let status = await instance.status(); // 返回 InstanceStatus

status 的可能值如下：

  status:
    | "queued" // 实例等待启动（参见并发限制）
    | "running"
    | "paused"
    | "errored"
    | "terminated" // 用户在运行中终止了实例
    | "complete"
    | "waiting" // 实例处于休眠状态，等待 sleep 或 event 完成
    | "waitingForPause" // 实例正在完成当前工作以暂停
    | "unknown";
  error?: {
    name: string,
    message: string
  };
	output?: unknown;
	rollback:
		| {
				outcome: "complete" | "failed";
				error: {
					name: string,
					message: string,
				} | null,
		  }
		| null;

如果 Workflow 在 step.do() 上注册了回滚处理程序，在实例完成后检查 rollback 以查看补偿步骤是否成功完成。回滚 actively 运行时，Workers API 继续返回 status: "running"。

显式暂停 Workflow

你可以通过针对特定实例 ID 调用 pause 显式暂停 Workflow 实例（稍后恢复）。

let instance = await env.MY_WORKFLOW.get("abc-123");
await instance.pause(); // 返回 Promise<void>

恢复 Workflow

你可以通过针对特定实例 ID 调用 resume 恢复已暂停的 Workflow 实例。

let instance = await env.MY_WORKFLOW.get("abc-123");
await instance.resume(); // 返回 Promise<void>

对当前未暂停的实例调用 resume 不会有任何效果。

停止 Workflow

你可以通过针对特定实例 ID 调用 terminate 停止/终止 Workflow 实例。

let instance = await env.MY_WORKFLOW.get("abc-123");
await instance.terminate(); // 返回 Promise<void>

要在终止前运行已注册的回滚处理程序，传递 rollback: true：

let instance = await env.MY_WORKFLOW.get("abc-123");
await instance.terminate({ rollback: true }); // 返回 Promise<void>

你也可以从 Wrangler 运行回滚处理程序：

npx wrangler workflows instances terminate <WORKFLOW_NAME> <INSTANCE_ID> --rollback
# 对于 wrangler dev 期间的本地 Workflows 实例：
npx wrangler workflows instances terminate <WORKFLOW_NAME> <INSTANCE_ID> --local --rollback

停止/终止后，Workflow 实例无法恢复。

重启 Workflow

let instance = await env.MY_WORKFLOW.get("abc-123");
await instance.restart(); // 返回 Promise<void>

重启实例将立即取消任何进行中的步骤，清除任何中间状态，并将 Workflow 视为首次运行。

要从特定步骤而非开头重启实例，请参阅 Workers API 参考中的 restart。

从另一个 Workflow 触发 Workflow

你可以在一个 Workflow 的步骤内创建另一个 Workflow 的新实例。父 Workflow 不会阻塞等待子 Workflow 完成——在子实例成功创建后立即继续执行。

export class ParentWorkflow extends WorkflowEntrypoint {
	async run(event, step) {
		// 执行初始工作
		const result = await step.do("initial processing", async () => {
			// ... 处理逻辑
			return { fileKey: "output.pdf" };
		});

		// 触发子 Workflow 进行额外处理
		const childInstance = await step.do("trigger child workflow", async () => {
			return await this.env.CHILD_WORKFLOW.create({
				id: `child-${event.instanceId}`,
				params: { fileKey: result.fileKey },
			});
		});

		// 父 Workflow 立即继续 - 不被子 Workflow 阻塞
		await step.do("continue with other work", async () => {
			console.log(`Started child workflow: ${childInstance.id}`);
			// 无论子 Workflow 状态如何，这都会立即运行
		});
	}
}
export class ParentWorkflow extends WorkflowEntrypoint<Env, Params> {
	async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
		// 执行初始工作
		const result = await step.do("initial processing", async () => {
			// ... 处理逻辑
			return { fileKey: "output.pdf" };
		});

		// 触发子 Workflow 进行额外处理
		const childInstance = await step.do("trigger child workflow", async () => {
			return await this.env.CHILD_WORKFLOW.create({
				id: `child-${event.instanceId}`,
				params: { fileKey: result.fileKey },
			});
		});

		// 父 Workflow 立即继续 - 不被子 Workflow 阻塞
		await step.do("continue with other work", async () => {
			console.log(`Started child workflow: ${childInstance.id}`);
			// 无论子 Workflow 状态如何，这都会立即运行
		});
	}
}

如果子 Workflow 启动失败，步骤将失败并根据你的重试配置重试。子实例成功创建后，它独立于父 Workflow 运行。

REST API (HTTP)

请参阅 Workflows REST API 文档。

命令行 (CLI)

请参阅 CLI 快速入门 了解如何通过命令行管理和触发 Workflows。

这篇文档对您有帮助吗？