cloudflare-queues

Cloudflare 队列

状态：已可投入生产 ✅ | 上次验证时间：2025-12-27

依赖项：cloudflare-worker-base（用于 Worker 设置）

目录：快速入门 • 关键规则 • 常见错误 • 使用场景 • 何时加载参考资料 • 限制

---

快速入门（10 分钟）

1. 创建队列

bunx wrangler queues create my-queue
bunx wrangler queues list

2. 生产者（发送消息）

wrangler.jsonc：

{
  "name": "my-producer",
  "main": "src/index.ts",
  "queues": {
    "producers": [
      {
        "binding": "MY_QUEUE",
        "queue": "my-queue"
      }
    ]
  }
}

src/index.ts：

import { Hono } from 'hono';

type Bindings = {
  MY_QUEUE: Queue;
};

const app = new Hono<{ Bindings: Bindings }>();

app.post('/send', async (c) => {
  await c.env.MY_QUEUE.send({
    userId: '123',
    action: 'process-order',
    timestamp: Date.now(),
  });

  return c.json({ status: 'queued' });
});

export default app;

3. 消费者（处理消息）

wrangler.jsonc：

{
  "name": "my-consumer",
  "main": "src/consumer.ts",
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_batch_size": 10,
        "max_retries": 3,
        "dead_letter_queue": "my-dlq"
      }
    ]
  }
}

src/consumer.ts：

import type { MessageBatch } from '@cloudflare/workers-types';

export default {
  async queue(batch: MessageBatch): Promise<void> {
    for (const message of batch.messages) {
      console.log('Processing:', message.body);
      // Your logic here
    }
    // Implicit ack: returning successfully acknowledges all messages
  },
};

部署：

bunx wrangler deploy

加载：references/setup-guide.md，查看包含 DLQ 配置的完整 6 步设置指南

---

关键规则

始终执行 ✅

1. 为生产队列配置死信队列
2. 对非幂等操作（数据库写入、API 调用）使用显式 ack()
3. 发送前验证消息大小（<128 KB）
4. 发送多条消息时使用 sendBatch()（效率更高）
5. 为重试实现指数退避
6. 根据工作负载设置适当的批处理参数
7. 监控队列积压和消费者错误
8. 在消费者中使用 ctx.waitUntil() 进行异步清理
9. 妥善处理错误——记录日志、发出警报、重试
10. 让并发自动扩缩容（除非确有需要，否则不要设置 max_concurrency）

绝不执行 ❌

1. 绝不假设消息会按顺序处理——不保证先进先出（FIFO）
2. 绝不对非幂等操作依赖隐式 ack——使用显式 ack()
3. 绝不发送超过 128 KB 的消息——此类操作会失败
4. 绝不删除含有活动消息的队列——会导致数据丢失
5. 在生产环境中绝不跳过 DLQ 配置
6. 绝不超过每个队列 5000 条消息/秒——会触发速率限制错误
7. 绝不在循环中同步处理消息——使用 Promise.all()
8. 绝不忽略 message.attempts——将其用于退避逻辑
9. 除非有非常具体的理由，否则绝不设置 max_concurrency=1
10. 使用显式确认模式时绝不忘记调用 ack()

---

3 个关键错误

错误 #1：消息过大

问题：消息超过 128 KB 限制

解决方案：将大数据存储在 R2 中，并发送其引用

// ❌ Wrong
await env.MY_QUEUE.send({ data: largeArray }); // >128 KB fails

// ✅ Correct
const message = { data: largeArray };
const size = new TextEncoder().encode(JSON.stringify(message)).length;

if (size > 128000) {
  const key = `messages/${crypto.randomUUID()}.json`;
  await env.MY_BUCKET.put(key, JSON.stringify(message));
  await env.MY_QUEUE.send({ type: 'large-message', r2Key: key });
} else {
  await env.MY_QUEUE.send(message);
}

错误 #2：吞吐量超限

问题：每个队列每秒处理超过 5000 条消息

解决方案：使用 sendBatch() 并实施速率限制

// ❌ Wrong
for (let i = 0; i < 10000; i++) {
  await env.MY_QUEUE.send({ id: i }); // Too fast!
}

// ✅ Correct
const messages = Array.from({ length: 10000 }, (_, i) => ({
  body: { id: i },
}));

// Send in batches of 100
for (let i = 0; i < messages.length; i += 100) {
  await env.MY_QUEUE.sendBatch(messages.slice(i, i + 100));
}

错误 #3：一条消息失败导致整个批次重试

问题：单条消息失败会导致所有消息重试

解决方案：使用显式确认

// ❌ Wrong - implicit ack
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      await env.DB.prepare('INSERT INTO orders VALUES (?, ?)').bind(
        message.body.id,
        message.body.amount
      ).run();
    }
    // If any fails, ALL retry!
  },
};

