RabbitMQ Guide: Message Queues, Exchanges, Routing, and Clustering in 2026

Master RabbitMQ with this comprehensive guide covering AMQP concepts, exchange types (direct, fanout, topic, headers), message acknowledgment, persistence, dead letter queues, retry patterns, publisher confirms, consumer prefetch, Node.js and Python client examples, Docker Compose setup, clustering, monitoring, and a RabbitMQ vs Kafka vs SQS vs Redis comparison.

What Is RabbitMQ and When to Use It

RabbitMQ is an open-source message broker originally developed by Rabbit Technologies and now maintained by Broadcom. It implements the Advanced Message Queuing Protocol (AMQP 0-9-1) and supports additional protocols including MQTT, STOMP, and AMQP 1.0 via plugins. RabbitMQ acts as a middleman between applications that produce messages and applications that consume them, enabling asynchronous, decoupled communication.

RabbitMQ is deployed at companies like Bloomberg, Goldman Sachs, and Instagram. It runs on the Erlang/OTP platform, which gives it exceptional reliability, concurrency, and fault tolerance. A single RabbitMQ node can handle tens of thousands of messages per second; a properly configured cluster can handle hundreds of thousands.

Use RabbitMQ when you need:

• Task queues — distribute background jobs (email sending, image processing, PDF generation) across worker processes.
• Event-driven architecture — decouple microservices so producers and consumers evolve independently.
• Request buffering — absorb traffic spikes by queuing requests and processing them at a sustainable rate.
• Complex routing — route messages to different queues based on content, headers, or pattern matching.
• Reliable delivery — guarantee message delivery with acknowledgments, persistence, and publisher confirms.
• Fan-out notifications — broadcast events to multiple consumers simultaneously.

Core Concepts: The AMQP Model

Understanding AMQP is essential before writing any RabbitMQ code. The protocol defines a precise message flow that separates publishing from consuming through an intermediary routing layer.

AMQP Message Flow:

  Producer                    RabbitMQ Broker                     Consumer
  ┌──────┐    publish     ┌───────────────────────┐    deliver   ┌──────────┐
  │ App  │ ──────────────>│ Exchange              │             │ Worker 1 │
  │      │   routing_key  │   │                   │             └──────────┘
  └──────┘   + payload    │   │ binding rules     │    deliver   ┌──────────┐
                          │   ▼                   │ ──────────> │ Worker 2 │
                          │ Queue 1               │             └──────────┘
                          │ Queue 2               │    deliver   ┌──────────┐
                          │ Queue 3               │ ──────────> │ Worker 3 │
                          └───────────────────────┘             └──────────┘

Key entities:
  Exchange     — receives messages and routes them to queues
  Queue        — stores messages in FIFO order until consumed
  Binding      — rule linking an exchange to a queue (with optional routing key)
  Routing Key  — label attached to a message; used by exchanges for routing
  Virtual Host — logical grouping / namespace for isolation (like a database schema)
  Channel      — multiplexed lightweight connection within a TCP connection
  Connection   — single TCP connection to the broker (expensive to create)

Connections vs Channels

A connection is a TCP socket between your application and the RabbitMQ server. Creating connections is expensive (TCP handshake + AMQP handshake + TLS negotiation). A channel is a virtual connection inside a real TCP connection. Channels are lightweight and can be created/destroyed quickly. Best practice: one connection per application process, one channel per thread or async context. Never share a channel across threads.

Virtual Hosts

Virtual hosts (vhosts) provide logical separation within a single RabbitMQ instance. Each vhost has its own exchanges, queues, bindings, permissions, and policies. Use vhosts to separate environments (dev, staging, production) or tenants in a multi-tenant system. The default vhost is /.

# Create a virtual host
rabbitmqctl add_vhost my-app-staging

# Set permissions for a user on a vhost
# Format: set_permissions -p <vhost> <user> <configure> <write> <read>
rabbitmqctl set_permissions -p my-app-staging myuser ".*" ".*" ".*"

# List virtual hosts
rabbitmqctl list_vhosts

# Delete a virtual host (deletes ALL its exchanges, queues, bindings)
rabbitmqctl delete_vhost my-app-staging

Exchange Types: Direct, Fanout, Topic, Headers

Exchanges are the routing engine of RabbitMQ. A producer never sends messages directly to a queue; it publishes to an exchange, which then decides which queues receive the message based on the exchange type and binding configuration.

Direct Exchange

A direct exchange routes messages to queues whose binding key exactly matches the message routing key. This is the simplest and most common pattern — one-to-one or selective delivery.

Direct Exchange Routing:

  Producer publishes with routing_key="order.created"

  Exchange (direct)
    │
    ├── binding_key="order.created"  ──> Queue: order-processing  ✓ MATCH
    ├── binding_key="order.shipped"  ──> Queue: shipping-notify   ✗ no match
    └── binding_key="order.created"  ──> Queue: analytics         ✓ MATCH

  Result: message delivered to order-processing AND analytics queues

Fanout Exchange

A fanout exchange broadcasts every message to all queues bound to it, ignoring the routing key entirely. Use fanout for event broadcasting, logging, or any scenario where every consumer needs every message.

Fanout Exchange Routing:

  Producer publishes (routing_key is ignored)

  Exchange (fanout)
    │
    ├── Queue: email-service       ✓ receives copy
    ├── Queue: push-notifications  ✓ receives copy
    ├── Queue: analytics-pipeline  ✓ receives copy
    └── Queue: audit-log           ✓ receives copy

  Result: ALL bound queues receive a copy of the message

Topic Exchange

A topic exchange routes messages using wildcard pattern matching on the routing key. Routing keys are dot-separated words (e.g., order.us.created). Bindings use two wildcards: * matches exactly one word and # matches zero or more words. Topic exchanges are the most flexible and commonly used in event-driven architectures.

Topic Exchange Routing:

  Routing key format: <entity>.<region>.<action>

  Producer publishes with routing_key="order.us.created"

  Exchange (topic)
    │
    ├── binding="order.us.*"       ──> Queue: us-orders         ✓ matches
    ├── binding="order.#"          ──> Queue: all-orders         ✓ matches
    ├── binding="order.eu.*"       ──> Queue: eu-orders          ✗ no match
    ├── binding="*.*.created"      ──> Queue: created-events     ✓ matches
    ├── binding="#"                 ──> Queue: everything         ✓ matches all
    └── binding="payment.#"        ──> Queue: payments           ✗ no match

  Wildcard rules:
    *  = exactly one word          "order.*"    matches "order.created" but NOT "order.us.created"
    #  = zero or more words        "order.#"    matches "order", "order.created", "order.us.created"

