Queues 批次、重試與 DLQ:生產級容錯調校 | Cloudflare 完整教學

// wrangler.jsonc
{
  "name": "payment-notify-worker",
  "main": "src/index.ts",
  "compatibility_date": "2025-01-01",
  "observability": { "enabled": true },  // ★ 生產環境務必開,才看得到 backlog 與延遲
  "queues": {
    "producers": [
      { "queue": "payment-notify", "binding": "NOTIFY_QUEUE" },
      // DLQ 也綁成 producer:讓程式能主動把「永久性失敗」訊息送進 DLQ
      { "queue": "payment-notify-dlq", "binding": "NOTIFY_DLQ" }
    ],
    "consumers": [
      {
        "queue": "payment-notify",
        "max_batch_size": 20,          // ★ 一批最多 20 則(1–100)
        "max_batch_timeout": 10,       // ★ 湊不滿就等最多 10 秒(0–60)
        "max_concurrency": 5,          // ★ 最多 5 個實例併發,保護下游支付 API
        "max_retries": 5,              // ★ 重試 5 次仍失敗 → 轉入 DLQ
        "retry_delay": 30,             // 佇列層級預設延遲(會被程式碼的 delaySeconds 覆蓋)
        "dead_letter_queue": "payment-notify-dlq"  // ★ 失敗訊息的去處
      },
      {
        // DLQ 自己也需要一個 consumer 來處理進來的死信
        "queue": "payment-notify-dlq",
        "max_batch_size": 50,
        "max_batch_timeout": 30,
        "max_retries": 1               // DLQ 通常不需要多次重試
      }
    ]
  },
  "d1_databases": [
    { "binding": "DB", "database_name": "payments", "database_id": "your-db-id" }
  ]
}

# 建立主佇列與 DLQ
wrangler queues create payment-notify
wrangler queues create payment-notify-dlq

# 也可以在 CLI 直接把 consumer 加上 DLQ 與重試設定
wrangler queues consumer add payment-notify payment-notify-worker \
  --batch-size=20 --max-retries=5 --dead-letter-queue=payment-notify-dlq

wrangler queues list

// src/index.ts

// 一則「支付通知」訊息
interface PaymentNotice {
  noticeId: string;        // ★ 唯一 id,冪等去重與 DLQ 追蹤都靠它
  orderId: string;
  userEmail: string;
  amount: number;
}

interface Env {
  NOTIFY_QUEUE: Queue<PaymentNotice>;
  NOTIFY_DLQ: Queue<PaymentNotice>;   // 讓程式能主動送進 DLQ
  DB: D1Database;
}

export default {
  async queue(
    batch: MessageBatch<PaymentNotice>,
    env: Env,
    ctx: ExecutionContext
  ): Promise<void> {
    console.log(`收到批次:${batch.messages.length} 則,佇列 ${batch.queue}`);

    for (const msg of batch.messages) {
      try {
        // ★ 冪等檢查:至少一次交付代表可能重送,先確認沒處理過
        const done = await env.DB
          .prepare("SELECT 1 FROM sent_notices WHERE notice_id = ?")
          .bind(msg.body.noticeId)
          .first();
        if (done) {
          msg.ack();          // 已處理過,直接確認,避免重複寄送
          continue;
        }

        // 真正的工作:呼叫外部支付通知 API
        const res = await fetch("https://pay.example.com/notify", {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify(msg.body),
        });

        // ---- 錯誤分類:決定「重試」還是「進 DLQ」----
        if (res.status === 429 || res.status >= 500) {
          // 暫時性錯誤(限流 / 上游 5xx)→ 指數退避重試
          throw new TransientError(`上游暫時失敗:${res.status}`);
        }
        if (res.status === 400 || res.status === 401 || res.status === 403) {
          // 永久性錯誤(格式壞 / 認證失敗)→ 毒訊息,重試也沒用
          throw new PermanentError(`永久性失敗:${res.status}`);
        }
        if (!res.ok) {
          throw new TransientError(`未預期狀態:${res.status}`);
        }

        // 成功:寫入去重表 + ack
        await env.DB
          .prepare("INSERT INTO sent_notices (notice_id, order_id, sent_at) VALUES (?, ?, ?)")
          .bind(msg.body.noticeId, msg.body.orderId, new Date().toISOString())
          .run();
        msg.ack();  // ★ 明確確認,訊息移出佇列

      } catch (err) {
        if (err instanceof PermanentError) {
          // ★ 毒訊息:主動送進 DLQ 存證,然後 ack 讓它離開主佇列
          console.error(`毒訊息 ${msg.body.noticeId}:${err.message}`);
          await env.NOTIFY_DLQ.send(msg.body);
          msg.ack();  // 不再重試,防止無窮迴圈拖垮佇列
          continue;
        }

        // ★ 暫時性錯誤:指數退避 —— 30s → 60s → 120s → ... 上限 12 小時
        const delaySeconds = Math.min(30 * 2 ** (msg.attempts - 1), 43200);
        console.warn(`第 ${msg.attempts} 次失敗,${delaySeconds}s 後重試:${msg.body.noticeId}`);
        // 重試 5 次(max_retries)仍失敗,Cloudflare 會自動把它轉進 DLQ
        msg.retry({ delaySeconds });
      }
    }
  },

  // DLQ consumer 見下一段 ↓
} satisfies ExportedHandler<Env>;

