cloudflare-queues

Cloudflare Queues

状态：已可投入生产 ✅ 最后更新：2025-10-21 依赖项：cloudflare-worker-base（用于 Worker 设置） 最新版本：[email protected], @cloudflare/[email protected]

---

快速入门（10 分钟）

1. 创建队列

# Create a new queue
npx wrangler queues create my-queue

# Output:
# ✅ Successfully created queue my-queue

# List all queues
npx wrangler queues list

# Get queue info
npx wrangler queues info my-queue

2. 设置生产者（发送消息）

wrangler.jsonc：

{
  "name": "my-producer",
  "main": "src/index.ts",
  "compatibility_date": "2025-10-11",
  "queues": {
    "producers": [
      {
        "binding": "MY_QUEUE",           // Available as env.MY_QUEUE
        "queue": "my-queue"               // Queue name from step 1
      }
    ]
  }
}

src/index.ts（生产者）：

import { Hono } from 'hono';

type Bindings = {
  MY_QUEUE: Queue;
};

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

// Send single message
app.post('/send', async (c) => {
  const body = await c.req.json();

  await c.env.MY_QUEUE.send({
    userId: body.userId,
    action: 'process-order',
    timestamp: Date.now(),
  });

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

// Send batch of messages
app.post('/send-batch', async (c) => {
  const items = await c.req.json();

  await c.env.MY_QUEUE.sendBatch(
    items.map((item) => ({
      body: { userId: item.userId, action: item.action },
    }))
  );

  return c.json({ status: 'queued', count: items.length });
});

export default app;

3. 设置消费者（处理消息）

创建消费者 Worker：

# In a new directory
npm create cloudflare@latest my-consumer -- --type hello-world --ts
cd my-consumer

wrangler.jsonc：

{
  "name": "my-consumer",
  "main": "src/index.ts",
  "compatibility_date": "2025-10-11",
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",              // Queue to consume from
        "max_batch_size": 10,             // Process up to 10 messages at once
        "max_batch_timeout": 5            // Or wait max 5 seconds
      }
    ]
  }
}

src/index.ts（消费者）：

export default {
  async queue(
    batch: MessageBatch,
    env: Env,
    ctx: ExecutionContext
  ): Promise<void> {
    console.log(`Processing batch of ${batch.messages.length} messages`);

    for (const message of batch.messages) {
      console.log('Message:', message.id, message.body, `Attempt: ${message.attempts}`);

      // Your processing logic here
      await processMessage(message.body);
    }

    // Implicit acknowledgement: if this function returns without error,
    // all messages are automatically acknowledged
  },
};

async function processMessage(body: any) {
  // Process the message
  console.log('Processing:', body);
}

4. 部署并测试

# Deploy producer
cd my-producer
npm run deploy

# Deploy consumer
cd my-consumer
npm run deploy

# Test by sending a message
curl -X POST https://my-producer.<your-subdomain>.workers.dev/send \
  -H "Content-Type: application/json" \
  -d '{"userId": "123", "action": "welcome-email"}'

# Watch consumer logs
npx wrangler tail my-consumer

---

完整的生产者 API

send() - 发送单条消息

interface QueueSendOptions {
  delaySeconds?: number;  // Delay delivery (0-43200 seconds / 12 hours)
}

await env.MY_QUEUE.send(body: any, options?: QueueSendOptions);

示例：

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

// Send with delay (10 minutes)
await env.MY_QUEUE.send(
  { userId: '123', action: 'reminder' },
  { delaySeconds: 600 }
);

// Send structured data
await env.MY_QUEUE.send({
  type: 'order-confirmation',
  orderId: 'ORD-123',
  email: '[email protected]',
  items: [{ sku: 'ITEM-1', quantity: 2 }],
  total: 49.99,
  timestamp: Date.now(),
});

重要：

