ch19-queues

對應 19. Queues：把工作推離熱路徑

在 GitHub 上檢視·4 個檔案·5.4 KB

取得並執行

這個範例可以獨立 clone 執行，不依賴其他章節。

npx degit mrsuner/cloudflare-serverless-platform-manuals/examples/ch19-queues
cd ch19-queues
npm install

可用指令

npm run dev	# wrangler dev
npm run typecheck	# tsc --noEmit
npm run cf-typegen	# wrangler types --env-interface CloudflareBindings

說明

ch19 — Cloudflare Queues

Companion example for docs/19-queues.md.

One Worker acting as producer and as consumer for two queues (main + DLQ).

npm install && npm run dev

B=localhost:8787
curl -s "$B/metrics"
curl -s "$B/contenttypes"
curl -s "$B/delay"
curl -s "$B/clear" && curl -s "$B/sendbatch?n=12" && sleep 6 && curl -s "$B/seen"
curl -s "$B/clear" && curl -s "$B/send?n=2&fail=1" && sleep 15 && curl -s "$B/seen"

Verified findings

2026-08-01 · wrangler@4.114.0 · workerd@1.20260722.1

Producer surface, including the new metrics()

{"metrics":{"backlogCount":0,"backlogBytes":0},
 "producerProto":["metrics","send","sendBatch","constructor"]}

metrics() shipped 2026-04-28. Docs also list oldestMessageTimestamp; it did not appear locally against an empty queue — confirm in production.

Content types

{"json":"accepted","text":"accepted","bytes":"accepted","v8":"accepted",
 "invalid":"THREW: TypeError: Unsupported queue message content type: nope"}

Default is json (changed from v8, because pull consumers cannot decode v8).

delaySeconds bounds

{"delaySeconds=0":"accepted","delaySeconds=1":"accepted","delaySeconds=86400":"accepted",
 "delaySeconds=86401":"THREW: Error: Unknown Internal Error (15000)",
 "delaySeconds=-1":"THREW: Error: Unknown Internal Error (15000)"}

0–86400 inclusive. Out-of-range gives a useless error — validate before sending.

Batching honours max_batch_size

12 messages with max_batch_size: 5:

batches: 3
  ch19-clicks size=5 attempts=[1,1,1,1,1]
  ch19-clicks size=5 attempts=[1,1,1,1,1]
  ch19-clicks size=2 attempts=[1,1]

messageProto: ['retry','ack','constructor']
batchProto:   ['retryAll','ackAll','constructor']
timestampIsDate: True

attempts starts at 1, not 0.

Retry then DLQ, with attempts reset

max_retries: 2, retry_delay: 1, DLQ configured, two failing messages:

t+4s   ch19-clicks     n=2 attempts=[1,1]
t+8s   ch19-clicks     n=2 attempts=[2,2]
t+12s  ch19-clicks     n=2 attempts=[3,3]
       ch19-clicks-dlq n=2 attempts=[1,1]   <- reset
t+24s  (no further change)

max_retries: 2 means three total attempts. And the DLQ consumer sees attempts: 1, so the original failure count must be carried in the message body if you need it.

Config schema beats the docs

The configure-queues docs page omits delivery_delay and retry_delay, and its example uses non-default values. Some sources also claim visibility_timeout is CLI-only — but wrangler’s own schema lists visibility_timeout_ms as a consumer field:

queue  type  max_batch_size  max_batch_timeout  max_retries
dead_letter_queue  max_concurrency  visibility_timeout_ms  retry_delay

Read node_modules/wrangler/config-schema.json when in doubt.

原始碼

• wrangler.jsonc
• package.json
• src/index.ts
• tsconfig.json

wrangler.jsonc

{
  "$schema": "./node_modules/wrangler/config-schema.json",
  "name": "ch19-queues",
  "main": "src/index.ts",
  "compatibility_date": "2026-07-24",
  "observability": { "enabled": true },

  "queues": {
    "producers": [
      { "binding": "CLICKS", "queue": "ch19-clicks" },
      { "binding": "DLQ_PRODUCER", "queue": "ch19-clicks-dlq" }
    ],
    "consumers": [
      {
        "queue": "ch19-clicks",
        "max_batch_size": 5,        // default 10
        "max_batch_timeout": 2,     // default 5
        "max_retries": 2,           // default 3
        "dead_letter_queue": "ch19-clicks-dlq",
        "retry_delay": 1
      },
      { "queue": "ch19-clicks-dlq", "max_batch_size": 10 }
    ]
  }
}

package.json

{ "name": "ch19-queues", "private": true, "type": "module",
  "scripts": { "dev": "wrangler dev", "typecheck": "tsc --noEmit",
    "cf-typegen": "wrangler types --env-interface CloudflareBindings" },
  "devDependencies": { "typescript": "^5.9.0", "wrangler": "^4.114.0" } }