Headers Exchange

A headers exchange routes messages based on message header attributes instead of the routing key. Bindings specify a set of header key-value pairs and an x-match argument: all means all headers must match, any means at least one must match. Use headers exchanges when routing logic cannot be expressed as a string key.

// Publishing with headers for a headers exchange
channel.publish('headers-exchange', '', Buffer.from(JSON.stringify(payload)), {
  headers: {
    'x-region': 'us-east',
    'x-priority': 'high',
    'x-type': 'order',
  },
});

// Binding a queue to a headers exchange
// x-match: 'all' means ALL specified headers must match
channel.bindQueue('urgent-us-orders', 'headers-exchange', '', {
  'x-match': 'all',
  'x-region': 'us-east',
  'x-priority': 'high',
});

// x-match: 'any' means AT LEAST ONE header must match
channel.bindQueue('any-high-priority', 'headers-exchange', '', {
  'x-match': 'any',
  'x-priority': 'high',
  'x-type': 'payment',
});

Message Acknowledgment, Persistence, and Durability

RabbitMQ provides multiple layers of reliability guarantees. Understanding each layer is critical for building systems that do not lose messages.

Consumer Acknowledgments

When a consumer receives a message, it must tell RabbitMQ whether it was processed successfully. There are three acknowledgment modes:

// 1. Manual acknowledgment (recommended for reliability)
channel.consume('task-queue', (msg) => {
  try {
    const task = JSON.parse(msg.content.toString());
    processTask(task);

    // Success: acknowledge the message — removes it from the queue
    channel.ack(msg);
  } catch (error) {
    // Failure: reject and requeue for another attempt
    // nack(message, allUpTo, requeue)
    channel.nack(msg, false, true);

    // Or: reject without requeue (message is discarded or dead-lettered)
    // channel.nack(msg, false, false);
  }
}, { noAck: false }); // noAck: false = manual acknowledgment

// 2. Automatic acknowledgment (fire-and-forget, risk of message loss)
channel.consume('log-queue', (msg) => {
  console.log(msg.content.toString());
  // No ack needed — message removed from queue on delivery
}, { noAck: true });

// 3. Batch acknowledgment (acknowledge multiple messages at once)
let count = 0;
channel.consume('batch-queue', (msg) => {
  processBatch(msg);
  count++;

  if (count % 10 === 0) {
    // ack(message, allUpTo=true) — acks this message and all previous unacked
    channel.ack(msg, true);
  }
}, { noAck: false });

Queue Durability

A durable queue survives a broker restart. A transient queue is deleted when the broker restarts. Durability is set when the queue is first declared and cannot be changed afterward.

// Durable queue — survives broker restart
channel.assertQueue('important-tasks', {
  durable: true,     // queue definition persisted to disk
  autoDelete: false,  // do not delete when last consumer disconnects
  exclusive: false,   // allow multiple connections to consume
});

// Transient queue — deleted on broker restart
channel.assertQueue('temp-results', {
  durable: false,
  autoDelete: true,   // delete when last consumer disconnects
});

// Exclusive queue — private to this connection, auto-deleted
channel.assertQueue('', {
  exclusive: true,    // only this connection can consume; auto-named
});

Message Persistence

Even if a queue is durable, messages inside it are lost on restart unless they are marked as persistent. Persistent messages are written to disk by the broker.

// Persistent message — written to disk
channel.sendToQueue('important-tasks', Buffer.from(JSON.stringify(task)), {
  persistent: true,   // equivalent to deliveryMode: 2
  contentType: 'application/json',
  messageId: uuidv4(),
  timestamp: Date.now(),
});

// Transient message — kept in memory only (faster, but lost on restart)
channel.sendToQueue('cache-updates', Buffer.from(data), {
  persistent: false,  // equivalent to deliveryMode: 1
});

// IMPORTANT: persistent + durable queue alone is NOT a guarantee.
// Use publisher confirms for true delivery assurance (see next section).

Publisher Confirms and Consumer Prefetch

Publisher confirms and consumer prefetch are two critical mechanisms for production reliability and performance tuning.

Publisher Confirms

Without publisher confirms, a channel.publish() call returns immediately with no guarantee that the broker received the message. Network failures, broker crashes, or full queues can silently drop messages. Publisher confirms make the broker send an ack/nack back to the producer for every message.

// Enable publisher confirms on the channel
await channel.confirmSelect();

// Method 1: Callback-based confirms
channel.sendToQueue('orders', Buffer.from(data), { persistent: true }, (err, ok) => {
  if (err) {
    console.error('Message nacked by broker:', err);
    // Retry or store in local fallback
  } else {
    console.log('Message confirmed by broker');
  }
});

// Method 2: Promise-based with waitForConfirms
channel.sendToQueue('orders', Buffer.from(data), { persistent: true });
channel.sendToQueue('orders', Buffer.from(data2), { persistent: true });

try {
  await channel.waitForConfirms();
  console.log('All pending messages confirmed');
} catch (err) {
  console.error('Some messages were nacked:', err);
}

// Method 3: Individual promise per publish (amqplib pattern)
const confirmChannel = await connection.createConfirmChannel();

function publishWithConfirm(exchange, routingKey, content, options) {
  return new Promise((resolve, reject) => {
    confirmChannel.publish(exchange, routingKey, content, options, (err) => {
      if (err) reject(err);
      else resolve(true);
    });
  });
}

await publishWithConfirm('orders-exchange', 'order.created', buffer, { persistent: true });

Consumer Prefetch (QoS)

Prefetch count controls how many unacknowledged messages a consumer can hold. Without prefetch, RabbitMQ pushes messages as fast as possible, overwhelming slow consumers while fast consumers sit idle. Setting prefetch enables fair dispatch and backpressure.

// Set prefetch count — limits unacked messages per consumer
channel.prefetch(10); // Each consumer gets at most 10 unacked messages