• 消息正文必须可通过 JSON 序列化（结构化克隆算法）
• 消息大小上限：128 KB（包括约 100 字节的内部元数据）
• 超过 128 KB 的消息将发送失败——请拆分消息，或将其存储在 R2 中并发送引用

---

sendBatch() - 发送多条消息

interface MessageSendRequest<Body = any> {
  body: Body;
  delaySeconds?: number;
}

interface QueueSendBatchOptions {
  delaySeconds?: number;  // Default delay for all messages
}

await env.MY_QUEUE.sendBatch(
  messages: Iterable<MessageSendRequest>,
  options?: QueueSendBatchOptions
);

示例：

// Send batch of messages
await env.MY_QUEUE.sendBatch([
  { body: { userId: '1', action: 'email' } },
  { body: { userId: '2', action: 'email' } },
  { body: { userId: '3', action: 'email' } },
]);

// Send batch with individual delays
await env.MY_QUEUE.sendBatch([
  { body: { task: 'task1' }, delaySeconds: 60 },   // 1 min
  { body: { task: 'task2' }, delaySeconds: 300 },  // 5 min
  { body: { task: 'task3' }, delaySeconds: 600 },  // 10 min
]);

// Send batch with default delay
await env.MY_QUEUE.sendBatch(
  [
    { body: { task: 'task1' } },
    { body: { task: 'task2' } },
  ],
  { delaySeconds: 3600 } // All delayed by 1 hour
);

// Dynamic batch from array
const tasks = await getTasks();
await env.MY_QUEUE.sendBatch(
  tasks.map((task) => ({
    body: {
      taskId: task.id,
      userId: task.userId,
      priority: task.priority,
    },
  }))
);

限制：

• 每批最多 100 条消息
• 每批总大小最多 256 KB
• 每条消息的大小仍限制为 128 KB

---

完整的消费者 API

队列处理函数

export default {
  async queue(
    batch: MessageBatch,
    env: Env,
    ctx: ExecutionContext
  ): Promise<void> {
    // Process messages
  },
};

参数：

• batch - 包含消息的 MessageBatch 对象
• env - 环境绑定（KV、D1、R2 等）
• ctx - 执行上下文，可用于 waitUntil()、passThroughOnException()

---

MessageBatch 接口

interface MessageBatch<Body = unknown> {
  readonly queue: string;              // Queue name
  readonly messages: Message<Body>[];  // Array of messages
  ackAll(): void;                      // Acknowledge all messages
  retryAll(options?: QueueRetryOptions): void;  // Retry all messages
}

属性：

• queue - 此批次来源队列的名称

  • 在一个消费者处理多个队列时很有用

• messages - Message 对象数组

  • 排序尽力而为，不作保证
  • 不应假定消息的处理顺序

方法：

• ackAll() - 将所有消息标记为已成功投递

  • 即使处理函数抛出错误，这些消息也不会重试
  • 在安全处理完所有消息后使用

• retryAll(options?) - 将所有消息标记为需要重试

  • 消息会立即重新入队（也可以在延迟后入队）
  • 会计入 max_retries 限制

---

Message 接口

interface Message<Body = unknown> {
  readonly id: string;          // Unique message ID
  readonly timestamp: Date;     // When message was sent
  readonly body: Body;          // Message content
  readonly attempts: number;    // Retry count (starts at 1)
  ack(): void;                  // Acknowledge this message
  retry(options?: QueueRetryOptions): void;  // Retry this message
}

属性：

• id - 系统生成的唯一 ID（UUID）
• timestamp - 消息发送到队列时的 Date 对象
• body - 消息内容（任何可通过 JSON 序列化的类型）
• attempts - 消费者处理此消息的次数
  • 首次投递时从 1 开始
  • 每次重试时递增
  • 可用于指数退避：delaySeconds: 60 * message.attempts

方法：

• ack() - 将消息标记为已成功投递

  • 即使处理函数之后失败，消息也不会重试
  • 对非幂等操作至关重要（数据库写入、API 调用）

