Workers Queues -- Async Message Processing

In this tutorial, you will learn how to use Cloudflare Workers Queues to decouple message producers from consumers for reliable asynchronous processing at the edge. Message Queues are important because they prevent data loss when downstream services are unavailable, smooth traffic spikes by buffering requests, and enable parallel processing of independent tasks. A real-world example is an image processing pipeline where uploads are queued for resizing, thumbnail generation, and virus scanning without blocking the user's upload request.

How Workers Queues Work

Workers Queues is a fully managed message queue service integrated into the Workers runtime. Producers send messages to a queue using the env.QUEUE.send() method. The queue persists messages until a consumer Worker processes them. Consumers are defined by exporting a queue handler that receives batches of messages. Messages are delivered at least once, and failed messages can be sent to a dead-letter queue (DLQ) after a configurable number of retries.

Queue Architecture

flowchart LR
    P[Producer Worker] -->|env.QUEUE.send msg| Q[Workers Queue]
    Q -->|Batch delivery| C[Consumer Worker queue handler]
    C -->|Success| OK[Acknowledged]
    C -->|Failure after retries| DLQ[Dead Letter Queue]
    DLQ -->|Manual reprocess| C

    style Q fill:#f90,color:#fff
    style C fill:#3498db,color:#fff
    style DLQ fill:#e74c3c,color:#fff

Messages are stored redundantly and are available for consumption within seconds of being sent. Consumers pull messages in batches for efficient processing. Each message can be retried up to the configured maximum number of times before moving to the DLQ.

Creating a Queue and Producer

// wrangler.toml
// [[queues.producers]]
// queue = "image-queue"
// binding = "IMAGE_QUEUE"
//
// [[queues.consumers]]
// queue = "image-queue"
// max_batch_size = 10
// max_batch_timeout = 5

// producer.js -- sends messages to the queue
export default {
  async fetch(request, env) {
    const { imageUrl, userId } = await request.json();

    // Send message to queue for async processing
    await env.IMAGE_QUEUE.send({
      imageUrl,
      userId,
      timestamp: Date.now()
    });

    return new Response(JSON.stringify({
      status: 'queued',
      imageUrl
    }), {
      status: 202,
      headers: { 'Content-Type': 'application/json' }
    });
  }
};

// Request body: {"imageUrl": "https://example.com/photo.jpg", "userId": "user-42"}
// Expected response:
// {"status": "queued", "imageUrl": "https://example.com/photo.jpg"}

The send() method accepts any serializable JavaScript object. Messages are queued immediately and the response is returned without waiting for processing.

Building a Consumer

// consumer.js -- processes messages from the queue
export default {
  async queue(batch, env) {
    for (const message of batch.messages) {
      const { imageUrl, userId, timestamp } = message.body;

      try {
        // Download and process the image
        const response = await fetch(imageUrl);
        const imageData = await response.arrayBuffer();

        // Store processed versions in R2
        await env.ASSETS.put(
          `processed/${userId}/${Date.now()}-original.jpg`,
          imageData
        );

        // Acknowledge successful processing
        message.ack();
      } catch (error) {
        console.error(`Failed to process ${imageUrl}: ${error.message}`);
        // Retry or send to DLQ after 3 attempts
        message.retry({ maxRetries: 3 });
      }
    }
  }
};

// Expected log output on success:
// (no output -- message acknowledged quietly)
// Expected log output on failure:
// Failed to process https://example.com/photo.jpg: Fetch failed

Each message must be explicitly acknowledged with ack() or scheduled for retry with retry(). Unacknowledged messages are retried automatically. The consumer runs as a dedicated Worker bound to the queue.

Batching for Efficiency