// Per-channel vs per-consumer prefetch
channel.prefetch(10, false); // false = per-consumer (default, recommended)
channel.prefetch(50, true);  // true  = per-channel (shared across all consumers on this channel)

// Prefetch tuning guidelines:
//
// | Workload Type    | Prefetch | Rationale                          |
// |------------------|----------|------------------------------------|
// | CPU-bound tasks  | 1-5      | Slow processing, fair distribution |
// | I/O-bound tasks  | 10-50    | Waiting on network, higher useful  |
// | Logging/metrics  | 100-250  | Fast processing, maximize batch    |
// | Streaming        | 250-1000 | Consumer can handle high volume    |
//
// Start low and increase while monitoring consumer utilization
// in the Management UI (should be close to 100%)

Node.js Client (amqplib) Examples

The amqplib package is the standard RabbitMQ client for Node.js. It supports both callback and promise-based APIs. Always use the promise API (amqplib) for modern async/await code.

Connection and Basic Producer/Consumer

// install: npm install amqplib

const amqp = require('amqplib');

// === Producer ===
async function producer() {
  // Connect to RabbitMQ
  const connection = await amqp.connect('amqp://user:password@localhost:5672/my-vhost');
  const channel = await connection.createConfirmChannel();

  const exchange = 'orders-exchange';
  const queue = 'order-processing';
  const routingKey = 'order.created';

  // Declare exchange and queue (idempotent)
  await channel.assertExchange(exchange, 'direct', { durable: true });
  await channel.assertQueue(queue, { durable: true });
  await channel.bindQueue(queue, exchange, routingKey);

  // Publish a message
  const order = { id: 'ord_123', product: 'Widget', quantity: 5, total: 49.95 };
  const content = Buffer.from(JSON.stringify(order));

  channel.publish(exchange, routingKey, content, {
    persistent: true,
    contentType: 'application/json',
    messageId: 'msg_' + Date.now(),
    timestamp: Math.floor(Date.now() / 1000),
    headers: { 'x-retry-count': 0 },
  }, (err) => {
    if (err) console.error('Publish nacked:', err);
    else console.log('Order published and confirmed');
  });

  await channel.waitForConfirms();

  // Cleanup
  await channel.close();
  await connection.close();
}

// === Consumer ===
async function consumer() {
  const connection = await amqp.connect('amqp://user:password@localhost:5672/my-vhost');
  const channel = await connection.createChannel();

  const queue = 'order-processing';
  await channel.assertQueue(queue, { durable: true });

  // Set prefetch
  channel.prefetch(10);

  console.log('Waiting for messages on', queue);

  channel.consume(queue, async (msg) => {
    if (!msg) return; // consumer cancelled by server

    try {
      const order = JSON.parse(msg.content.toString());
      console.log('Processing order:', order.id);

      // Simulate async processing
      await processOrder(order);

      // Acknowledge success
      channel.ack(msg);
    } catch (error) {
      console.error('Failed to process:', error.message);

      const retryCount = (msg.properties.headers['x-retry-count'] || 0);
      if (retryCount < 3) {
        // Requeue for retry
        channel.nack(msg, false, true);
      } else {
        // Max retries exceeded — send to dead letter queue
        channel.nack(msg, false, false);
      }
    }
  }, { noAck: false });

  // Handle connection errors
  connection.on('error', (err) => console.error('Connection error:', err));
  connection.on('close', () => {
    console.error('Connection closed, reconnecting...');
    setTimeout(consumer, 5000);
  });
}

async function processOrder(order) {
  // Your business logic here
  console.log('Order ' + order.id + ' processed successfully');
}

producer().catch(console.error);
consumer().catch(console.error);

Topic Exchange with Multiple Consumers

const amqp = require('amqplib');

async function setupTopicExchange() {
  const conn = await amqp.connect('amqp://localhost');
  const ch = await conn.createChannel();

  const exchange = 'events';
  await ch.assertExchange(exchange, 'topic', { durable: true });

  // Queue 1: All order events
  const q1 = await ch.assertQueue('order-events', { durable: true });
  await ch.bindQueue(q1.queue, exchange, 'order.#');

  // Queue 2: Only US creation events
  const q2 = await ch.assertQueue('us-created', { durable: true });
  await ch.bindQueue(q2.queue, exchange, '*.us.created');

  // Queue 3: All creation events regardless of entity or region
  const q3 = await ch.assertQueue('all-created', { durable: true });
  await ch.bindQueue(q3.queue, exchange, '#.created');

  // Publish test messages
  const messages = [
    { key: 'order.us.created', data: 'US order created' },
    { key: 'order.eu.created', data: 'EU order created' },
    { key: 'order.us.shipped', data: 'US order shipped' },
    { key: 'payment.us.created', data: 'US payment created' },
    { key: 'user.signup', data: 'New user signup' },
  ];

  for (const m of messages) {
    ch.publish(exchange, m.key, Buffer.from(m.data), { persistent: true });
    console.log('Published: [' + m.key + '] ' + m.data);
  }

  // Results:
  // order-events queue receives: order.us.created, order.eu.created, order.us.shipped
  // us-created queue receives:   order.us.created, payment.us.created
  // all-created queue receives:  order.us.created, order.eu.created, payment.us.created
}

setupTopicExchange().catch(console.error);

Robust Connection Management

const amqp = require('amqplib');

class RabbitMQClient {
  constructor(url, options = {}) {
    this.url = url;
    this.prefetch = options.prefetch || 10;
    this.reconnectDelay = options.reconnectDelay || 5000;
    this.connection = null;
    this.channel = null;
    this.isConnecting = false;
    this.consumers = new Map();
  }

  async connect() {
    if (this.isConnecting) return;
    this.isConnecting = true;

    try {
      this.connection = await amqp.connect(this.url, {
        heartbeat: 30, // seconds — detects dead connections
      });
      this.channel = await this.connection.createChannel();
      await this.channel.prefetch(this.prefetch);

      // Re-register consumers after reconnection
      for (const [queue, handler] of this.consumers) {
        await this._subscribe(queue, handler);
      }

      this.connection.on('error', (err) => {
        console.error('[RabbitMQ] Connection error:', err.message);
      });

      this.connection.on('close', () => {
        console.warn('[RabbitMQ] Connection closed. Reconnecting...');
        this.connection = null;
        this.channel = null;
        this.isConnecting = false;
        setTimeout(() => this.connect(), this.reconnectDelay);
      });

      this.channel.on('error', (err) => {
        console.error('[RabbitMQ] Channel error:', err.message);
      });

      console.log('[RabbitMQ] Connected successfully');
    } catch (err) {
      console.error('[RabbitMQ] Connect failed:', err.message);
      this.isConnecting = false;
      setTimeout(() => this.connect(), this.reconnectDelay);
      return;
    }

    this.isConnecting = false;
  }