• retry(options?) - 将消息标记为需要重试

  • 立即重新入队，或在 delaySeconds 指定的延迟后重新入队
  • 会计入 max_retries（默认值为 3）

---

QueueRetryOptions

interface QueueRetryOptions {
  delaySeconds?: number;  // Delay retry (0-43200 seconds / 12 hours)
}

示例：

// Retry immediately
message.retry();

// Retry after 5 minutes
message.retry({ delaySeconds: 300 });

// Exponential backoff based on attempts
message.retry({
  delaySeconds: Math.min(60 * Math.pow(2, message.attempts - 1), 3600),
});

---

消费者模式

1. 基础消费者（隐式确认）

适用于： 重试安全的幂等操作

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      // Process message
      await sendEmail(message.body.email, message.body.content);
    }

    // Implicit ack: returning successfully acknowledges ALL messages
  },
};

行为：

• 函数成功返回 → 确认所有消息
• 函数抛出错误 → 重试所有消息
• 简单易用，但部分失败时可能导致重复处理

---

2. 显式确认（非幂等操作）

适用于： 数据库写入、API 调用、金融交易

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

        // Explicitly acknowledge success
        message.ack();
      } catch (error) {
        console.error(`Failed to process ${message.id}:`, error);

        // Don't ack - will retry (or let it fail)
        // Optionally: message.retry() for explicit retry
      }
    }
  },
};

为什么要显式确认？

• 如果批次中的一条消息失败，可避免重复写入
• 仅确认已成功处理的消息
• 失败的消息会单独重试

---

3. 指数退避重试

适用于： 有速率限制的 API、临时故障

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      try {
        // Call rate-limited API
        await fetch('https://api.example.com/process', {
          method: 'POST',
          body: JSON.stringify(message.body),
        });

        message.ack();
      } catch (error) {
        if (error.status === 429) {
          // Rate limited - retry with exponential backoff
          const delaySeconds = Math.min(
            60 * Math.pow(2, message.attempts - 1), // 60s, 120s, 240s, ...
            3600 // Max 1 hour
          );

          console.log(`Rate limited. Retrying in ${delaySeconds}s (attempt ${message.attempts})`);
          message.retry({ delaySeconds });
        } else {
          // Other error - retry immediately
          message.retry();
        }
      }
    }
  },
};

---

4. 死信队列（DLQ）模式

**最适合：**处理永久失败的消息

设置 DLQ：

# Create DLQ
npx wrangler queues create my-dlq

# Configure consumer with DLQ

wrangler.jsonc：

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_batch_size": 10,
        "max_retries": 3,
        "dead_letter_queue": "my-dlq"  // Failed messages go here
      }
    ]
  }
}

DLQ 消费者：

// Consumer for dead letter queue
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      // Log failed message
      console.error('PERMANENTLY FAILED MESSAGE:', {
        id: message.id,
        attempts: message.attempts,
        body: message.body,
        timestamp: message.timestamp,
      });

      // Store in database for manual review
      await env.DB.prepare(
        'INSERT INTO failed_messages (id, body, attempts, failed_at) VALUES (?, ?, ?, ?)'
      ).bind(
        message.id,
        JSON.stringify(message.body),
        message.attempts,
        new Date().toISOString()
      ).run();

      // Optionally: send alert to ops team
      await sendAlert({
        type: 'queue-dlq',
        messageId: message.id,
        queue: batch.queue,
      });

      // Acknowledge to remove from DLQ
      message.ack();
    }
  },
};

工作原理：

1. 消息在主队列中处理失败
2. 最多重试 max_retries 次（默认值为 3）
3. 达到最大重试次数后，消息会发送到 DLQ
4. DLQ 消费者处理失败的消息
5. 如果没有 DLQ，消息将被永久删除

---

5. 多个队列，单个消费者

