cloudflare-queues

Cloudflare Queues

状态：生产就绪 ✅ 上次更新：2026-01-09 依赖项：cloudflare-worker-base（用于 Worker 设置） 最新版本：[email protected]，@cloudflare/[email protected]

近期更新（2025 年）：

• 2025 年 4 月：拉取式消费者提高了限制（每个队列 5,000 条消息/秒，之前为 1,200 个请求/5 分钟）
• 2025 年 3 月：暂停和清空 API（wrangler queues pause-delivery、queues purge）
• 2025 年：可自定义保留期限（60 秒至 14 天，之前固定为 4 天）
• 2025 年：提高队列限制（每个账户 10,000 个队列，之前为 10 个）

---

快速入门（5 分钟）

# 1. Create queue
npx wrangler queues create my-queue

# 2. Add producer binding to wrangler.jsonc
# { "queues": { "producers": [{ "binding": "MY_QUEUE", "queue": "my-queue" }] } }

# 3. Send message from Worker
await env.MY_QUEUE.send({ userId: '123', action: 'process-order' });

# Or publish via HTTP (May 2025+) from any service
curl -X POST "https://api.cloudflare.com/client/v4/accounts/{account_id}/queues/my-queue/messages" \
  -H "Authorization: Bearer YOUR_API_TOKEN" \
  -d '{"messages": [{"body": {"userId": "123"}}]}'

# 4. Add consumer binding to wrangler.jsonc
# { "queues": { "consumers": [{ "queue": "my-queue", "max_batch_size": 10 }] } }

# 5. Process messages
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      await processMessage(message.body);
      message.ack(); // Explicit acknowledgement
    }
  }
};

# 6. Deploy and test
npx wrangler deploy
npx wrangler tail my-consumer

---

生产者 API

// Send single message
await env.MY_QUEUE.send({ userId: '123', action: 'send-email' });

// Send with delay (max 12 hours)
await env.MY_QUEUE.send({ action: 'reminder' }, { delaySeconds: 600 });

// Send batch (max 100 messages or 256 KB)
await env.MY_QUEUE.sendBatch([
  { body: { userId: '1' } },
  { body: { userId: '2' } },
]);

关键限制：

• 消息大小：最大 128 KB（包含约 100 字节的元数据）
• 超过 128 KB 的消息将发送失败——请将其存储在 R2 中，并改为发送引用
• 批次大小：100 条消息或总计 256 KB
• 延迟：0–43200 秒（最长 12 小时）

---

HTTP 发布（2025 年 5 月起）

2025 年 5 月新增：通过 HTTP 从任何服务或编程语言向队列发布消息。

来源：Cloudflare 更新日志

身份验证：需要具有 Queues Edit 权限的 Cloudflare API 令牌。

# Single message
curl -X POST "https://api.cloudflare.com/client/v4/accounts/{account_id}/queues/my-queue/messages" \
  -H "Authorization: Bearer YOUR_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [
      {"body": {"userId": "123", "action": "process-order"}}
    ]
  }'

# Batch messages
curl -X POST "https://api.cloudflare.com/client/v4/accounts/{account_id}/queues/my-queue/messages" \
  -H "Authorization: Bearer YOUR_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [
      {"body": {"userId": "1"}},
      {"body": {"userId": "2"}},
      {"body": {"userId": "3"}}
    ]
  }'

使用场景：

• 从外部微服务发布（Node.js、Python、Go 等）
• 在 Cloudflare 之外运行的 Cron 作业
• Webhook 接收器
• 集成旧系统
• 未使用 Cloudflare Workers SDK 的服务

---

事件订阅（2025 年 8 月起）

2025 年 8 月新增：订阅 Cloudflare 服务中的事件，并通过 Queues 消费这些事件。

来源：Cloudflare 更新日志

支持的事件源：

• R2（bucket.created、object.uploaded、object.deleted 等）
• Workers KV
• Workers AI
• Vectorize
• Workflows
• Super Slurper
• Workers Builds

创建订阅：

npx wrangler queues subscription create my-queue \
  --source r2 \
  --events bucket.created,object.uploaded

事件结构：

interface CloudflareEvent {
  type: string;           // 'r2.bucket.created', 'kv.namespace.created'
  source: string;         // 'r2', 'kv', 'ai', etc.
  payload: any;           // Event-specific data
  metadata: {
    accountId: string;
    timestamp: string;
  };
}