  async publish(exchange, routingKey, data, options = {}) {
    if (!this.channel) throw new Error('Not connected');

    const content = Buffer.from(JSON.stringify(data));
    this.channel.publish(exchange, routingKey, content, {
      persistent: true,
      contentType: 'application/json',
      timestamp: Math.floor(Date.now() / 1000),
      ...options,
    });
  }

  async subscribe(queue, handler) {
    this.consumers.set(queue, handler);
    if (this.channel) {
      await this._subscribe(queue, handler);
    }
  }

  async _subscribe(queue, handler) {
    await this.channel.assertQueue(queue, { durable: true });
    this.channel.consume(queue, async (msg) => {
      if (!msg) return;
      try {
        const data = JSON.parse(msg.content.toString());
        await handler(data, msg);
        this.channel.ack(msg);
      } catch (err) {
        console.error('[RabbitMQ] Handler error:', err.message);
        this.channel.nack(msg, false, false); // dead-letter on failure
      }
    }, { noAck: false });
  }

  async close() {
    if (this.channel) await this.channel.close();
    if (this.connection) await this.connection.close();
  }
}

// Usage
const client = new RabbitMQClient('amqp://user:pass@localhost:5672');
await client.connect();

await client.subscribe('order-processing', async (order) => {
  console.log('Processing:', order.id);
  await saveToDatabase(order);
});

await client.publish('orders-exchange', 'order.created', {
  id: 'ord_456', product: 'Gadget', total: 29.99
});

Python Client (pika) Examples

The pika library is the most popular Python client for RabbitMQ. It supports blocking (synchronous) and asynchronous (SelectConnection, asyncio) connection adapters. For production services, use the asynchronous adapter or combine the blocking adapter with threading.

Basic Producer and Consumer

# install: pip install pika

import pika
import json
import uuid
from datetime import datetime

# === Producer ===
def produce_message():
    credentials = pika.PlainCredentials('user', 'password')
    parameters = pika.ConnectionParameters(
        host='localhost',
        port=5672,
        virtual_host='/',
        credentials=credentials,
        heartbeat=30,
        blocked_connection_timeout=300,
    )

    connection = pika.BlockingConnection(parameters)
    channel = connection.channel()

    # Declare exchange and queue
    channel.exchange_declare(exchange='tasks', exchange_type='direct', durable=True)
    channel.queue_declare(queue='email-tasks', durable=True)
    channel.queue_bind(queue='email-tasks', exchange='tasks', routing_key='email.send')

    # Enable publisher confirms
    channel.confirm_delivery()

    # Publish message
    task = {
        'id': str(uuid.uuid4()),
        'to': 'user@example.com',
        'subject': 'Welcome!',
        'template': 'welcome_email',
        'created_at': datetime.utcnow().isoformat(),
    }

    try:
        channel.basic_publish(
            exchange='tasks',
            routing_key='email.send',
            body=json.dumps(task),
            properties=pika.BasicProperties(
                delivery_mode=pika.DeliveryMode.Persistent,
                content_type='application/json',
                message_id=task['id'],
                timestamp=int(datetime.utcnow().timestamp()),
            ),
        )
        print(f"Published task {task['id']}")
    except pika.exceptions.UnroutableError:
        print('Message could not be routed')

    connection.close()


# === Consumer ===
def consume_messages():
    credentials = pika.PlainCredentials('user', 'password')
    parameters = pika.ConnectionParameters(
        host='localhost',
        port=5672,
        credentials=credentials,
        heartbeat=30,
    )

    connection = pika.BlockingConnection(parameters)
    channel = connection.channel()

    channel.queue_declare(queue='email-tasks', durable=True)
    channel.basic_qos(prefetch_count=5)

    def callback(ch, method, properties, body):
        try:
            task = json.loads(body)
            print(f"Processing email task: {task['id']}")

            # Simulate sending email
            send_email(task)

            # Acknowledge success
            ch.basic_ack(delivery_tag=method.delivery_tag)
            print(f"Task {task['id']} completed")

        except Exception as e:
            print(f"Task failed: {e}")
            # Reject without requeue (will go to dead letter queue if configured)
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

    channel.basic_consume(queue='email-tasks', on_message_callback=callback)
    print('Waiting for email tasks...')
    channel.start_consuming()


def send_email(task):
    """Placeholder for email sending logic."""
    print(f"Email sent to {task['to']}: {task['subject']}")


if __name__ == '__main__':
    import sys
    if len(sys.argv) > 1 and sys.argv[1] == 'produce':
        produce_message()
    else:
        consume_messages()

Python Async Consumer with aio-pika

# install: pip install aio-pika

import asyncio
import aio_pika
import json

async def main():
    # Connect using asyncio
    connection = await aio_pika.connect_robust(
        'amqp://user:password@localhost/',
        heartbeat=30,
    )

    async with connection:
        channel = await connection.channel()
        await channel.set_qos(prefetch_count=10)

        # Declare queue
        queue = await channel.declare_queue('async-tasks', durable=True)

        async def process_message(message: aio_pika.IncomingMessage):
            async with message.process():
                # .process() auto-acks on success, auto-nacks on exception
                data = json.loads(message.body.decode())
                print(f"Processing: {data}")
                await asyncio.sleep(0.1)  # simulate async work

        # Start consuming
        await queue.consume(process_message)
        print('Waiting for messages...')

        # Run forever
        await asyncio.Future()

asyncio.run(main())

Dead Letter Queues and Retry Patterns

Dead letter queues (DLQs) are essential for handling messages that cannot be processed. A message is dead-lettered when:

• The consumer rejects it with requeue=false (channel.nack(msg, false, false)).
• The message TTL expires (per-message or per-queue TTL).
• The queue exceeds its x-max-length or x-max-length-bytes limit.

Setting Up a Dead Letter Queue

