Queues 消息队列

这一课你将能做到

1. 创建 Queue、配置 Producer 和 Consumer Worker，实现生产消费模型
2. 处理批量消息、错误重试、死信队列（DLQ）
3. 判断 Queues 和 ctx.waitUntil、Cron Triggers 的适用边界

为什么要学

用户在页面上点击「生成报告」，这个操作可能要调用 AI 模型、写数据库、发邮件，耗时 30 秒。如果让用户干等 30 秒，页面早就超时了。正确的做法是：用户点击后，Worker 立刻返回「任务已提交」，然后把真正的工作丢进队列，让另一个 Worker 慢慢处理。

人话：Queues 就是 Workers 之间的「异步任务管道」。它解决三个问题：(1) 耗时任务不阻塞用户请求；(2) 流量削峰，突发请求不会压垮后端；(3) 任务失败自动重试，保证最终执行。

先建立图像

[前端请求]
    │
    ▼
[Producer Worker] ── 消息 A ──> [Queue]
    │                           │
    └── 立刻返回响应             ├── 消息 B
                                │
                                ▼
                        [Consumer Worker]
                            │
                            ├── 处理消息 A (重试 3 次)
                            └── 处理消息 B (失败 -> DLQ)

跟着做

步骤 1：创建 Queue

npx wrangler queues create report-queue

步骤 2：Producer Worker（生产消息）

wrangler.toml：

name = "producer"
[[queues.producers]]
binding = "REPORT_QUEUE"
queue = "report-queue"

export interface Env {
  REPORT_QUEUE: Queue;
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const body = await request.json();
    
    // 把任务丢进队列，立刻返回
    await env.REPORT_QUEUE.send({
      userId: body.userId,
      type: 'generate-report',
      params: body.params,
      timestamp: Date.now(),
    });
    
    return Response.json({ status: 'queued', message: 'Report generation has been queued.' });
  },
} satisfies ExportedHandler<Env>;

步骤 3：Consumer Worker（消费消息）

创建另一个 Worker，或同一个项目里再配一个 consumer。

wrangler.toml：

name = "consumer"
[[queues.consumers]]
queue = "report-queue"
max_batch_size = 10
max_batch_timeout = 30
max_retries = 3
dead_letter_queue = "report-dlq"

export interface Env {
  DB: D1Database;
}

export default {
  async queue(batch: MessageBatch, env: Env, ctx: ExecutionContext): Promise<void> {
    for (const message of batch.messages) {
      const task = message.body as any;
      
      try {
        await generateReport(task);
        await env.DB.prepare('UPDATE reports SET status = "done" WHERE id = ?')
          .bind(task.userId)
          .run();
        message.ack(); // 确认消费成功
      } catch (err) {
        console.error('Failed to process message:', err);
        message.retry(); // 让平台重试（最多 max_retries 次）
      }
    }
  },
} satisfies ExportedHandler<Env>;

async function generateReport(task: any) {
  // 模拟耗时任务
  await new Promise(r => setTimeout(r, 2000));
}

步骤 4：创建死信队列（DLQ）

npx wrangler queues create report-dlq

消费者配置里的 dead_letter_queue = "report-dlq" 会在消息重试耗尽后自动把消息丢进 DLQ。你可以再配一个 DLQ consumer 做人工审计或告警。

步骤 5：通过 Dashboard 监控队列

Dashboard → Workers & Pages → Queues → report-queue。可以看到：

• 队列深度（积压消息数）
• 消费速率
• 失败重试次数
• DLQ 积压

讲清楚要点

1. Queue 是 Workers 之间通信的桥梁：Producer 和 Consumer 可以是两个完全独立的 Worker，通过 Queue 解耦。
2. Consumer 是事件驱动，不是轮询：Cloudflare 会在有消息时自动触发 Consumer，不需要你自己写 setInterval 去 poll。
3. 批量消费效率高：max_batch_size 可以一次拉多条消息处理，减少触发次数。但要注意：一批中只要有一条没 ack()，整批可能重试。
4. 死信队列是兜底：重试次数用尽的消息不会消失，而是进入 DLQ。你可以定期检查 DLQ，分析失败原因。
5. 消息体有大小限制：单条消息最大 128KB。大文件用 R2 存，Queue 里只传引用。

你可能会懵的地方

• Queue 和 ctx.waitUntil 有什么区别？ waitUntil 是在当前 Worker 里异步执行，不保证一定完成，且受 30 秒限制；Queue 是持久化的消息队列，Consumer 独立运行，有重试、DLQ、批量处理。

• Queue 能保证顺序吗？ 不保证全局有序。同一批内的消息按接收顺序处理，但不同批次之间没有顺序保证。需要顺序处理时用 Durable Objects。

• Consumer 挂了消息会丢吗？ 不会。消息在 Consumer 成功 ack() 前会一直保留在队列里。如果 Consumer 崩溃，消息会重新投递。

• Queue 免费吗？ 免费套餐有每日消息数限制（当前约 100,000 条/天）。超出后需要付费。

• Queue 能跨账号吗？ 不能直接跨账号共享。Producer 和 Consumer 必须在同一个 Cloudflare Account 下。

错法对照

错法	问题	正解
在 queue() handler 里不调用 ack()	消息被认为处理失败，无限重试	成功处理完必须 message.ack()
在 Consumer 里做同步阻塞 I/O	拖慢整批消费	用异步并行处理，或缩小 batch size
把大文件直接塞进消息体	超过 128KB 限制，发送失败	大文件用 R2，Queue 里传引用
不配置 DLQ	消息重试耗尽后丢失	重要队列必须配 dead_letter_queue
用 Queue 做实时通信	Queue 有延迟（秒级）	实时通信用 WebSocket + Durable Objects

卡点急救

1. 消息一直不进 Consumer：检查 wrangler.toml 的 queues.consumers 配置是否正确；检查 Consumer Worker 是否已部署；检查 Queue 名是否拼写一致。
2. Consumer 频繁触发但处理很慢：增大 max_batch_size，让一次触发处理更多消息；或优化 Consumer 的处理逻辑。
3. DLQ 积压越来越多：说明 Consumer 有系统性 bug。暂停 Consumer，修复代码，然后手动重试 DLQ 里的消息（Dashboard 支持批量重试）。

面试常问

• Queue 和 Pub/Sub 有什么区别？ Queue 是「点对点」：一条消息只会被一个 Consumer 处理；Pub/Sub 是「广播」：一条消息可以被多个订阅者收到。Cloudflare Queues 目前更接近点对点模型。

• 消息积压了怎么办？ 先看 Consumer 是否挂掉；再看 Consumer 处理速度是否跟不上生产速度。解决方向：(1) 扩容 Consumer（增加并发，增大 batch size）；(2) 优化 Consumer 逻辑减少单次处理时间；(3) 限流 Producer。

• 如何保证消息至少被处理一次？ Queues 天然支持至少一次投递（at-least-once）。Consumer 需要做幂等处理（如用消息 ID 去重），防止重复处理带来的副作用。

自测三问

1. ctx.waitUntil 和 Queue 分别在什么场景下使用？
2. 如果 Consumer 处理消息时抛出异常，Queue 会怎么做？
3. 怎么保证 Queue Consumer 不会重复处理同一条消息？（提示：幂等）

小结一句话

Queues 让 Workers 拥有了生产级的异步任务管道：解耦、削峰、重试、DLQ 兜底，是边缘架构中处理耗时任务的标配。

延伸

• Queues 官方文档
• Queues Pricing & Limits
• 下一课：Durable Objects，学习强一致状态与协调原语。