消费者示例：

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      const event = message.body as CloudflareEvent;

      switch (event.type) {
        case 'r2.bucket.created':
          console.log('New R2 bucket:', event.payload.bucketName);
          await notifyAdmin(event.payload);
          break;

        case 'r2.object.uploaded':
          console.log('File uploaded:', event.payload.key);
          await processNewFile(event.payload.key);
          break;

        case 'kv.namespace.created':
          console.log('New KV namespace:', event.payload.namespaceId);
          break;

        case 'ai.inference.completed':
          console.log('AI inference done:', event.payload.modelId);
          break;
      }

      message.ack();
    }
  }
};

使用场景：

• 构建由 R2 上传触发的自定义工作流
• 监控基础设施变更（新增 KV 命名空间、存储桶）
• 跟踪 AI 推理作业
• 审计账户活动
• 无需自定义 Webhook 的事件驱动架构

---

消费者 API

export default {
  async queue(batch: MessageBatch, env: Env, ctx: ExecutionContext): Promise<void> {
    for (const message of batch.messages) {
      // message.id - unique UUID
      // message.timestamp - Date when sent
      // message.body - your content
      // message.attempts - retry count (starts at 1)

      await processMessage(message.body);
      message.ack(); // Explicit ack (critical for non-idempotent ops)
    }
  }
};

// Retry with exponential backoff
message.retry({ delaySeconds: Math.min(60 * Math.pow(2, message.attempts - 1), 3600) });

// Batch methods
batch.ackAll();   // Ack all messages
batch.retryAll(); // Retry all messages

重要事项：

• message.ack()——标记成功；即使处理程序之后失败，也不会重试
• 对非幂等操作使用显式 ack（数据库写入、API 调用、付款）
• 隐式 ack——如果处理程序成功返回且未调用 ack()，则所有消息都会自动确认
• 不保证顺序——不要假定消息按 FIFO 顺序排列

---

关键消费者模式

显式确认（非幂等操作）

**务必对以下操作使用显式 ack()：**数据库写入、API 调用、金融交易

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      try {
        await env.DB.prepare('INSERT INTO orders (id, amount) VALUES (?, ?)')
          .bind(message.body.orderId, message.body.amount).run();
        message.ack(); // Only ack on success
      } catch (error) {
        console.error(`Failed ${message.id}:`, error);
        // Don't ack - will retry
      }
    }
  }
};

**为什么？**如果批次中的一条消息失败，可避免重复写入。失败的消息会各自独立重试。

---

对受速率限制的 API 使用指数退避

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      try {
        await fetch('https://api.example.com/process', {
          method: 'POST',
          body: JSON.stringify(message.body),
        });
        message.ack();
      } catch (error) {
        if (error.status === 429) {
          const delaySeconds = Math.min(60 * Math.pow(2, message.attempts - 1), 3600);
          message.retry({ delaySeconds });
        } else {
          message.retry();
        }
      }
    }
  }
};

---

死信队列（DLQ）——生产环境至关重要

⚠️ 若未配置 DLQ，失败的消息在达到最大重试次数后会被永久删除

npx wrangler queues create my-dlq

wrangler.jsonc：

{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "max_retries": 3,
      "dead_letter_queue": "my-dlq"  // Messages go here after 3 failed retries
    }]
  }
}

DLQ 消费者：

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      console.error('PERMANENTLY FAILED:', message.id, message.body);
      await env.DB.prepare('INSERT INTO failed_messages (id, body) VALUES (?, ?)')
        .bind(message.id, JSON.stringify(message.body)).run();
      message.ack(); // Remove from DLQ
    }
  }
};

---

已知问题预防

此技能可避免 13 个已记录的问题：

问题 #1：运行多个开发命令——队列无法在进程间流转

错误：在一个 wrangler dev 进程中发送的队列消息，不会出现在另一个 wrangler dev 消费者进程中 来源：GitHub 问题 #9795

原因：wrangler 使用的虚拟队列存储在进程内存中。不同的开发进程无法共享队列状态。

预防措施：

# ❌ Don't run producer and consumer as separate processes
# Terminal 1: wrangler dev (producer)
# Terminal 2: wrangler dev (consumer)  # Won't receive messages!

# ✅ Option 1: Run both in single dev command
wrangler dev -c producer/wrangler.jsonc -c consumer/wrangler.jsonc

# ✅ Option 2: Use Vite plugin with auxiliaryWorkers
# vite.config.ts:
export default defineConfig({
  plugins: [
    cloudflare({
      auxiliaryWorkers: ['./consumer/wrangler.jsonc']
    })
  ]
})

---

Issue #2：队列生产者绑定导致 Remote Dev 出现 500 错误