// 自訂錯誤型別:用來區分暫時性 vs 永久性
class TransientError extends Error {}
class PermanentError extends Error {}

export default {
  // ...(承上,主 queue handler 略)

  // ★ 用 batch.queue 分流:同一個 handler 同時服務主佇列與 DLQ
  async queue(batch: MessageBatch<PaymentNotice>, env: Env): Promise<void> {
    if (batch.queue === "payment-notify-dlq") {
      return handleDlq(batch, env);
    }
    // ... 主佇列邏輯(如上一段)
  },
} satisfies ExportedHandler<Env>;

// 專責處理死信
async function handleDlq(batch: MessageBatch<PaymentNotice>, env: Env): Promise<void> {
  for (const msg of batch.messages) {
    // 1. 落地存證:寫入 failed_notices,保留完整 body 與嘗試次數,供事後排查
    await env.DB
      .prepare(
        "INSERT INTO failed_notices (notice_id, body, attempts, failed_at) VALUES (?, ?, ?, ?)"
      )
      .bind(
        msg.body.noticeId,
        JSON.stringify(msg.body),
        msg.attempts,
        new Date().toISOString()
      )
      .run();

    // 2. 發告警(示意:實務可接 Email Worker、Slack Webhook、PagerDuty)
    console.error(`[ALERT] 死信 ${msg.body.noticeId} 進入 DLQ,已重試 ${msg.attempts} 次`);

    // 3. ★ ack:確認死信已妥善存證,從 DLQ 移除
    //    修好根因後,可從 failed_notices 撈出 body,重新 send() 回主佇列做「重放」
    msg.ack();
  }
}

// wrangler.jsonc —— HTTP pull consumer 設定
{
  "queues": {
    "consumers": [
      {
        "queue": "payment-notify",
        "type": "http_pull",              // ★ 改為 HTTP 拉取模式
        "visibility_timeout_ms": 30000,   // 訊息被拉走後多久內未 ack 就重新可見
        "max_retries": 5,
        "dead_letter_queue": "payment-notify-dlq"
      }
    ]
  }
}

const BASE = `https://api.cloudflare.com/client/v4/accounts/${ACCOUNT_ID}/queues/${QUEUE_ID}`;

// 拉取一批(最多 batch_size 則,visibility_timeout 期間對其他消費者不可見)
const pull = await fetch(`${BASE}/messages/pull`, {
  method: "POST",
  headers: { Authorization: `Bearer ${API_TOKEN}`, "Content-Type": "application/json" },
  body: JSON.stringify({ visibility_timeout_ms: 6000, batch_size: 50 }),
});
const { result } = await pull.json<{ result: { messages: PullMessage[] } }>();

// 處理後,用 lease_id 逐則 ack 或 retry
await fetch(`${BASE}/messages/ack`, {
  method: "POST",
  headers: { Authorization: `Bearer ${API_TOKEN}`, "Content-Type": "application/json" },
  body: JSON.stringify({
    acks: succeeded.map((m) => ({ lease_id: m.lease_id })),
    retries: failed.map((m) => ({ lease_id: m.lease_id, delay_seconds: 60 })),
  }),
});

// ❌ 危險:沒有安全網,重試耗盡即丟棄
"consumers": [{ "queue": "jobs", "max_retries": 5 }]

// ✅ 正確:一律配 DLQ,失敗有帳可查
"consumers": [{ "queue": "jobs", "max_retries": 5, "dead_letter_queue": "jobs-dlq" }]

// ❌ 錯誤:所有失敗都無腦重試,毒訊息永遠出不去
catch (err) { msg.retry({ delaySeconds: 5 }); }

// ✅ 正確:分類處理,永久性錯誤送 DLQ、暫時性才退避重試
catch (err) {
  if (err instanceof PermanentError) { await env.DLQ.send(msg.body); msg.ack(); }
  else msg.retry({ delaySeconds: Math.min(30 * 2 ** (msg.attempts - 1), 43200) });
}

// 讀取上游 Retry-After,精準退避(比自算更禮貌)
if (res.status === 429) {
  const wait = parseInt(res.headers.get("Retry-After") ?? "60", 10);
  msg.retry({ delaySeconds: wait });
  continue;
}

// 情境 A:高吞吐批寫 D1(延遲不敏感)→ 大批次、長逾時
{ "max_batch_size": 100, "max_batch_timeout": 30, "max_concurrency": 10 }

// 情境 B:即時通知(延遲敏感)→ 小批次、短逾時
{ "max_batch_size": 5, "max_batch_timeout": 1, "max_concurrency": 20 }

// 情境 C:下游脆弱(外部 API 有嚴格額度)→ 壓低併發設天花板
{ "max_batch_size": 10, "max_batch_timeout": 5, "max_concurrency": 2 }