cloudflare-queues（Cloudflare 队列）

Cloudflare Queues

Queues is a message queue for Workers. Supports push (Worker consumer) and pull (HTTP API) patterns. At-least-once delivery.

Quick Start

Create queue

npx wrangler queues create my-queue

Producer binding

// wrangler.jsonc
{
  "queues": {
    "producers": [
      {
        "queue": "my-queue",
        "binding": "MY_QUEUE"
      }
    ]
  }
}

Consumer binding

// wrangler.jsonc
{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_batch_size": 10,
        "max_batch_timeout": 5
      }
    ]
  }
}

Producer Worker

export interface Env {
  MY_QUEUE: Queue;
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    await env.MY_QUEUE.send({ url: request.url, method: request.method });
    return new Response("Message sent");
  },
};

Consumer Worker

export interface Env {}

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const msg of batch.messages) {
      console.log(msg.body);
      msg.ack();
    }
  },
};

Producer API

send(body, options?)

await env.MY_QUEUE.send({ action: "process", id: 123 });

// With delay
await env.MY_QUEUE.send(message, { delaySeconds: 600 }); // 10 min delay

// With content type
await env.MY_QUEUE.send(message, { contentType: "json" });

sendBatch(messages, options?)

await env.MY_QUEUE.sendBatch([{ body: { id: 1 } }, { body: { id: 2 }, options: { delaySeconds: 300 } }, { body: { id: 3 } }]);

// Global delay for batch
await env.MY_QUEUE.sendBatch(messages, { delaySeconds: 600 });

Limits:

• Max 100 messages per batch
• Max 128 KB per message
• Total batch ≤ 256 KB

Content Types

Note: Pull consumers cannot decode v8 content type.

See api.md for type definitions.

Consumer API

MessageBatch

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

Message

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

Acknowledgment Patterns

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const msg of batch.messages) {
      try {
        await processMessage(msg.body);
        msg.ack(); // Explicit success
      } catch (error) {
        msg.retry({ delaySeconds: 60 }); // Retry with delay
      }
    }
  },
};

Batch-level operations

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    try {
      await processAll(batch.messages);
      batch.ackAll(); // All succeeded
    } catch (error) {
      batch.retryAll({ delaySeconds: 300 }); // Retry all
    }
  },
};

Precedence: Per-message calls override batch-level.

Consumer Configuration

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_batch_size": 10, // 1-100, default 10
        "max_batch_timeout": 5, // 0-60 seconds, default 5
        "max_retries": 3, // default 3
        "max_concurrency": 10, // default: auto-scale
        "dead_letter_queue": "dlq", // optional DLQ
        "retry_delay": 60 // default retry delay (seconds)
      }
    ]
  }
}

See consumer.md for details.

Dead Letter Queues

Messages that fail after max_retries go to DLQ.

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

Create DLQ:

npx wrangler queues create my-dlq

DLQ retention: 4 days without consumer.

Process DLQ:

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

Delivery Delay

On send

await env.MY_QUEUE.send(message, { delaySeconds: 600 }); // 10 min

On retry

msg.retry({ delaySeconds: 3600 }); // 1 hour

Queue-level default

npx wrangler queues create my-queue --delivery-delay-secs=300

Exponential backoff

const backoff = (attempts: number, base = 10) => base ** attempts;

msg.retry({ delaySeconds: Math.min(backoff(msg.attempts), 43200) });

Maximum delay: 12 hours (43200 seconds).

Concurrency

Consumers auto-scale based on backlog. Set max:

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_concurrency": 5
      }
    ]
  }
}

max_concurrency: 1 = sequential processing.

Scaling factors:

• Backlog size and growth
• Success/failure ratio
• max_concurrency limit

Note: retry() calls don't count as failures for scaling.

Pull Consumers (HTTP API)

For consuming outside Workers.

Enable pull consumer

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "type": "http_pull",
        "visibility_timeout_ms": 5000,
        "max_retries": 5
      }
    ]
  }
}

Pull messages

curl -X POST "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/queues/$QUEUE_ID/messages/pull" \
  -H "Authorization: Bearer $API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"batch_size": 10, "visibility_timeout_ms": 30000}'

Acknowledge messages

curl -X POST "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/queues/$QUEUE_ID/messages/ack" \
  -H "Authorization: Bearer $API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "acks": [{"lease_id": "..."}],
    "retries": [{"lease_id": "...", "delay_seconds": 60}]
  }'

See pull-consumer.md for details.

Wrangler Commands

# Queue management
wrangler queues create <name> [--delivery-delay-secs=N]
wrangler queues delete <name>
wrangler queues list
wrangler queues info <name>

# Pause/resume
wrangler queues pause-delivery <name>
wrangler queues resume-delivery <name>

# Purge all messages
wrangler queues purge <name>

# Consumer management
wrangler queues consumer add <queue> <script> [options]
wrangler queues consumer remove <queue> <script>
wrangler queues consumer http add <queue> [options]
wrangler queues consumer http remove <queue>

Limits

Increase CPU limit

{
  "limits": {
    "cpu_ms": 300000 // 5 minutes
  }
}

Pricing

Workers Paid: 1M operations/month included, then $0.40/million.

Operation = 64 KB chunk written, read, or deleted.

Formula: (Messages × 3 - 1M) / 1M × $0.40

No egress fees.

See pricing.md for examples.

Delivery Guarantees

At-least-once delivery: Messages delivered at least once, possibly duplicated.

Handle duplicates:

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const msg of batch.messages) {
      const key = `processed:${msg.id}`;
      if (await env.KV.get(key)) {
        msg.ack(); // Already processed
        continue;
      }
      await processMessage(msg.body);
      await env.KV.put(key, "1", { expirationTtl: 86400 });
      msg.ack();
    }
  },
};

Event Notifications

R2 and other services can send events to Queues.

# R2 → Queue
wrangler r2 bucket notification create my-bucket \
  --event-type object-create \
  --queue my-queue

See cloudflare-r2 skill for event notification setup.

Prohibitions

• ❌ Do not use v8 content type with pull consumers
• ❌ Do not exceed 128 KB per message
• ❌ Do not rely on exactly-once delivery (use idempotency)
• ❌ Do not ignore DLQ — process failed messages
• ❌ Do not set excessive concurrency without testing

References

• api.md — Producer/Consumer API reference
• consumer.md — Consumer configuration
• pull-consumer.md — HTTP pull API
• pricing.md — Billing details

Links

• Documentation
• Changelog

Related Skills

• cloudflare-workers — Worker development
• cloudflare-r2 — R2 event notifications
• cloudflare-durable-objects — Queue producer from DO
• cloudflare-kv — Idempotency tracking