错误：使用 wrangler dev --remote 和队列绑定时，所有路由都返回 500 Internal Server Error 来源：GitHub Issue #9642

原因：wrangler dev --remote 模式尚不支持队列。即使路由没有使用队列绑定，也会失败。

预防措施：

// When using remote dev, temporarily comment out queue bindings
{
  "queues": {
    // "producers": [{ "queue": "my-queue", "binding": "MY_QUEUE" }]
  }
}

// Or use local dev instead
// wrangler dev (without --remote)

---

Issue #3：设置 Queue Remote 后 D1 Remote 失效

错误：在队列生产者绑定上设置 remote: true 后，D1 远程绑定停止工作 来源：GitHub Issue #11106

原因：绑定冲突问题，影响本地与远程混合开发。

预防措施：

// ❌ Don't mix D1 remote with queue remote
{
  "d1_databases": [{
    "binding": "DB",
    "database_id": "...",
    "remote": true
  }],
  "queues": {
    "producers": [{
      "binding": "QUEUE",
      "queue": "my-queue",
      "remote": true  // ❌ Breaks D1 remote
    }]
  }
}

// ✅ Avoid remote on queues when using D1 remote
{
  "d1_databases": [{ "binding": "DB", "remote": true }],
  "queues": {
    "producers": [{ "binding": "QUEUE", "queue": "my-queue" }]
  }
}

状态：目前尚无解决方法。关注 Issue 以获取更新。

---

Issue #4：本地与远程绑定混用时缺少队列消费者

错误：将本地队列与远程 AI/Vectorize 绑定混用时，不会出现队列消费者绑定 来源：GitHub Issue #9887

原因：Wrangler 不支持在同一个 Worker 中混用本地与远程绑定。

预防措施：

// ❌ Don't mix local queues with remote AI
{
  "queues": {
    "consumers": [{ "queue": "my-queue" }]
  },
  "ai": {
    "binding": "AI",
    "experimental_remote": true  // ❌ Breaks queue consumer
  }
}

// ✅ Option 1: All local (no remote bindings)
wrangler dev

// ✅ Option 2: Separate workers for queues vs AI
// Worker 1: Queue processing (local)
// Worker 2: AI operations (remote)

---

Issue #5：http_pull 类型会阻止 Worker 消费者执行

错误：使用 type: "http_pull" 的队列消费者在生产环境中不会执行 来源：GitHub Issue #6619

原因：http_pull 用于基于外部 HTTP 的消费者，而非基于 Worker 的消费者。

预防措施：

// ❌ Don't use type: "http_pull" for Worker consumers
{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "type": "http_pull",  // ❌ Wrong for Workers
      "max_batch_size": 10
    }]
  }
}

// ✅ Omit type field for push-based Worker consumers
{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "max_batch_size": 10
      // No "type" field - defaults to Worker consumer
    }]
  }
}

---

重大变更与弃用

生产者配置中的 delivery_delay（即将移除）

警告：生产者绑定中的 delivery_delay 参数将在未来版本的 wrangler 中移除。

来源：GitHub Issue #10286

// ❌ Will be removed - don't use
{
  "queues": {
    "producers": [{
      "binding": "MY_QUEUE",
      "queue": "my-queue",
      "delivery_delay": 300  // ❌ Don't use this
    }]
  }
}

迁移：改用逐条消息设置延迟：

// ✅ Correct approach - per-message delay
await env.MY_QUEUE.send({ data }, { delaySeconds: 300 });

原因：Worker 不应影响队列级别的设置。存在多个生产者时，最后部署的生产者所设置的值会生效，导致行为不可预测。

---

社区提示

注意：这些提示来自社区讨论和 GitHub Issues。请根据你的 wrangler 版本进行验证。

提示：max_batch_timeout 可能会影响本地开发

来源：GitHub Issue #6619 可信度：中 适用范围：使用 wrangler dev 进行本地开发

如果队列消费者在本地无法执行，请尝试移除 max_batch_timeout：

{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "max_batch_size": 10
      // Remove max_batch_timeout for local dev
    }]
  }
}

这似乎与版本有关，可能不会影响所有配置。

---

提示：生产者绑定无法获取队列名称

来源：GitHub Issue #10131 可信度：高 适用范围：多环境部署（预发布环境、PR 预览、租户专属队列）

队列名称只能通过消费者处理程序中的 batch.queue 获取，而不能通过生产者绑定获取。这会导致环境专属队列名称（如 email-queue-staging 或 email-queue-pr-123）出现问题。

当前限制：

// ❌ Can't access queue name from producer binding
const queueName = env.MY_QUEUE.name; // Doesn't exist!