const amqp = require('amqplib');

async function setupDLQ() {
  const conn = await amqp.connect('amqp://localhost');
  const ch = await conn.createChannel();

  // Step 1: Declare the dead letter exchange and queue
  await ch.assertExchange('dlx', 'direct', { durable: true });
  await ch.assertQueue('dead-letters', { durable: true });
  await ch.bindQueue('dead-letters', 'dlx', 'failed');

  // Step 2: Declare the main exchange and queue WITH dead letter config
  await ch.assertExchange('main-exchange', 'direct', { durable: true });
  await ch.assertQueue('work-queue', {
    durable: true,
    arguments: {
      'x-dead-letter-exchange': 'dlx',           // route dead letters here
      'x-dead-letter-routing-key': 'failed',      // with this routing key
      'x-message-ttl': 30000,                     // optional: expire after 30s
      'x-max-length': 10000,                      // optional: max queue size
    },
  });
  await ch.bindQueue('work-queue', 'main-exchange', 'work');

  console.log('DLQ setup complete');
  console.log('  Main queue: work-queue');
  console.log('  Dead letter queue: dead-letters');
}

setupDLQ().catch(console.error);

Retry with Exponential Backoff Using TTL

RabbitMQ does not have native retry with delay, but you can implement it using multiple queues with different TTLs that dead-letter back to the main queue. This creates a retry loop with increasing delays.

async function setupRetryQueues(ch) {
  const mainExchange = 'app-exchange';
  const retryExchange = 'retry-exchange';
  const dlExchange = 'dl-exchange';

  // Main exchange and queue
  await ch.assertExchange(mainExchange, 'direct', { durable: true });

  // Retry exchange (messages loop back here after delay)
  await ch.assertExchange(retryExchange, 'direct', { durable: true });

  // Final dead letter exchange (after max retries)
  await ch.assertExchange(dlExchange, 'direct', { durable: true });

  // Dead letter queue for permanently failed messages
  await ch.assertQueue('permanent-failures', { durable: true });
  await ch.bindQueue('permanent-failures', dlExchange, 'dead');

  // Retry queues with increasing TTL
  const retryDelays = [5000, 15000, 60000]; // 5s, 15s, 60s

  for (let i = 0; i < retryDelays.length; i++) {
    const retryQueue = 'retry-' + (i + 1);
    await ch.assertQueue(retryQueue, {
      durable: true,
      arguments: {
        'x-dead-letter-exchange': mainExchange,    // dead-letter back to main
        'x-dead-letter-routing-key': 'work',
        'x-message-ttl': retryDelays[i],           // delay before retry
      },
    });
    await ch.bindQueue(retryQueue, retryExchange, 'retry-' + (i + 1));
  }

  // Main work queue
  await ch.assertQueue('work-queue', { durable: true });
  await ch.bindQueue('work-queue', mainExchange, 'work');

  // Consumer with retry logic
  ch.prefetch(10);
  ch.consume('work-queue', async (msg) => {
    try {
      const data = JSON.parse(msg.content.toString());
      await processMessage(data);
      ch.ack(msg);
    } catch (error) {
      const retryCount = (msg.properties.headers['x-retry-count'] || 0) + 1;
      const maxRetries = retryDelays.length;

      if (retryCount <= maxRetries) {
        // Route to appropriate retry queue
        const retryKey = 'retry-' + retryCount;
        ch.publish(retryExchange, retryKey, msg.content, {
          persistent: true,
          headers: { ...msg.properties.headers, 'x-retry-count': retryCount },
        });
        ch.ack(msg); // ack original to remove from work queue
        console.log('Retry ' + retryCount + '/' + maxRetries + ' in ' + retryDelays[retryCount - 1] + 'ms');
      } else {
        // Max retries exceeded, send to permanent DLQ
        ch.publish(dlExchange, 'dead', msg.content, {
          persistent: true,
          headers: { ...msg.properties.headers, 'x-retry-count': retryCount, 'x-failure-reason': error.message },
        });
        ch.ack(msg);
        console.error('Max retries exceeded. Message sent to DLQ.');
      }
    }
  }, { noAck: false });
}

async function processMessage(data) {
  // Your business logic
  if (Math.random() < 0.3) throw new Error('Random processing failure');
  console.log('Processed:', data);
}

RabbitMQ Clustering and High Availability

A single RabbitMQ node is a single point of failure. Clustering connects multiple RabbitMQ nodes into a logical broker, sharing metadata (exchanges, bindings, vhosts, users) across all nodes. Queue data replication requires either quorum queues or classic mirrored queues (deprecated).

Cluster Architecture

RabbitMQ Cluster Architecture:

  ┌────────────────────────────────────────────────────┐
  │                 Load Balancer                      │
  │              (HAProxy / Nginx)                     │
  │         port 5672 (AMQP) + 15672 (Management)     │
  └──────────┬──────────────┬──────────────┬───────────┘
             │              │              │
       ┌─────▼────┐  ┌─────▼────┐  ┌─────▼────┐
       │  Node 1  │  │  Node 2  │  │  Node 3  │
       │ (leader) │  │(follower)│  │(follower)│
       │          │  │          │  │          │
       │ Quorum Q │  │ Quorum Q │  │ Quorum Q │
       │ (leader) │  │ (replica)│  │ (replica)│
       └──────────┘  └──────────┘  └──────────┘
             │              │              │
             └──────────────┴──────────────┘
                    Erlang distribution
                  (port 25672, same cookie)

  Shared across all nodes:
    - Exchanges, bindings, vhosts, users, policies
  Replicated by quorum queues:
    - Queue data (messages) via Raft consensus
  NOT shared (classic queues):
    - Queue data lives on the declaring node only

Setting Up a Cluster

# All nodes must share the same Erlang cookie
# Copy .erlang.cookie from node1 to node2 and node3
scp /var/lib/rabbitmq/.erlang.cookie node2:/var/lib/rabbitmq/
scp /var/lib/rabbitmq/.erlang.cookie node3:/var/lib/rabbitmq/

# On node2: join the cluster
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

# On node3: join the cluster
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

# Verify cluster status
rabbitmqctl cluster_status

# Output will show:
# Cluster name: rabbit@node1
# Running Nodes: rabbit@node1, rabbit@node2, rabbit@node3