**最适合：**集中处理逻辑

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    // Switch based on queue name
    switch (batch.queue) {
      case 'high-priority-queue':
        await processHighPriority(batch.messages, env);
        break;

      case 'low-priority-queue':
        await processLowPriority(batch.messages, env);
        break;

      case 'email-queue':
        await processEmails(batch.messages, env);
        break;

      default:
        console.warn(`Unknown queue: ${batch.queue}`);
        // Log to DLQ or monitoring
    }
  },
};

async function processHighPriority(messages: Message[], env: Env) {
  for (const message of messages) {
    // Process with urgency
    await fastProcess(message.body);
    message.ack();
  }
}

async function processLowPriority(messages: Message[], env: Env) {
  for (const message of messages) {
    // Can take longer
    await slowProcess(message.body);
    message.ack();
  }
}

wrangler.jsonc：

{
  "queues": {
    "consumers": [
      { "queue": "high-priority-queue" },
      { "queue": "low-priority-queue" },
      { "queue": "email-queue" }
    ]
  }
}

---

消费者配置

批处理设置

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_batch_size": 100,        // 1-100 messages (default: 10)
        "max_batch_timeout": 30       // 0-60 seconds (default: 5)
      }
    ]
  }
}

批处理工作原理：

• 满足以下任一条件时调用消费者：
  • 累积了 max_batch_size 条消息
  • 经过了 max_batch_timeout 秒（以先发生者为准）

示例：

• max_batch_size: 100，max_batch_timeout: 10
• 如果 3 秒内到达 100 条消息 → 立即发送批次
• 如果 10 秒内只到达 50 条消息 → 发送包含 50 条消息的批次

调优指南：

• 高流量、低延迟 → max_batch_size: 100，max_batch_timeout: 1
• 低流量、批量写入 → max_batch_size: 50，max_batch_timeout: 30
• 优化成本 → 批次越大，调用次数越少

---

重试设置

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_retries": 5,             // 0-100 (default: 3)
        "retry_delay": 300            // Seconds (default: 0)
      }
    ]
  }
}

max_retries：

• 重试失败消息的次数
• 达到最大重试次数后：
  • 配置了 DLQ → 消息发送到 DLQ
  • 未配置 DLQ → 消息将被永久删除

retry_delay：

• 所有重试消息的默认延迟时间（秒）
• 可通过 message.retry({ delaySeconds }) 覆盖
• 最大延迟时间：43200 秒（12 小时）

---

并发设置

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_concurrency": 10         // 1-250 (default: auto-scale)
      }
    ]
  }
}

并发工作原理：

• 队列会根据积压自动扩缩消费者
• 默认值：最多扩缩至 250 个并发调用
• 设置 max_concurrency 可限制扩缩

何时设置 max_concurrency：

• ✅ 上游 API 有速率限制
• ✅ 数据库连接数受限
• ✅ 需要控制成本
• ❌ 大多数情况下，无需设置即可获得最佳性能

自动扩缩触发条件：

• 积压增加（消息不断累积）
• 错误率较高
• 处理速度与消息到达速率

---

死信队列

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_retries": 3,
        "dead_letter_queue": "my-dlq"  // Name of DLQ
      }
    ]
  }
}

重要：

• 必须单独创建 DLQ：npx wrangler queues create my-dlq
• 如果没有 DLQ，失败的消息将被永久删除
• 如果没有消费者，DLQ 中的消息会保留 4 天
• 生产环境队列务必配置 DLQ

---

Wrangler 命令

创建队列

# Create queue with defaults
npx wrangler queues create my-queue

# Create with custom retention (4 days default, max 14 days)
npx wrangler queues create my-queue --message-retention-period-secs 1209600

# Create with delivery delay
npx wrangler queues create my-queue --delivery-delay-secs 60

---

列出队列

npx wrangler queues list

---

获取队列信息

npx wrangler queues info my-queue

# Output shows:
# - Queue name
# - Message retention period
# - Delivery delay
# - Consumers
# - Backlog size

