Token导航 LogoToken导航

Agent Skill · 前端设计

cloudflare-queues(Cloudflare 队列)

用于辅助前端页面、组件、样式和交互逻辑的开发与维护。它适合让 Agent 生成或审查 React、Next.js、Vue、Tailwind、CSS 等相关代码,整理组件结构,或定位布局和性能问题。使用时需要结合项目现有设计系统、路由和构建方式,避免只生成孤立片段;涉及页面改动时,应配合本地预览和构建检查确认视觉效果。

cloudflare-queues 是前端设计类 Agent Skill,可用宿主为 Codex、Claude、Cursor、Gemini CLI,通过GitHub 仓库获取,权限为需要联网;许可证为 MIT,第三方扫描已通过,按安装量在前端设计类的 10,216 个 Skill 里排第 5,300。

总安装 699周安装 28GitHub Stars 14下载 226核验 2026-05-01

安装说明

本站只整理中文说明,不托管安装包,也不代用户安装。
GitHub
许可证
MIT

安装方式

通过对话安装

复制提示词发给支持本地命令或 Skills 的 AI 助手,先确认命令和权限,再让它执行。

请帮我安装这个 Agent Skill:cloudflare-queues(Cloudflare 队列)
来源仓库:https://github.com/jackspace/claudeskillz
仓库路径:skills/cloudflare-queues
安装命令:
npx skills add https://github.com/jackspace/claudeskillz --skill cloudflare-queues
安装前请先检查当前环境是否支持对应 CLI,并向我确认将要执行的命令、安装目录、联网范围和文件读写权限;确认后再执行。

命令行安装

复制命令到本机终端执行。不同来源提供的安装方式可能略有差异;本站展示可直接复制的安装命令,安装前请核对来源页面。

skills.shnpx skills
npx skills add https://github.com/jackspace/claudeskillz --skill cloudflare-queues

这个 Skill 是做什么的

前端设计类 Skill 关注页面结构、视觉规范和组件写法,输出通常是可运行的界面代码。

创建和管理 Cloudflare Queues,实现消息队列与异步任务处理。支持生产者发送消息、消费者批量处理,适用于事件驱动架构。通过 Wrangler 命令行工具快速创建队列并配置触发器规则。需在 wrangler.jsonc 中声明 queues 字段并指定目标服务名称。

生产环境应设置重试策略与死信队列,防止消息丢失或重复消费。

来源仓库内路径:skills/cloudflare-queues

上手步骤

4 步
  1. 1

    确认自己的宿主工具

    公开资料标注支持 Codex、Claude、Cursor、Gemini CLI;先确认自己用的是其中之一,再按对应工具的 Skills 目录放置。

  2. 2

    拿到安装命令

    可以直接复制的命令是 npx skills add https://github.com/jackspace/claudeskillz --skill cloudflare-queu…;命令里的占位符换成自己的路径或密钥再执行。

  3. 3

    安装后重启宿主

    命令执行完要完全退出宿主工具再打开,让新装的 Skill 被重新扫描到,否则列表里可能看不到。

  4. 4

    按最小权限试用

    它需要需要联网,先在测试目录或测试账号里跑,确认没问题再放到正式项目。

常见问题

6 条

cloudflare-queues 怎么安装?

在终端执行 npx skills add https://github.com/jackspace/claudeskillz --skill cloudflare-queues。装完完全退出宿主工具再打开,让它重新扫描本地 Skill。

cloudflare-queues 支持哪些 AI 工具?

公开资料标注支持 Codex、Claude、Cursor、Gemini CLI。其他工具只要能读取本地 Skill 目录,一般也能手动放进去,但要自己确认触发方式是否一致。

cloudflare-queues 需要什么权限?

权限级别是需要联网。该 Skill 可能需要联网访问来源站点、仓库或外部 API;具体网络访问范围需要结合源码和 README 复核。

cloudflare-queues 是免费的吗?

来源标注的许可证是 MIT。本站不代收任何费用,也不托管安装包;能否商用、要不要署名,以许可证原文为准。

cloudflare-queues 用的人多吗?

公开数据是累计安装 699 次,近一周安装 28 次,下载 226 次,GitHub 14 星。这些数字来自第三方目录统计,只能当作热度参考,不代表质量或安全性。

cloudflare-queues 安全吗?