Quorum Queues (Recommended for HA)

Quorum queues are the recommended queue type for high availability. They use the Raft consensus algorithm to replicate messages across cluster nodes. A quorum queue with a replication factor of 3 tolerates 1 node failure; a factor of 5 tolerates 2 failures. Quorum queues replace the older, deprecated classic mirrored queues.

// Declare a quorum queue (Node.js)
channel.assertQueue('important-orders', {
  durable: true,
  arguments: {
    'x-queue-type': 'quorum',            // use Raft-based replication
    'x-quorum-initial-group-size': 3,    // replicas across 3 nodes
    'x-delivery-limit': 5,              // redeliver at most 5 times before DLQ
  },
});

// Quorum queues automatically:
// - Replicate messages to all group members
// - Elect a new leader if the current leader fails
// - Support poison message handling via x-delivery-limit
// - Provide at-least-once delivery guarantees

# Declare a quorum queue (Python)
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(
    queue='important-orders',
    durable=True,
    arguments={
        'x-queue-type': 'quorum',
        'x-quorum-initial-group-size': 3,
        'x-delivery-limit': 5,
    },
)

HAProxy Load Balancer Configuration

# /etc/haproxy/haproxy.cfg

frontend rabbitmq_amqp
    bind *:5672
    mode tcp
    default_backend rabbitmq_nodes

frontend rabbitmq_management
    bind *:15672
    mode http
    default_backend rabbitmq_mgmt_nodes

backend rabbitmq_nodes
    mode tcp
    balance roundrobin
    option tcpka         # enable TCP keep-alive
    timeout connect 5s
    timeout server 30s

    server node1 10.0.1.1:5672 check inter 5s rise 2 fall 3
    server node2 10.0.1.2:5672 check inter 5s rise 2 fall 3
    server node3 10.0.1.3:5672 check inter 5s rise 2 fall 3

backend rabbitmq_mgmt_nodes
    mode http
    balance roundrobin
    server node1 10.0.1.1:15672 check
    server node2 10.0.1.2:15672 check
    server node3 10.0.1.3:15672 check

Monitoring with the Management Plugin

The RabbitMQ Management Plugin provides a web-based UI, a REST API, and the rabbitmqadminCLI tool for monitoring and managing your broker. It is the primary tool for production observability.

Enabling the Management Plugin

# Enable the management plugin
rabbitmq-plugins enable rabbitmq_management

# Access the web UI at http://localhost:15672
# Default credentials: guest / guest (only works from localhost)

# Create an admin user for remote access
rabbitmqctl add_user admin securepassword123
rabbitmqctl set_user_tags admin administrator
rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"

Key Metrics to Monitor

Critical RabbitMQ Metrics:

  Queue Metrics:
  ├── messages_ready           — messages waiting to be delivered
  ├── messages_unacknowledged  — delivered but not yet acked
  ├── message_bytes            — total bytes of messages in queue
  ├── consumers                — number of active consumers
  └── consumer_utilisation     — % of time consumers are busy (target: >95%)

  Node Metrics:
  ├── mem_used / mem_limit     — memory usage vs high watermark
  ├── disk_free / disk_limit   — free disk vs low watermark
  ├── fd_used / fd_total       — file descriptors in use
  ├── sockets_used             — TCP connections
  ├── proc_used                — Erlang processes
  └── run_queue                — Erlang scheduler run queue (>0 = CPU pressure)

  Connection/Channel Metrics:
  ├── connection_count         — total TCP connections
  ├── channel_count            — total channels
  ├── publish_rate             — messages published per second
  ├── deliver_rate             — messages delivered per second
  └── ack_rate                 — messages acknowledged per second

  Alerts to set:
  ├── messages_ready > 10000           → consumers not keeping up
  ├── messages_unacknowledged > 1000   → consumer processing too slow
  ├── mem_used > 80% of mem_limit      → approaching memory alarm
  ├── disk_free < 2x disk_limit        → approaching disk alarm
  ├── consumer_utilisation < 50%       → prefetch may be too low
  └── connection_count > 1000          → possible connection leak

Management HTTP API Examples

# List all queues with message counts
curl -u admin:password http://localhost:15672/api/queues | jq '.[] | {name, messages, consumers}'

# Get details of a specific queue
curl -u admin:password http://localhost:15672/api/queues/%2F/order-processing

# Purge a queue (delete all messages)
curl -u admin:password -X DELETE http://localhost:15672/api/queues/%2F/old-queue/contents

# List connections
curl -u admin:password http://localhost:15672/api/connections

# Get node health check
curl -u admin:password http://localhost:15672/api/health/checks/alarms

# Export broker definitions (exchanges, queues, bindings, policies)
curl -u admin:password http://localhost:15672/api/definitions > rabbitmq-definitions.json

# Import definitions (useful for disaster recovery or environment cloning)
curl -u admin:password -X POST -H "Content-Type: application/json" \
  -d @rabbitmq-definitions.json http://localhost:15672/api/definitions

Prometheus and Grafana Integration

# Enable the Prometheus plugin (built-in since RabbitMQ 3.8)
rabbitmq-plugins enable rabbitmq_prometheus

# Prometheus metrics endpoint:
# http://localhost:15692/metrics

# Key Prometheus metrics:
# rabbitmq_queue_messages_ready
# rabbitmq_queue_messages_unacked
# rabbitmq_queue_consumers
# rabbitmq_connections
# rabbitmq_channels
# rabbitmq_node_mem_used
# rabbitmq_node_disk_free

RabbitMQ vs Kafka vs SQS vs Redis Pub/Sub

Choosing the right message system depends on your requirements for throughput, ordering, retention, routing complexity, and operational overhead. Here is a detailed comparison.

When to choose RabbitMQ: You need flexible message routing, per-message acknowledgment, dead letter handling, priority queues, request-reply patterns, or you are processing up to ~100K messages per second. RabbitMQ is the best general-purpose message broker for microservice architectures.

When to choose Kafka: You need event sourcing, stream processing, message replay, long-term retention, or throughput exceeding 100K messages per second. Kafka is an event streaming platform, not a traditional message broker.

When to choose SQS: You are on AWS, want zero operational overhead, and need simple point-to-point queuing without complex routing. Pair with SNS for fan-out.