// ✅ Correct - explicit ack
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      try {
        await env.DB.prepare('INSERT INTO orders VALUES (?, ?)').bind(
          message.body.id,
          message.body.amount
        ).run();

        message.ack(); // Only ack on success
      } catch (error) {
        console.error(`Failed: ${message.id}`, error);
        // Don't ack - will retry independently
      }
    }
  },
};

加载 references/error-catalog.md，查看全部 10 个错误及其解决方案，包括 DLQ 配置、自动扩缩容问题、防止消息删除和详细解决方案。

---

常见使用场景

使用场景 1：基本消息队列

适用场景：简单的异步作业处理（电子邮件、通知）

快速模式：

// Producer
await env.MY_QUEUE.send({ type: 'email', to: '[email protected]' });

// Consumer (implicit ack - for idempotent operations)
export default {
  async queue(batch: MessageBatch): Promise<void> {
    for (const message of batch.messages) {
      await sendEmail(message.body.to, message.body.content);
    }
  },
};

加载：templates/queues-producer.ts + templates/queues-consumer-basic.ts

使用场景 2：数据库写入（非幂等）

适用场景：写入数据库时必须避免重复

加载：templates/queues-consumer-explicit-ack.ts + references/consumer-api.md

使用场景 3：指数退避重试

适用场景：调用受速率限制的 API、处理临时故障

加载：templates/queues-retry-with-delay.ts + references/error-catalog.md（错误 #2、#3）

使用场景 4：死信队列模式

适用场景：生产系统需要捕获永久失败的消息

加载：templates/queues-dlq-pattern.ts + references/setup-guide.md（第 4 步）

使用场景 5：高吞吐量处理

适用场景：每秒处理数千条消息

快速模式：

{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "max_batch_size": 100,        // Large batches
      "max_batch_timeout": 5,       // Fast processing
      "max_concurrency": null       // Auto-scale
    }]
  }
}

加载：references/best-practices.md → 优化吞吐量

---

何时加载参考资料

在以下情况下加载 references/setup-guide.md：

• 用户需要完整的设置指南（队列 → 生产者 → 消费者 → DLQ）
• 首次设置 Cloudflare Queues
• 需要生产环境配置示例
• 需要完整的端到端示例

在以下情况下加载 references/error-catalog.md：

• 遇到文档中记录的 10 种错误中的任意一种
• 排查队列问题
• 消息未能送达或处理
• 需要预防措施清单

在以下情况下加载 references/producer-api.md：

• 需要完整的生产者 API 参考
• 使用 send() 或 sendBatch() 方法
• 需要消息格式规范
• 处理大型消息或批次

在以下情况下加载 references/consumer-api.md：

• 需要完整的消费者 API 参考
• 使用显式的 ack()、retry() 或 retryAll()
• 需要批处理模式
• 实现复杂的消费者逻辑

在以下情况下加载 references/best-practices.md：

• 优化队列性能
• 需要生产环境部署指南
• 设置监控和可观测性
• 需要安全性和可靠性模式

在以下情况下加载 references/wrangler-commands.md：

• 需要 CLI 命令参考
• 管理队列（创建、删除、列出）
• 控制投递（暂停、恢复）
• 调试队列问题
• 实时监控和性能分析

在以下情况下加载 references/typescript-types.md：

• 需要完整的 TypeScript 类型定义
• 使用 Queue、MessageBatch 或 Message 接口
• 实现类型安全的生产者或消费者
• 为消息正文使用泛型类型
• 需要用于消息验证的类型守卫

在以下情况下加载 references/production-checklist.md：

• 准备部署到生产环境
• 需要部署前验证清单
• 需要生产环境最佳实践的详细说明
• 设置监控、DLQ 或错误处理
• 规划负载测试或安全审查

在以下情况下加载 references/pull-consumers.md：

• 需要在 Workers 环境之外消费消息
• 与现有后端服务（Node.js、Python、Go）集成
• 实现基于拉取的轮询，而非基于推送的消费
• 使用容器化或旧版应用程序

在以下情况下加载 references/http-publishing.md：

• 通过 HTTP 从外部系统发布消息
• 集成第三方服务的 Webhook
• 需要在不部署 Workers 的情况下发送消息
• 实现 CI/CD 流水线通知

在以下情况下加载 references/r2-event-integration.md：

• 在 R2 存储桶事件发生时触发队列消息
• 实现自动化图像或文档处理
• 设置事件驱动的数据流水线
• 需要 R2 对象上传或删除通知

---

Agent 和命令

可用 Agent：

• queue-debugger - 针对队列问题进行 9 阶段诊断分析（系统化排查）
• queue-optimizer - 性能调优和成本优化（批次大小、并发、重试）

