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));
}