// ❌ Must hardcode or normalize in consumer
switch (batch.queue) {
  case 'email-queue':           // What about email-queue-staging?
  case 'email-queue-staging':   // Must handle all variants
  case 'email-queue-pr-123':    // Dynamic env names break this
}

解决方法：

// In consumer: Normalize queue name
function normalizeQueueName(queueName: string): string {
  return queueName.replace(/-staging|-pr-\d+|-dev/g, '');
}

switch (normalizeQueueName(batch.queue)) {
  case 'email-queue':
    // Handle all email-queue-* variants
}

状态：功能请求已在内部跟踪：MQ-923

---

消费者配置

{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "max_batch_size": 100,           // 1-100 (default: 10)
      "max_batch_timeout": 30,         // 0-60s (default: 5s)
      "max_retries": 5,                // 0-100 (default: 3)
      "retry_delay": 300,              // Seconds (default: 0)
      "max_concurrency": 10,           // 1-250 (default: auto-scale)
      "dead_letter_queue": "my-dlq"    // REQUIRED for production
    }]
  }
}

关键设置：

• 批处理 - 满足任一条件（max_batch_size 或 max_batch_timeout）时调用消费者
• max_retries - 重试次数耗尽后：有 DLQ → 发送到 DLQ；无 DLQ → 永久删除
• max_concurrency - 仅当上游有速率限制或连接数限制时才设置。否则留空以启用自动扩缩（最多 250 个并发调用）
• DLQ - 需单独创建：npx wrangler queues create my-dlq

---

Wrangler 命令

# Create queue
npx wrangler queues create my-queue
npx wrangler queues create my-queue --message-retention-period-secs 1209600  # 14 days

# Manage queues
npx wrangler queues list
npx wrangler queues info my-queue
npx wrangler queues delete my-queue  # ⚠️ Deletes ALL messages!

# Pause/Purge (March 2025 - NEW)
npx wrangler queues pause-delivery my-queue   # Pause processing, keep receiving
npx wrangler queues resume-delivery my-queue
npx wrangler queues purge my-queue            # ⚠️ Permanently deletes all messages!

# Consumer management
npx wrangler queues consumer add my-queue my-consumer-worker \
  --batch-size 50 --batch-timeout 10 --message-retries 5
npx wrangler queues consumer remove my-queue my-consumer-worker

---

限制与配额

功能	限制
每个账户的队列数	10,000
消息大小	128 KB（包含约 100 字节的元数据）
消息重试次数	最多 100 次
批次大小	1-100 条消息
批次超时时间	0-60 秒
每次 sendBatch 的消息数	100 条（或总计 256 KB）
队列吞吐量	每个队列每秒 5,000 条消息
消息保留时间	4 天（默认），14 天（最长）
队列积压大小	每个队列 25 GB
并发使用者	250 个（基于推送，自动扩缩容）
使用者运行时长	15 分钟（墙上时钟时间）
使用者 CPU 时间	30 秒（默认），5 分钟（最长）
可见性超时	12 小时（拉取式使用者）
消息延迟	12 小时（最长）
API 速率限制	每 5 分钟 1200 个请求

---

定价

需要 Workers 付费计划（$5/月）

操作定价：

• 每月前 1,000,000 次操作：免费
• 超出部分：每百万次操作 $0.40

计为一次操作的情况：

• 每写入、读取或删除 64 KB 数据块，计为一次操作
• 大于 64 KB 的消息计为多次操作：
  • 65 KB 的消息 = 2 次操作
  • 127 KB 的消息 = 2 次操作
  • 128 KB 的消息 = 2 次操作

典型消息生命周期：

• 1 次写入 + 1 次读取 + 1 次删除 = 3 次操作

重试：

• 每次重试 = 额外 1 次读取操作
• 消息重试 3 次 = 1 次写入 + 4 次读取 + 1 次删除 = 6 次操作

死信队列：

• 写入 DLQ = 额外 1 次写入操作

费用示例：

• 每月 1M 条消息（无重试）：((1M × 3) - 1M) / 1M × $0.40 = $0.80
• 每月 10M 条消息：((10M × 3) - 1M) / 1M × $0.40 = $11.60
• 每月 100M 条消息：((100M × 3) - 1M) / 1M × $0.40 = $119.60

---

错误处理

常见错误

1. 消息过大

// ❌ Bad: Message >128 KB
await env.MY_QUEUE.send({
  data: largeArray, // >128 KB
});

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