可用命令：

• /queue-setup - 用于完整队列设置的交互式向导
• /queue-troubleshoot - 快速诊断常见问题
• /queue-monitor - 实时指标和状态显示

---

限制和配额

重要限制：

• 消息大小：每条消息最大 128 KB
• 吞吐量：每个队列每秒最多 5,000 条消息
• 批次大小：每次 sendBatch() 最多 100 条消息
• 消费者 CPU 时间：30 秒（默认），300 秒（通过配置设置的上限）
• 最大重试次数：可配置（默认 3 次）
• 队列名称：最多 63 个字符，仅限小写字母、数字和连字符

加载 references/best-practices.md，了解如何处理限制及优化策略。

---

配置参考

生产者：在 wrangler.jsonc 的 queues.producers 数组中添加队列绑定，并设置 binding 和 queue 字段。

消费者：在 wrangler.jsonc 的 queues.consumers 数组中配置 queue、max_batch_size（1-100）、max_batch_timeout（0-60s）、max_retries、dead_letter_queue，以及可选的 max_concurrency（默认：自动扩缩容）。

CPU 限制：如果处理所需时间较长，请将 limits.cpu_ms 从默认的 30,000ms 调高。

加载 references/setup-guide.md 查看完整配置示例，并加载 templates/wrangler-queues-config.jsonc 查看适用于生产环境的配置。

---

使用捆绑资源

参考文档（references/）

• setup-guide.md - 完整的 6 步设置指南（队列 → 生产者 → 消费者 → DLQ → 部署 → 生产环境配置）
• error-catalog.md - 全部 10 种错误的解决方案 + 预防检查清单
• producer-api.md - 完整的生产者 API（send、sendBatch、消息格式）
• consumer-api.md - 完整的消费者 API（ack、retry、批处理）
• best-practices.md - 性能、监控、安全和可靠性模式
• wrangler-commands.md - CLI 参考（创建、删除、列出、暂停、恢复）
• typescript-types.md - Queue、MessageBatch、Message 的完整 TypeScript 类型定义
• production-checklist.md - 部署前验证和最佳实践
• pull-consumers.md - 适用于非 Workers 环境的拉取式消费者（HTTP 轮询）
• http-publishing.md - 从外部系统通过 HTTP API 发布消息
• r2-event-integration.md - 触发队列消息的 R2 事件通知

模板（templates/）

• queues-producer.ts - 使用单条和批量发送的基础生产者
• queues-consumer-basic.ts - 隐式 ack 消费者（幂等操作）
• queues-consumer-explicit-ack.ts - 显式 ack 消费者（非幂等操作）
• queues-retry-with-delay.ts - 指数退避重试模式
• queues-dlq-pattern.ts - 死信队列设置和消费者
• wrangler-queues-config.jsonc - 完整的生产环境配置

---

TypeScript 类型

使用 @cloudflare/workers-types 包获取完整类型定义：Queue、MessageBatch<Body>、Message<Body>、QueueSendOptions。

加载 references/typescript-types.md 查看完整类型参考，包含接口、泛型、类型守卫和用法示例。

---

监控与调试

关键命令：wrangler queues info（状态）、wrangler tail（日志）、wrangler queues pause-delivery/resume-delivery（控制）。

加载 references/wrangler-commands.md 查看完整 CLI 参考，包含实时监控、调试工作流和性能分析命令。

---

生产环境检查清单

部署前的 12 项检查清单：DLQ 配置、消息确认策略、大小验证、批处理优化、并发设置、CPU 限制、错误处理、监控、速率限制、幂等性、负载测试和安全审查。

加载 references/production-checklist.md 查看完整检查清单，包含详细说明、代码示例和部署工作流。

---

相关技能

• cloudflare-worker-base - Worker 设置和配置
• cloudflare-d1 - 队列消费者的数据库集成
• cloudflare-r2 - 存储大型消息负载
• cloudflare-workflows - 更复杂的编排需求

---

官方文档

• Cloudflare Queues：https://developers.cloudflare.com/queues/
• 配置：https://developers.cloudflare.com/queues/configuration/
• Wrangler 命令：https://developers.cloudflare.com/workers/wrangler/commands/#queues
• 限制：https://developers.cloudflare.com/queues/platform/limits/
• 故障排除：https://developers.cloudflare.com/queues/observability/troubleshooting/
• 最佳实践：https://developers.cloudflare.com/queues/configuration/best-practices/

---

有问题？遇到故障？

1. 查看 references/error-catalog.md，了解全部 10 个错误及其解决方案
2. 阅读 references/setup-guide.md，查看完整的设置指南
3. 查看 references/best-practices.md，了解生产环境模式
4. 查看官方文档：https://developers.cloudflare.com/queues/