---

更新队列

# Update message retention (60 seconds to 14 days)
npx wrangler queues update my-queue --message-retention-period-secs 604800

# Update delivery delay (0 to 12 hours)
npx wrangler queues update my-queue --delivery-delay-secs 3600

---

删除队列

npx wrangler queues delete my-queue

# Warning: This deletes ALL messages in the queue!
# Use with caution in production

---

消费者管理

# Add consumer to queue
npx wrangler queues consumer add my-queue my-consumer-worker \
  --batch-size 50 \
  --batch-timeout 10 \
  --message-retries 5 \
  --max-concurrency 20 \
  --retry-delay-secs 300

# Remove consumer from queue
npx wrangler queues consumer remove my-queue my-consumer-worker

---

清空队列

# Delete ALL messages in queue (use with caution!)
npx wrangler queues purge my-queue

---

暂停/恢复投递

# Pause message delivery to consumers
npx wrangler queues pause-delivery my-queue

# Resume message delivery
npx wrangler queues resume-delivery my-queue

使用场景：

• 维护消费者 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 速率限制	1200 次请求 / 5 分钟

---

定价

需要 Workers Paid 计划（$5/月）

操作定价：

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

计为一次操作的情况：

• 每个写入、读取或删除的 64 KB 数据块
• 超过 64 KB 的消息会计为多次操作：
  • 65 KB 消息 = 2 次操作
  • 127 KB 消息 = 2 次操作
  • 128 KB 消息 = 2 次操作

典型消息生命周期：

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

重试：

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

死信队列：

• 写入 DLQ = 额外的写入操作

成本示例：

• 每月 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

---

TypeScript 类型

// Queue binding (producer)
interface Queue<Body = any> {
  send(body: Body, options?: QueueSendOptions): Promise<void>;
  sendBatch(
    messages: Iterable<MessageSendRequest<Body>>,
    options?: QueueSendBatchOptions
  ): Promise<void>;
}

interface QueueSendOptions {
  delaySeconds?: number;
}

interface MessageSendRequest<Body = any> {
  body: Body;
  delaySeconds?: number;
}

interface QueueSendBatchOptions {
  delaySeconds?: number;
}

// Consumer handler
export default {
  queue(
    batch: MessageBatch,
    env: Env,
    ctx: ExecutionContext
  ): Promise<void>;
}

interface MessageBatch<Body = unknown> {
  readonly queue: string;
  readonly messages: Message<Body>[];
  ackAll(): void;
  retryAll(options?: QueueRetryOptions): void;
}

interface Message<Body = unknown> {
  readonly id: string;
  readonly timestamp: Date;
  readonly body: Body;
  readonly attempts: number;
  ack(): void;
  retry(options?: QueueRetryOptions): void;
}

interface QueueRetryOptions {
  delaySeconds?: number;
}

---

错误处理

常见错误

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

---

务必做到 ✅

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()

---

故障排除

问题：消息未投递给消费者

可能原因：

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
    }]
  }
}

---

生产环境检查清单

部署到生产环境之前：

• 已创建并配置死信队列
• 对非幂等操作使用了显式 ack()
• 已验证消息大小（<128 KB）
• 已根据工作负载优化批处理参数
• 已配置重试参数（max_retries、retry_delay）
• 消费者并发设置适当
• 已实现错误处理和日志记录
• 已设置监控和警报
• DLQ 消费者已部署
• 对受速率限制的 API 使用指数退避
• 必要时已提高 CPU 限制（limits.cpu_ms）
• 消息保留期限适当
• 已进行负载测试

---

Related Documentation

• Cloudflare Queues Docs
• How Queues Works
• JavaScript APIs
• Batching & Retries
• Consumer Concurrency
• Dead Letter Queues
• Wrangler Commands
• Limits
• Pricing

---

Last Updated: 2025-10-21 Version: 1.0.0 Maintainer: Jeremy Dawes | [email protected]