// Consumer with batch processing and progress tracking
export default {
  async queue(batch, env) {
    const results = { processed: 0, failed: 0, total: batch.messages.length };

    // Process all messages in the batch concurrently
    await Promise.allSettled(
      batch.messages.map(async (message) => {
        try {
          const { email, subject, body } = message.body;

          // Send email via external API
          await fetch('https://mail-api.example.com/send', {
            method: 'POST',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify({ to: email, subject, body })
          });

          message.ack();
          results.processed++;
        } catch (error) {
          message.retry({ maxRetries: 3, delay: 60 });
          results.failed++;
        }
      })
    );

    // Store batch processing report
    await env.DB.prepare(
      'INSERT INTO batch_logs (batch_id, processed, failed, total, created_at) VALUES (?1, ?2, ?3, ?4, datetime("now"))'
    ).bind(
      batch.batch.id,
      results.processed,
      results.failed,
      results.total
    ).run();

    console.log(`Batch ${batch.batch.id}: ${results.processed}/${results.total} processed`);
  }
};

// Expected log:
// Batch abc-123: 8/10 processed

Batching reduces overhead by processing multiple messages per invocation. The batch.batch.id uniquely identifies each batch for tracing. Use Promise.allSettled to avoid failing the entire batch when one message fails.

Using a Dead Letter Queue

// wrangler.toml
// [[queues.producers]]
// queue = "email-queue"
// binding = "EMAIL_QUEUE"
//
// [[queues.consumers]]
// queue = "email-queue"
// dead_letter_queue = "email-dlq"
// max_retries = 3
//
// [[queues.consumers]]
// queue = "email-dlq"
// binding = "EMAIL_DLQ"

// DLQ consumer -- handles permanently failed messages
export default {
  async queue(batch, env) {
    for (const message of batch.messages) {
      const { email, subject, error } = message.body;

      // Log permanently failed messages for manual inspection
      await env.DB.prepare(
        'INSERT INTO failed_emails (email, subject, error, failed_at) VALUES (?1, ?2, ?3, datetime("now"))'
      ).bind(email, subject, error || 'unknown').run();

      // Notify admin
      await env.ADMIN_QUEUE.send({
        type: 'alert',
        message: `Email to ${email} failed permanently`
      });

      message.ack();
    }
  }
};

// Expected outcome:
// Permanently failed messages are logged to D1 and an admin alert is queued

The dead-letter queue receives messages that exceeded the maximum retry count. A separate consumer processes the DLQ for manual review, logging, or alternative handling.

Common Errors and Troubleshooting

Message Too Large

Queues messages are limited to 128 KB. Larger payloads should store data in R2 or KV and pass only the key reference in the queue message.

Consumer Not Processing

If a consumer Worker is not deployed or not bound to the queue, messages accumulate in the queue without being processed. Verify the consumer binding in wrangler.toml and ensure the Worker is deployed.

Retry Loop

A consumer that fails to Process a message and always calls retry() without changing conditions can create an infinite loop. Implement exponential backoff or move to DLQ after a fixed number of retries.

Batch Timeout

Consumers have 30 seconds (free) or 15 minutes (paid) to Process a batch. If the batch timeout is exceeded, unacknowledged messages are retried. Keep batch sizes small enough to Process within the timeout.

Queue Deleted

Deleting a queue that still has producer or consumer bindings causes Workers to throw binding errors at runtime. Remove bindings before deleting queues.

Practice Questions

1. How does a consumer Worker acknowledge successful message processing?
2. What is the purpose of a dead-letter queue in Workers Queues?
3. What is the maximum message size supported by Workers Queues?

FAQ

{{< faq "Can I use Workers Queues with services outside of Cloudflare?">}} Yes. While producers and consumers are typically Workers, you can send messages to a queue from any HTTP client using the Cloudflare API. External services can also poll queue metrics via the API for monitoring purposes.{{< /faq >}}

Summary

Workers Queues provides reliable asynchronous message processing integrated directly into the Workers runtime. Producers send messages with send(), consumers Process batches in the queue handler, and messages are retried automatically on failure. Dead-letter queues capture permanently failed messages for manual review. Use Workers Queues for image processing pipelines, email dispatch, data synchronization, and any workload that benefits from decoupled, resilient processing at the edge.

This guide is brought to you by the developers of Cloudflare, Cloudflare Workers, and Durga Antivirus Pro at DodaTech.