if (size > 128000) {
  // Store in R2, send reference
  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. 超出吞吐量限制

// ❌ Bad: Exceeding 5000 msg/s per queue
for (let i = 0; i < 10000; i++) {
  await env.MY_QUEUE.send({ id: i }); // Too fast!
}

// ✅ Good: Use sendBatch
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));
}

// ✅ Even better: Rate limit with delay
for (let i = 0; i < messages.length; i += 100) {
  await env.MY_QUEUE.sendBatch(messages.slice(i, i + 100));
  if (i + 100 < messages.length) {
    await new Promise(resolve => setTimeout(resolve, 100)); // 100ms delay
  }
}

---

3. 使用者超时

// ❌ Bad: Long processing without CPU limit increase
export default {
  async queue(batch: MessageBatch): Promise<void> {
    for (const message of batch.messages) {
      await processForMinutes(message.body); // CPU timeout!
    }
  },
};

// ✅ Good: Increase CPU limit in wrangler.jsonc

wrangler.jsonc：

{
  "limits": {
    "cpu_ms": 300000  // 5 minutes (max allowed)
  }
}

---

4. 积压持续增长

// Issue: Consumer too slow, backlog growing

// ✅ Solution 1: Increase batch size
{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "max_batch_size": 100  // Process more per invocation
    }]
  }
}

// ✅ Solution 2: Let concurrency auto-scale (don't set max_concurrency)

// ✅ Solution 3: Optimize consumer code
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    // Process in parallel
    await Promise.all(
      batch.messages.map(async (message) => {
        await process(message.body);
        message.ack();
      })
    );
  },
};

---

关键规则

务必：

• ✅ 为生产环境配置 DLQ（在使用者配置中设置 dead_letter_queue）
• ✅ 对非幂等操作（数据库写入、API 调用）使用显式的 message.ack()
• ✅ 发送前验证消息大小小于 128 KB
• ✅ 多条消息使用 sendBatch()（效率更高）
• ✅ 实现指数退避：60 * Math.pow(2, message.attempts - 1)
• ✅ 允许并发自动扩缩容（除非上游有速率限制，否则不要设置 max_concurrency）

切勿：

• ❌ 切勿假设消息按 FIFO 顺序排列——不保证顺序
• ❌ 对非幂等操作切勿依赖隐式 ack——请使用显式的 ack()
• ❌ 切勿发送大于 128 KB 的消息——将会失败（请改为存储在 R2 中）
• ❌ 生产环境切勿省略 DLQ——没有 DLQ 时，失败的消息会被永久删除
• ❌ 切勿超过每个队列每秒 5,000 条消息（推送式使用者），否则会受到速率限制
• ❌ 切勿同步处理消息——使用 Promise.all() 实现并行处理

---

故障排查

问题：消息未传递给使用者

可能原因：

1. 使用者尚未部署
2. wrangler.jsonc 中的队列名称错误
3. 投递已暂停
4. 使用者抛出错误

解决方案：

# Check queue info
npx wrangler queues info my-queue

# Check if delivery paused
npx wrangler queues resume-delivery my-queue

# Check consumer logs
npx wrangler tail my-consumer

---

问题：一条消息失败时整个批次都会重试

原因： 对非幂等操作使用了隐式确认

解决方案： 使用显式 ack()

// ✅ Explicit ack
for (const message of batch.messages) {
  try {
    await dbWrite(message.body);
    message.ack(); // Only ack on success
  } catch (error) {
    console.error(`Failed: ${message.id}`);
    // Don't ack - will retry
  }
}

---

问题：消息未经处理便被删除

原因： 未配置死信队列

解决方案：

# Create DLQ
npx wrangler queues create my-dlq

# Add to consumer config

{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "dead_letter_queue": "my-dlq"
    }]
  }
}

---

问题：使用者未自动扩缩容

可能原因：

1. max_concurrency 设置为 1
2. 使用者返回错误（未处理消息）
3. 批次处理过快（没有积压）

解决方案：

{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      // Don't set max_concurrency - let it auto-scale
      "max_batch_size": 50  // Increase batch size instead
    }]
  }
}

---

相关文档

• Cloudflare Queues 文档
• Queues 的工作原理
• JavaScript API
• 批处理与重试
• 消费者并发
• 死信队列
• Wrangler 命令
• 限制
• 定价

---

最后更新：2026-01-21 版本：2.0.0 变更：新增 HTTP 发布（2025 年 5 月）、事件订阅（2025 年 8 月）、已知问题预防（13 个问题）、破坏性变更章节和社区技巧。错误数量：0 → 13。新增主要功能并全面记录问题。 维护者：Jeremy Dawes | [email protected]