用 Queues 打造事件驅動應用

有些工作很慢：寄一封 email、產生縮圖、把資料同步到別的系統。如果你在使用者等待時才做，頁面會卡卡的，而且一個小閃失就把工作弄丟了。事件驅動（event-driven）設計同時解決這兩個問題：請求只先記下『這件事要做』就立刻返回；真正的工作之後在背景可靠地進行。

在 Cloudflare 上，把它們黏起來的膠水就是 Queues——一條可靠的訊息排隊線。生產者（producer）Worker 接下請求，把一則訊息丟進佇列；消費者（consumer）Worker 之後被叫來，帶著一批訊息做粗重工作（寫進 D1、把檔案存到 R2、呼叫外部 API）。這兩個 Worker 從不直接互相呼叫，佇列就坐在中間。這種分離就叫做「解耦（decoupling）」。

注意從生產者出發有兩個箭頭：一個進入佇列（送出訊息）；另一個直接回到呼叫端，回傳 202 Accepted——而且是在任何慢工作開始之前。這個分岔，就是事件驅動應用的核心。

事件驅動流程裡有五個你會一直碰到的概念：生產者、佇列、消費者，再加上可靠度三寶——批次、重試、死信佇列。下面用一句話講清楚每一個。

想像一個註冊表單。使用者送出後，我們欠他一封歡迎信。訣竅是：工作一進佇列就回 202 Accepted，之後再寄信。下面是完整的來回——注意使用者在信還沒寄出之前，就已經被放走了。

每一步在做什麼

• POST /signup：瀏覽器把表單送到生產者 Worker。
• 把寄信工作丟進佇列：生產者呼叫 send()，丟出一則描述這份工作的訊息。
• 202 Accepted：生產者馬上回應——『202』的意思是『我收下了，之後會處理』。
• 稍後成批投遞：訊息備妥時，Cloudflare 帶著一批訊息來呼叫消費者。
• 寄出 email：消費者向信件服務做真正的慢工作。
• ack 確認訊息：成功後消費者送出確認，訊息就永遠離開佇列了。

背景工作有時會失敗——信件伺服器掛了、API 逾時。Queues 用一條簡單規則處理：訊息會一直活著，直到被 ack 為止。如果處理時丟出錯誤（或你呼叫 retry()），訊息之後會被重新投遞。失敗達到 max_retries 次後，它會被移到死信佇列（DLQ），而不是被丟掉。

冪等性：扛得住重複

Queues 是「至少一次（at-least-once）」投遞：同一則訊息偶爾可能被投遞超過一次（例如消費者處理成功了，卻在 ack 之前當機）。所以你的處理函式必須是「冪等的（idempotent）」——用同一則訊息跑兩次，結果要跟跑一次一樣。經典做法：給每則訊息一個唯一 id，把做完的 id 記下來，遇到已經看過的 id 就略過。

下面是一個完整、可跑的範例：一個把工作丟進佇列並回 202 的生產者 Worker、一個成批冪等處理並 ack 的消費者 Worker、把佇列接好（含重試與 DLQ）的 wrangler.toml、追蹤已完成 id 的 D1 資料表，以及把所有東西建好並部署的指令。

1. 1. 生產者 Worker — 接下請求、回 202

   每次請求時，它建一則帶唯一 id 的訊息，透過 MY_QUEUE 綁定送進佇列，然後立刻回 202。這裡不做任何慢工作。

   jscontent_copy複製

   // src/producer.js
   export default {
     // PRODUCER: runs on every HTTP request
     async fetch(request, env, ctx) {
       const { email } = await request.json();
   
       // Build a job with a unique id (used later for idempotency)
       const msg = {
         id: crypto.randomUUID(),
         type: "welcome_email",
         email,
         ts: Date.now(),
       };
   
       // Hand the slow work to the queue, then reply immediately
       await env.MY_QUEUE.send(msg);
   
       return new Response(JSON.stringify({ queued: true }), {
         status: 202,
         headers: { "content-type": "application/json" },
       });
     },
   };