When to choose Redis Pub/Sub: You need real-time fire-and-forget notifications with sub-millisecond latency and already have Redis. Not suitable when you cannot afford to lose messages. Consider Redis Streams for persistence.

Docker Compose Setup for Development

The fastest way to run RabbitMQ locally for development is with Docker Compose. The management image tag includes the Management Plugin (web UI).

# docker-compose.yml
version: "3.9"

services:
  rabbitmq:
    image: rabbitmq:3.13-management-alpine
    container_name: rabbitmq
    hostname: rabbitmq
    ports:
      - "5672:5672"     # AMQP
      - "15672:15672"   # Management UI
      - "15692:15692"   # Prometheus metrics (if plugin enabled)
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: secretpassword
      RABBITMQ_DEFAULT_VHOST: /
      RABBITMQ_VM_MEMORY_HIGH_WATERMARK: 0.7
    volumes:
      - rabbitmq-data:/var/lib/rabbitmq
      - ./rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro
      - ./definitions.json:/etc/rabbitmq/definitions.json:ro
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "check_port_connectivity"]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 30s
    restart: unless-stopped

volumes:
  rabbitmq-data:

Custom Configuration File

# rabbitmq.conf — place next to docker-compose.yml

# Load definitions on startup (exchanges, queues, bindings, users)
management.load_definitions = /etc/rabbitmq/definitions.json

# Logging
log.console = true
log.console.level = info

# Memory and disk limits
vm_memory_high_watermark.relative = 0.7
disk_free_limit.absolute = 1GB

# Channel and connection limits
channel_max = 2047
heartbeat = 30

# Consumer timeout (unacked messages)
consumer_timeout = 1800000

# Prometheus metrics
prometheus.return_per_object_metrics = true

Pre-loaded Definitions

{
  "vhosts": [
    { "name": "/" },
    { "name": "staging" }
  ],
  "users": [
    {
      "name": "admin",
      "password_hash": "use_rabbitmqctl_to_generate",
      "tags": "administrator"
    },
    {
      "name": "app-service",
      "password_hash": "use_rabbitmqctl_to_generate",
      "tags": ""
    }
  ],
  "permissions": [
    { "user": "admin", "vhost": "/", "configure": ".*", "write": ".*", "read": ".*" },
    { "user": "app-service", "vhost": "/", "configure": "^app\\..*", "write": ".*", "read": ".*" }
  ],
  "exchanges": [
    { "name": "orders", "vhost": "/", "type": "topic", "durable": true },
    { "name": "notifications", "vhost": "/", "type": "fanout", "durable": true },
    { "name": "dlx", "vhost": "/", "type": "direct", "durable": true }
  ],
  "queues": [
    {
      "name": "order-processing",
      "vhost": "/",
      "durable": true,
      "arguments": { "x-queue-type": "quorum" }
    },
    {
      "name": "dead-letters",
      "vhost": "/",
      "durable": true,
      "arguments": {}
    }
  ],
  "bindings": [
    { "source": "orders", "vhost": "/", "destination": "order-processing", "destination_type": "queue", "routing_key": "order.#" },
    { "source": "dlx", "vhost": "/", "destination": "dead-letters", "destination_type": "queue", "routing_key": "dead" }
  ]
}

Three-Node Cluster with Docker Compose

# docker-compose-cluster.yml
version: "3.9"

services:
  rabbitmq1:
    image: rabbitmq:3.13-management-alpine
    hostname: rabbitmq1
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: secretpassword
      RABBITMQ_ERLANG_COOKIE: "shared-secret-cookie-value"
      RABBITMQ_NODENAME: rabbit@rabbitmq1
    ports:
      - "5672:5672"
      - "15672:15672"
    volumes:
      - node1-data:/var/lib/rabbitmq
    networks:
      - rabbitmq-cluster

  rabbitmq2:
    image: rabbitmq:3.13-management-alpine
    hostname: rabbitmq2
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: secretpassword
      RABBITMQ_ERLANG_COOKIE: "shared-secret-cookie-value"
      RABBITMQ_NODENAME: rabbit@rabbitmq2
    depends_on:
      - rabbitmq1
    volumes:
      - node2-data:/var/lib/rabbitmq
    networks:
      - rabbitmq-cluster

  rabbitmq3:
    image: rabbitmq:3.13-management-alpine
    hostname: rabbitmq3
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: secretpassword
      RABBITMQ_ERLANG_COOKIE: "shared-secret-cookie-value"
      RABBITMQ_NODENAME: rabbit@rabbitmq3
    depends_on:
      - rabbitmq1
    volumes:
      - node3-data:/var/lib/rabbitmq
    networks:
      - rabbitmq-cluster

volumes:
  node1-data:
  node2-data:
  node3-data:

networks:
  rabbitmq-cluster:
    driver: bridge

Best Practices and Common Pitfalls

After running RabbitMQ in production at scale, these are the most important lessons and the most common mistakes to avoid.

Connection and Channel Management

• One connection per process, one channel per thread/coroutine. Creating a new TCP connection for every publish is the number one performance killer. Connections are expensive (TLS handshake, AMQP handshake). Channels are cheap.
• Never share a channel across threads. Channels are not thread-safe. In Node.js (single-threaded), this is not an issue, but in Python, Java, or Go, use one channel per thread.
• Handle connection drops gracefully. Network partitions and broker restarts will close connections. Implement automatic reconnection with exponential backoff.
• Set heartbeat intervals. Use heartbeat=30 (seconds) to detect dead connections. Without heartbeats, a dead connection may go undetected for hours, holding resources.

Queue Design

• Use quorum queues for anything important. Classic queues on a single node are a single point of failure. Quorum queues replicate across nodes using Raft consensus.
• Set queue length limits. Use x-max-length or x-max-length-bytes to prevent unbounded queue growth. Combine with a dead letter exchange to capture overflow.
• Avoid millions of queues. Each queue consumes memory for its metadata. If you need per-user queues, consider using a single queue with message filtering or a topic exchange.
• Name queues descriptively. Use a naming convention like service.entity.action (e.g., payment.invoice.created). Avoid auto-generated names unless the queue is exclusive/temporary.

Message Design