第三方扫描结果为通过,但这只覆盖已知的静态检查项,不能替代自己读一遍源码和安装命令。

值得选的地方和要先想清楚的地方

值得选的地方

  • 同时适配 4 种工具,换客户端不用重新找替代品。
  • 3 个目录同时收录,说明这个 Skill 流传比较广。
  • 许可证明确为 MIT,商用和分发边界好判断。
  • 第三方扫描已通过,常见的高风险行为没有命中。

先想清楚的地方

  • 第三方目录整理,版本更新和行为细节以来源仓库为准。

适不适合你

更适合

  • 已经在用 Codex,想直接扩展它能力的人。
  • 需要处理前端设计类任务、又不想每次都手写提示词的人。
  • 需要许可证清楚、并且有第三方扫描结果的团队项目。

不太适合

  • 需要厂商级技术支持和 SLA 的生产系统。

前端设计类 Skill 对比

6 个同类
Skill总安装GitHub 星权限许可证扫描
frontend-design

前端设计 · Codex、Claude

8,823,360126,072只读MIT通过
vercel-composition-patterns

Vercel 组合模式 · Codex、Claude

3,858,81625,893需要联网MIT通过
vercel-react-native-skills

Vercel React native skills 前端 · Codex、Claude

2,468,66425,878只读MIT通过
adapt

adapt 搜索 · Codex、Claude

2,027,04023,125操作浏览器未明确通过
bolder

bolder 分析 · Codex、Claude

2,004,28823,130只读未明确通过
cloudflare-queues

Cloudflare 队列 · 本页

69914需要联网MIT通过

按安装量取同分类前几名;数字来自第三方目录统计,只作热度参考,不代表质量结论。

SKILL.md 原文

Cloudflare Queues

Status: Production Ready ✅ Last Updated: 2025-10-21 Dependencies: cloudflare-worker-base (for Worker setup) Latest Versions: wrangler@4.43.0, @cloudflare/workers-types@4.20251014.0


Quick Start (10 Minutes)

1. Create a Queue

# 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. Set Up Producer (Send Messages)

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

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. Set Up Consumer (Process Messages)

Create consumer 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 (Consumer):

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 and Test

# 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

Complete Producer API

send() - Send Single Message

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

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

Examples:

// 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: 'user@example.com',
  items: [{ sku: 'ITEM-1', quantity: 2 }],
  total: 49.99,
  timestamp: Date.now(),
});

CRITICAL:

  • Message body must be JSON serializable (structured clone algorithm)
  • Maximum message size: 128 KB (including ~100 bytes internal metadata)
  • Messages >128 KB will fail - split them or store in R2 and send reference

sendBatch() - Send Multiple Messages

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

Examples:

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

Limits:

  • Maximum 100 messages per batch
  • Maximum 256 KB total batch size
  • Each message still limited to 128 KB individually

Complete Consumer API

Queue Handler Function

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

Parameters:

  • batch - MessageBatch object containing messages
  • env - Environment bindings (KV, D1, R2, etc.)
  • ctx - Execution context for waitUntil(), passThroughOnException()

MessageBatch Interface

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
}

Properties:

  • queue - Name of the queue this batch came from

- Useful when one consumer handles multiple queues

  • messages - Array of Message objects

- Ordering is best effort, not guaranteed - Process order should not be assumed

Methods:

  • ackAll() - Mark all messages as successfully delivered

- Even if handler throws error, these messages won't retry - Use when you've safely processed all messages

  • retryAll(options?) - Mark all messages for retry

- Messages re-queued immediately (or after delay) - Counts towards max_retries limit


Message Interface

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
}

Properties:

  • id - System-generated unique ID (UUID)
  • timestamp - Date object when message was sent to queue
  • body - Your message content (any JSON serializable type)
  • attempts - Number of times consumer has processed this message

- Starts at 1 on first delivery - Increments on each retry - Use for exponential backoff: delaySeconds: 60 * message.attempts

Methods:

  • ack() - Mark message as successfully delivered

- Message won't be retried even if handler fails later - Critical for non-idempotent operations (DB writes, API calls)

  • retry(options?) - Mark message for retry

- Re-queued immediately or after delaySeconds - Counts towards max_retries (default 3)


QueueRetryOptions

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

Example:

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

Consumer Patterns