src/index.ts

// ---------------------------------------------------------------------------
// Chapter 19 — Queues: batching, retries, DLQ.
// ---------------------------------------------------------------------------
type ClickMsg = { id: string; slug: string; fail?: boolean; ts: number };

/** Observation log, kept in module scope for the demo only. */
const seen: Array<Record<string, unknown>> = [];

export default {
  async fetch(request: Request, env: CloudflareBindings): Promise<Response> {
    const url = new URL(request.url);

    switch (url.pathname) {
      case "/send": {
        const n = Number(url.searchParams.get("n") ?? 1);
        const fail = url.searchParams.get("fail") === "1";
        for (let i = 0; i < n; i++) {
          await env.CLICKS.send({ id: `m${Date.now()}-${i}`, slug: "cf", fail, ts: Date.now() });
        }
        return Response.json({ sent: n, fail });
      }

      // sendBatch: up to 100 messages / 256 KB per call.
      case "/sendbatch": {
        const n = Number(url.searchParams.get("n") ?? 10);
        await env.CLICKS.sendBatch(
          Array.from({ length: n }, (_, i) => ({
            body: { id: `b${Date.now()}-${i}`, slug: "batch", ts: Date.now() } satisfies ClickMsg,
          })),
        );
        return Response.json({ sentBatch: n });
      }

      // Per-message contentType. Default is "json" (changed from "v8").
      case "/contenttypes": {
        const out: Record<string, unknown> = {};
        for (const ct of ["json", "text", "bytes", "v8"] as const) {
          try {
            const body = ct === "bytes" ? new Uint8Array([1, 2, 3]) : ct === "text" ? "plain" : { ct };
            await env.CLICKS.send(body as never, { contentType: ct });
            out[ct] = "accepted";
          } catch (e) { out[ct] = `THREW: ${String(e).slice(0, 120)}`; }
        }
        try {
          await env.CLICKS.send({ x: 1 }, { contentType: "nope" as never });
          out.invalid = "accepted";
        } catch (e) { out.invalid = `THREW: ${String(e).slice(0, 120)}`; }
        return Response.json(out);
      }

      // delaySeconds: 0..86400
      case "/delay": {
        const out: Record<string, unknown> = {};
        for (const d of [0, 1, 86400, 86401, -1]) {
          try {
            await env.CLICKS.send({ id: `d${d}`, slug: "delay", ts: Date.now() }, { delaySeconds: d });
            out[`delaySeconds=${d}`] = "accepted";
          } catch (e) { out[`delaySeconds=${d}`] = `THREW: ${String(e).slice(0, 140)}`; }
        }
        return Response.json(out);
      }

      // metrics() — added April 2026.
      case "/metrics": {
        const out: Record<string, unknown> = {};
        try { out.metrics = await env.CLICKS.metrics(); }
        catch (e) { out.metrics = `THREW: ${String(e).slice(0, 160)}`; }
        out.producerProto = Object.getOwnPropertyNames(Object.getPrototypeOf(env.CLICKS));
        return Response.json(out);
      }

      case "/seen":  return Response.json({ count: seen.length, seen });
      case "/clear": seen.length = 0; return Response.json({ ok: true });

      default:
        return new Response(
          "try /send?n=3 /send?n=1&fail=1 /sendbatch?n=10 /contenttypes /delay /metrics /seen /clear\n",
          { status: 404 },
        );
    }
  },

  async queue(batch: MessageBatch<ClickMsg>, env: CloudflareBindings): Promise<void> {
    const isDlq = batch.queue.endsWith("-dlq");

    seen.push({
      queue: batch.queue,
      size: batch.messages.length,
      // attempts starts at 1, not 0.
      attempts: batch.messages.map((m) => m.attempts),
      ids: batch.messages.map((m) => m.body?.id ?? "?"),
      timestampIsDate: batch.messages[0]?.timestamp instanceof Date,
      messageProto: Object.getOwnPropertyNames(Object.getPrototypeOf(batch.messages[0] ?? {})),
      batchProto: Object.getOwnPropertyNames(Object.getPrototypeOf(batch)),
    });

    if (isDlq) { batch.ackAll(); return; }

    for (const m of batch.messages) {
      if (m.body?.fail) {
        // Explicit retry. After max_retries it goes to the DLQ.
        m.retry();
      } else {
        m.ack();
      }
    }
  },
} satisfies ExportedHandler<CloudflareBindings, ClickMsg>;

tsconfig.json

{ "compilerOptions": { "target": "esnext", "lib": ["esnext"], "module": "esnext",
  "moduleResolution": "bundler", "types": ["./worker-configuration.d.ts"],
  "strict": true, "skipLibCheck": true, "noEmit": true, "isolatedModules": true },
  "include": ["src/**/*.ts", "worker-configuration.d.ts"] }