在 Durable Objects 中使用 Queues

以下示例展示如何编写 Worker 脚本，从 Durable Object 内向 Cloudflare Queues 发布消息。

前提条件：

通过 Cloudflare 仪表板或 wrangler CLI 创建的队列。
在 Cloudflare 仪表板或 Wrangler 文件中配置的生产者绑定。
Durable Object 命名空间绑定。

按如下方式配置 Wrangler 文件：

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "my-worker",
	"queues": {
		"producers": [
			{
				"queue": "my-queue",
				"binding": "YOUR_QUEUE"
			}
		]
	},
	"durable_objects": {
		"bindings": [
			{
				"name": "YOUR_DO_CLASS",
				"class_name": "YourDurableObject"
			}
		]
	},
	"migrations": [
		{
			"tag": "v1",
			"new_sqlite_classes": [
				"YourDurableObject"
			]
		}
	]
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "my-worker"

[[queues.producers]]
queue = "my-queue"
binding = "YOUR_QUEUE"

[[durable_objects.bindings]]
name = "YOUR_DO_CLASS"
class_name = "YourDurableObject"

[[migrations]]
tag = "v1"
new_sqlite_classes = [ "YourDurableObject" ]

以下 Worker 脚本：

创建 Durable Object stub，或根据 userId 检索现有 stub。
将请求数据传递给 Durable Object。
在 Durable Object 内向队列发布消息。

扩展 DurableObject 基类使 Env 在 Durable Object 的 fetch() 处理程序中可通过 this.env 访问，Durable Object 状态可通过 this.ctx 访问。

import { DurableObject } from "cloudflare:workers";

interface Env {
  YOUR_QUEUE: Queue;
  YOUR_DO_CLASS: DurableObjectNamespace<YourDurableObject>;
}

export default {
  async fetch(req, env, ctx): Promise<Response> {
    // Assume each Durable Object is mapped to a userId in a query parameter
    // In a production application, this will be a userId defined by your application
    // that you validate (and/or authenticate) first.
    const url = new URL(req.url);
    const userIdParam = url.searchParams.get("userId");

    if (userIdParam) {
      // Get a stub that allows you to call that Durable Object
      const durableObjectStub = env.YOUR_DO_CLASS.getByName(userIdParam);

      // Pass the request to that Durable Object and await the response
      // This invokes the constructor once on your Durable Object class (defined further down)
      // on the first initialization, and the fetch method on each request.
      // We pass the original Request to the Durable Object's fetch method
      const response = await durableObjectStub.fetch(req);

      // This would return "wrote to queue", but you could return any response.
      return response;
    }
    return new Response("userId must be provided", { status: 400 });
  },
} satisfies ExportedHandler<Env>;

export class YourDurableObject extends DurableObject<Env> {
  async fetch(req: Request): Promise<Response> {
    // Error handling elided for brevity.
    // Publish to your queue
    await this.env.YOUR_QUEUE.send({
      id: this.ctx.id.toString(), // Write the ID of the Durable Object to your queue
      // Write any other properties to your queue
    });

    return new Response("wrote to queue");
  }
}