1. Basic Consumer (Implicit Acknowledgement)

Best for: Idempotent operations where retries are safe

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

Behavior:

  • If function returns successfully → all messages acknowledged
  • If function throws error → all messages retried
  • Simple but can cause duplicate processing on partial failures

2. Explicit Acknowledgement (Non-Idempotent Operations)

Best for: Database writes, API calls, financial transactions

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

Why explicit ack?

  • Prevents duplicate writes if one message in batch fails
  • Only successfully processed messages are acknowledged
  • Failed messages retry independently

3. Retry with Exponential Backoff

Best for: Rate-limited APIs, temporary failures

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. Dead Letter Queue (DLQ) Pattern

Best for: Handling permanently failed messages

Setup 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:

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

How it works:

  1. Message fails in main queue
  2. Retries up to max_retries (default 3)
  3. After max retries, sent to DLQ
  4. DLQ consumer processes failed messages
  5. Without DLQ, messages are deleted permanently

5. Multiple Queues, Single Consumer

Best for: Centralized processing logic

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

Consumer Configuration

Batch Settings

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

How batching works:

  • Consumer called when either condition met:

- max_batch_size messages accumulated - max_batch_timeout seconds elapsed (whichever comes first)

Example:

  • max_batch_size: 100, max_batch_timeout: 10
  • If 100 messages arrive in 3 seconds → batch delivered immediately
  • If only 50 messages arrive in 10 seconds → batch of 50 delivered

Tuning guidelines:

  • High volume, low latency → max_batch_size: 100, max_batch_timeout: 1
  • Low volume, batch writes → max_batch_size: 50, max_batch_timeout: 30
  • Cost optimization → Larger batches = fewer invocations

Retry Settings

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

max_retries:

  • Number of times to retry failed message
  • After max retries:

- With DLQ → message sent to DLQ - Without DLQ → message deleted permanently

retry_delay:

  • Default delay for all retried messages (seconds)
  • Can be overridden with message.retry({delaySeconds})
  • Maximum delay: 43200 seconds (12 hours)

Concurrency Settings

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

How concurrency works:

  • Queues auto-scales consumers based on backlog
  • Default: Scales up to 250 concurrent invocations
  • Setting max_concurrency limits scaling

When to set max_concurrency:

  • ✅ Upstream API has rate limits
  • ✅ Database connection limits
  • ✅ Want to control costs
  • ❌ Most cases - leave unset for best performance

Auto-scaling triggers:

  • Growing backlog (messages accumulating)
  • High error rate
  • Processing speed vs. incoming rate

Dead Letter Queue

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

CRITICAL:

  • DLQ must be created separately: npx wrangler queues create my-dlq
  • Without DLQ, failed messages are deleted permanently
  • Messages in DLQ persist for 4 days without consumer
  • Always configure DLQ for production queues

Wrangler Commands

Create Queue

# 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

List Queues

npx wrangler queues list

Get Queue Info

npx wrangler queues info my-queue

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

Update Queue

# 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

Delete Queue

npx wrangler queues delete my-queue

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

Consumer Management

# 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

Purge Queue

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

Pause/Resume Delivery

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

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

Use cases:

  • Maintenance on consumer Workers
  • Temporarily stop processing
  • Debug issues without losing messages

Limits & Quotas

FeatureLimit
Queues per account10,000
Message size128 KB (includes ~100 bytes metadata)
Message retries100 max
Batch size1-100 messages
Batch timeout0-60 seconds
Messages per sendBatch100 (or 256 KB total)
Queue throughput5,000 messages/second per queue
Message retention4 days (default), 14 days (max)
Queue backlog size25 GB per queue
Concurrent consumers250 (push-based, auto-scale)
Consumer duration15 minutes (wall clock)
Consumer CPU time30 seconds (default), 5 minutes (max)
Visibility timeout12 hours (pull consumers)
Message delay12 hours (max)
API rate limit1200 requests / 5 minutes

Pricing

Requires Workers Paid plan ($5/month)

Operations Pricing:

  • First 1,000,000 operations/month: FREE
  • After that: $0.40 per million operations

What counts as an operation:

  • Each 64 KB chunk written, read, or deleted
  • Messages >64 KB count as multiple operations:

- 65 KB message = 2 operations - 127 KB message = 2 operations - 128 KB message = 2 operations