2. 2. 消費者 Worker — 成批處理並 ack

   Cloudflare 帶著一批訊息呼叫 queue()。我們對 batch.messages 跑迴圈，略過已完成的 id（冪等），做慢工作，成功就 ack、失敗就 retry。失敗的訊息會被重新投遞，達到 max_retries 後自動進 DLQ。

   jscontent_copy複製

   // src/consumer.js
   export default {
     // CONSUMER: Cloudflare calls this with a batch of messages
     async queue(batch, env, ctx) {
       for (const m of batch.messages) {
         try {
           const job = m.body;
   
           // Idempotency: skip if this id was already handled
           const done = await env.DB
             .prepare("SELECT 1 FROM processed WHERE id = ?")
             .bind(job.id)
             .first();
           if (done) {
             m.ack();
             continue;
           }
   
           // Do the slow work: send the email
           await sendEmail(env, job.email);
   
           // Record the id so a retry never sends twice
           await env.DB
             .prepare("INSERT INTO processed (id, ts) VALUES (?, ?)")
             .bind(job.id, Date.now())
             .run();
   
           m.ack(); // success -> remove from the queue
         } catch (err) {
           m.retry(); // failure -> redeliver later, then DLQ
         }
       }
     },
   };

3. 3. wrangler.toml — 接好生產者、消費者與 DLQ

   queues.producers 把佇列以 env.MY_QUEUE 的名義開放給你的程式碼。queues.consumers 告訴 Cloudflare 用一批批訊息來呼叫 queue()，並設定批次大小、重試次數，以及用盡重試的訊息要落到的 dead_letter_queue。

   tomlcontent_copy複製

   name = "queue-app"
   main = "src/index.js"
   compatibility_date = "2025-01-01"
   
   # PRODUCER: send to the "jobs" queue as env.MY_QUEUE
   [[queues.producers]]
   queue = "jobs"
   binding = "MY_QUEUE"
   
   # CONSUMER: Cloudflare invokes queue() with batches from "jobs"
   [[queues.consumers]]
   queue = "jobs"
   max_batch_size = 10
   max_batch_timeout = 5
   max_retries = 3
   dead_letter_queue = "jobs-dlq"
   
   # D1 used for idempotency bookkeeping (env.DB)
   [[d1_databases]]
   binding = "DB"
   database_name = "queue-app-db"
   database_id = "<paste-your-database-id>"

4. 4. D1 資料表 — 記住已完成的 id

   一張以訊息 id 為主鍵的小表。消費者做完後插入一列，並在動手前先查這張表，讓它在重試與重複投遞之間保持冪等。

   sqlcontent_copy複製

   -- schema.sql
   CREATE TABLE IF NOT EXISTS processed (
     id  TEXT PRIMARY KEY,
     ts  INTEGER NOT NULL
   );

5. 5. 建立佇列 + DLQ，然後部署

   建立主佇列和死信佇列（它就是另一個普通佇列），建立 D1 資料庫、載入 schema、部署，再用 tail 即時看到訊息被消費的過程。

   bashcontent_copy複製

   npx wrangler queues create jobs
   npx wrangler queues create jobs-dlq
   npx wrangler d1 create queue-app-db
   npx wrangler d1 execute queue-app-db --remote --file=./schema.sql
   npx wrangler deploy
   npx wrangler tail

初學者常踩的雷

• 忘了 ack：沒被 ack 的訊息會被當成失敗、重新投遞——成功時記得呼叫 m.ack()。
• 沒有冪等：至少一次投遞代表會出現重複；每個副作用都要用唯一 id 檢查擋一下。
• 把慢工作塞進生產者：讓 fetch() 保持快——只 send() 然後回 202，粗重工作交給消費者。
• 批次太小：調整 max_batch_size 與 max_batch_timeout，讓消費者高效處理，而不是一則一則做。
• 忽略 DLQ：訊息可能默默躺在那裡——加個告警或定期檢查，才會發現失敗。