• Keep messages small. RabbitMQ is optimized for messages under 128KB. For large payloads, store the data in S3/MinIO and send a reference in the message.
• Always set contentType. Use application/json so consumers know how to deserialize.
• Include a messageId and timestamp. These enable idempotency checks and debugging.
• Use headers for metadata. Put retry counts, correlation IDs, and tracing info in headers, not in the message body.

Common Pitfalls

Performance Tuning Checklist

• Use lazy queues (x-queue-mode: lazy) for queues that may accumulate millions of messages — messages are stored on disk instead of memory.
• Batch publishes when possible (publish multiple messages, then waitForConfirms once).
• Use noAck: true only for non-critical, high-throughput streams (logging, metrics).
• Tune Erlang VM: +P 1048576 (max processes), +Q 65536 (max ports).
• Set vm_memory_high_watermark to 0.4-0.7 depending on workload.
• Use SSDs for the RabbitMQ data directory — disk I/O is the primary bottleneck for persistent messages.
• Enable tcp_listen_options.nodelay = true for lower latency.
• Monitor and tune the number of Erlang schedulers (RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="+S 4:4" for 4 cores).

RabbitMQ CLI Quick Reference

# === Server Management ===
rabbitmqctl status                           # node health overview
rabbitmqctl cluster_status                   # cluster membership
rabbitmqctl environment                      # runtime configuration
rabbitmqctl stop_app                         # stop RabbitMQ application (keeps Erlang VM)
rabbitmqctl start_app                        # start RabbitMQ application
rabbitmqctl reset                            # factory reset (removes all data!)

# === User Management ===
rabbitmqctl add_user myuser mypassword       # create user
rabbitmqctl set_user_tags myuser administrator  # grant admin tag
rabbitmqctl set_permissions -p / myuser ".*" ".*" ".*"  # full permissions
rabbitmqctl list_users                       # list all users
rabbitmqctl change_password myuser newpass   # change password
rabbitmqctl delete_user myuser               # delete user

# === Queue Management ===
rabbitmqctl list_queues name messages consumers state  # list queues with details
rabbitmqctl list_queues name messages_ready messages_unacknowledged  # check backlog
rabbitmqctl purge_queue my-queue             # delete all messages from a queue
rabbitmqctl delete_queue my-queue            # delete queue entirely

# === Exchange and Binding Management ===
rabbitmqctl list_exchanges name type durable # list exchanges
rabbitmqctl list_bindings                    # list all bindings

# === Connection and Channel Management ===
rabbitmqctl list_connections name state channels  # list TCP connections
rabbitmqctl list_channels name consumer_count messages_unacknowledged  # list channels
rabbitmqctl close_connection "<connection_pid>" "maintenance"  # force close

# === Plugins ===
rabbitmq-plugins enable rabbitmq_management          # management UI
rabbitmq-plugins enable rabbitmq_prometheus           # Prometheus metrics
rabbitmq-plugins enable rabbitmq_shovel               # cross-cluster message transfer
rabbitmq-plugins enable rabbitmq_delayed_message_exchange  # delayed messages
rabbitmq-plugins list                                # list all plugins

Security Best Practices

• Change default credentials. The default guest/guest user can only connect from localhost. Create a strong admin user and delete or disable the guest user.
• Enable TLS. Encrypt AMQP connections with TLS. Use amqps:// in connection URLs. Configure ssl_options in rabbitmq.conf.
• Use vhost isolation. Give each application/team its own vhost with least-privilege permissions. A user that only publishes should not have configure or consume permissions.
• Limit permissions with regex. Instead of .* (everything), use ^app\\.orders\\..* to restrict access to specific exchange/queue patterns.
• Enable audit logging. Use the rabbitmq_event_exchange plugin to capture connection, channel, queue, and consumer events for security auditing.
• Network segmentation. RabbitMQ should not be directly accessible from the public internet. Use a VPN or private network for client connections and inter-node traffic.
• Rotate credentials. Use short-lived credentials or integrate with HashiCorp Vault for dynamic RabbitMQ credentials.

Frequently Asked Questions

What is RabbitMQ and when should I use it?

RabbitMQ is an open-source message broker implementing AMQP. Use it for asynchronous communication between services, task queues, event-driven architecture, request buffering, and complex message routing. It excels when you need per-message acknowledgment, dead letter handling, and flexible routing patterns.

What are the four exchange types in RabbitMQ?

Direct exchanges route by exact routing key match. Fanout exchanges broadcast to all bound queues. Topic exchanges use wildcard pattern matching with * (one word) and # (zero or more words). Headers exchanges route based on message header attributes using x-matchset to all or any.

How does message acknowledgment work in RabbitMQ?

With manual acknowledgment (noAck: false), consumers must call channel.ack(msg)after processing. Unacked messages are requeued if the consumer disconnects. Use channel.nack()to reject with or without requeue. Automatic acknowledgment removes messages on delivery, risking data loss if the consumer crashes.

What is a dead letter queue and how do I set one up?

A dead letter queue captures messages that are rejected, expire, or overflow. Configure it by settingx-dead-letter-exchange and optionally x-dead-letter-routing-key as queue arguments. Bind a dedicated queue to the dead letter exchange to collect failed messages.

How do I make RabbitMQ messages persistent?

Three requirements: (1) durable queue (durable: true), (2) persistent messages (persistent: true or deliveryMode: 2), and (3) publisher confirms to verify the broker wrote to disk. Without all three, messages may be lost on broker restart.

What is the difference between RabbitMQ and Kafka?

RabbitMQ is a message broker for routing and per-message acknowledgment. Kafka is an event streaming platform for high-throughput ordered logs with replay. Use RabbitMQ for task queues and complex routing up to ~100K msg/s. Use Kafka for event sourcing, stream processing, and throughput beyond 100K msg/s.

How do I set up RabbitMQ clustering?

All nodes must share the same Erlang cookie. Use rabbitmqctl join_cluster to add nodes. Enable quorum queues for data replication across nodes using Raft consensus. Place a load balancer (HAProxy) in front of cluster nodes for client connections.

What is consumer prefetch and how should I configure it?

Prefetch (channel.prefetch(n)) limits unacknowledged messages per consumer. It enables fair dispatch and prevents slow consumers from being overwhelmed. Use 1-5 for CPU-bound tasks, 10-50 for I/O-bound tasks, and 100-250 for high-throughput scenarios. Monitor consumer utilization and adjust accordingly.