Typical message lifecycle:

  • 1 write + 1 read + 1 delete = 3 operations

Retries:

  • Each retry = additional read operation
  • Message retried 3 times = 1 write + 4 reads + 1 delete = 6 operations

Dead Letter Queue:

  • Writing to DLQ = additional write operation

Cost examples:

  • 1M messages/month (no retries): ((1M × 3) - 1M) / 1M × $0.40 = $0.80
  • 10M messages/month: ((10M × 3) - 1M) / 1M × $0.40 = $11.60
  • 100M messages/month: ((100M × 3) - 1M) / 1M × $0.40 = $119.60

TypeScript Types

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

Error Handling

Common Errors

1. Message Too Large
// ❌ 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. Throughput Exceeded
// ❌ 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. Consumer Timeout
// ❌ 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. Backlog Growing
// 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();
      })
    );
  },
};

Always Do ✅

  1. Configure Dead Letter Queue for production queues
  2. Use explicit ack() for non-idempotent operations (DB writes, API calls)
  3. Validate message size before sending (<128 KB)
  4. Use sendBatch() for multiple messages (more efficient)
  5. Implement exponential backoff for retries
  6. Set appropriate batch settings based on workload
  7. Monitor queue backlog and consumer errors
  8. Use ctx.waitUntil() for async cleanup in consumers
  9. Handle errors gracefully - log, alert, retry
  10. Let concurrency auto-scale (don't set max_concurrency unless needed)

Never Do ❌

  1. Never assume message ordering - not guaranteed FIFO
  2. Never rely on implicit ack for non-idempotent ops - use explicit ack()
  3. Never send messages >128 KB - will fail
  4. Never delete queues with active messages - data loss
  5. Never skip DLQ configuration in production
  6. Never exceed 5000 msg/s per queue - rate limit error
  7. Never process messages synchronously in loop - use Promise.all()
  8. Never ignore message.attempts - use for backoff logic
  9. Never set max_concurrency=1 unless you have a very specific reason
  10. Never forget to ack() in explicit acknowledgement patterns

Troubleshooting

Issue: Messages not being delivered to consumer

Possible causes:

  1. Consumer not deployed
  2. Wrong queue name in wrangler.jsonc
  3. Delivery paused
  4. Consumer throwing errors

Solution:

# 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

Issue: Entire batch retried when one message fails

Cause: Using implicit acknowledgement with non-idempotent operations

Solution: Use explicit 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
  }
}

Issue: Messages deleted without processing

Cause: No Dead Letter Queue configured

Solution:

# Create DLQ
npx wrangler queues create my-dlq

# Add to consumer config
{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "dead_letter_queue": "my-dlq"
    }]
  }
}

Issue: Consumer not auto-scaling

Possible causes:

  1. max_concurrency set to 1
  2. Consumer returning errors (not processing)
  3. Batch processing too fast (no backlog)

Solution:

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

Production Checklist

Before deploying to production:

  • Dead Letter Queue created and configured
  • Explicit ack() used for non-idempotent operations
  • Message size validation (<128 KB)
  • Batch settings optimized for workload
  • Retry settings configured (max_retries, retry_delay)
  • Consumer concurrency settings appropriate
  • Error handling and logging implemented
  • Monitoring and alerting set up
  • DLQ consumer deployed
  • Exponential backoff for rate-limited APIs
  • CPU limits increased if needed (limits.cpu_ms)
  • Message retention period appropriate
  • Tested under load

Related Documentation


Last Updated: 2025-10-21 Version: 1.0.0 Maintainer: Jeremy Dawes | jeremy@jezweb.net

平台分布

Claude Code

29.13%
按下载量换算66

Codex

27.05%
按下载量换算61

Antigravity

17.46%
按下载量换算39

Gemini CLI

12.33%
按下载量换算28

windsurf

7.73%
按下载量换算17

OpenCode

3.72%
按下载量换算8

安全审计

Gen Agent Trust Hub

通过

Socket

通过

Snyk

通过

权限和风险

需要联网

该 Skill 可能需要联网访问来源站点、仓库或外部 API;具体网络访问范围需要结合源码和 README 复核。

安装前确认

本站仅展示第三方公开信息,不托管安装包,不提供自动安装或运行环境。安装前应自行审查源码、依赖和命令行为。

继续浏览同类 Skills