The Ultimate RabbitMQ Cheatsheet

# ── Install & Run RabbitMQ ──
# Docker (recommended)
docker run -d --name rabbitmq   -p 5672:5672   -p 15672:15672   rabbitmq:3-management

# Management UI: http://localhost:15672 (guest/guest)

# ── Key Concepts ──
# Producer:  sends messages to an exchange
# Exchange:  receives messages, routes to queues
# Queue:     stores messages until consumed
# Consumer:  receives messages from queues
# Binding:   links exchange to queue with a routing key

# ── rabbitmqctl Commands ──
rabbitmqctl status
rabbitmqctl list_queues name messages consumers
rabbitmqctl list_exchanges name type
rabbitmqctl list_bindings
rabbitmqctl list_users
rabbitmqctl add_user admin StrongPass123
rabbitmqctl set_user_tags admin administrator
rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"
rabbitmqctl delete_user guest
rabbitmqctl reset          # delete all data
rabbitmqctl force_reset    # force reset (cluster)

const amqp = require('amqplib/callback_api');

// ── Connect to RabbitMQ ──
amqp.connect('amqp://localhost', (err, connection) => {
  if (err) throw err;
  connection.createChannel((err, channel) => {
    if (err) throw err;

    const exchange = 'orders';
    const routingKey = 'order.created';
    const msg = JSON.stringify({ orderId: 123, amount: 99.99 });

    // Assert exchange (create if not exists)
    channel.assertExchange(exchange, 'direct', { durable: true });

    // Publish message
    channel.publish(exchange, routingKey, Buffer.from(msg), {
      persistent: true,          // survive broker restart
      contentType: 'application/json',
      messageId: 'msg-001',
      timestamp: Date.now(),
      headers: { source: 'api' },
    });

    console.log('Published:', routingKey, msg);
    setTimeout(() => connection.close(), 500);
  });
});

const amqp = require('amqplib/callback_api');

amqp.connect('amqp://localhost', (err, connection) => {
  connection.createChannel((err, channel) => {
    const exchange = 'orders';
    const queue = 'order-processing-queue';
    const routingKey = 'order.created';

    channel.assertExchange(exchange, 'direct', { durable: true });

    // Assert queue with options
    channel.assertQueue(queue, {
      durable: true,            // survive broker restart
      exclusive: false,         // not exclusive to this connection
      autoDelete: false,        // don't auto-delete when no consumers
      deadLetterExchange: 'dlx', // route undelivered messages
      messageTtl: 3600000,       // 1 hour TTL
      maxLength: 10000,          // max messages in queue
    });

    // Bind queue to exchange
    channel.bindQueue(queue, exchange, routingKey);

    // Prefetch: how many unacked messages per consumer
    channel.prefetch(10);

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

    channel.consume(queue, (msg) => {
      const data = JSON.parse(msg.content.toString());
      console.log('Received:', data);

      // Process message...
      processOrder(data)
        .then(() => channel.ack(msg))        // acknowledge
        .catch(() => {
          // Reject with requeue=false → goes to DLX
          channel.nack(msg, false, false);
        });
    }, { noAck: false });  // manual acknowledgement
  });
});

// ── Direct Exchange: Exact Routing Key Match ──
channel.assertExchange('logs', 'direct', { durable: true });

// Producer sends with specific routing key
channel.publish('logs', 'error', Buffer.from('Critical error!'));
channel.publish('logs', 'warning', Buffer.from('Disk space low'));
channel.publish('logs', 'info', Buffer.from('User logged in'));

// Consumer binds to specific routing keys
channel.bindQueue('error-queue', 'logs', 'error');
channel.bindQueue('all-logs-queue', 'logs', 'error');
channel.bindQueue('all-logs-queue', 'logs', 'warning');
channel.bindQueue('all-logs-queue', 'logs', 'info');

// ── Topic Exchange: Pattern Matching ──
// * = one word, # = zero or more words
channel.assertExchange('events', 'topic', { durable: true });

// Producers
channel.publish('events', 'order.created', Buffer.from('...'));
channel.publish('events', 'order.shipped', Buffer.from('...'));
channel.publish('events', 'user.registered', Buffer.from('...'));
channel.publish('events', 'user.login.success', Buffer.from('...'));
channel.publish('events', 'payment.failed', Buffer.from('...'));

// Consumers: flexible pattern matching
channel.bindQueue('order-queue',    'events', 'order.#');        // all order events
channel.bindQueue('user-queue',     'events', 'user.*');        // user.registered, user.login
channel.bindQueue('user-login-q',   'events', 'user.login.#');  // user.login.success/fail
channel.bindQueue('all-events-q',   'events', '#');              // everything
channel.bindQueue('payment-q',      'events', 'payment.*');     // payment.failed/succeeded

// ── Fanout Exchange: Broadcast to All Queues ──
channel.assertExchange('notifications', 'fanout', { durable: true });

// No routing key needed for fanout!
channel.publish('notifications', '', Buffer.from('New notification!'));

// Multiple queues receive the same message
channel.assertQueue('email-queue', { durable: true });
channel.assertQueue('sms-queue', { durable: true });
channel.assertQueue('push-queue', { durable: true });

channel.bindQueue('email-queue', 'notifications', '');
channel.bindQueue('sms-queue', 'notifications', '');
channel.bindQueue('push-queue', 'notifications', '');

// ── RPC Pattern (Request/Reply) ──
// Server (RPC Worker)
channel.assertQueue('rpc_queue', { durable: true });
channel.prefetch(1);

channel.consume('rpc_queue', (msg) => {
  const n = parseInt(msg.content.toString());
  const result = fibonacci(n);

  channel.sendToQueue(
    msg.properties.replyTo,
    Buffer.from(result.toString()),
    { correlationId: msg.properties.correlationId }
  );
  channel.ack(msg);
});

// Client
function rpcCall(n) {
  return new Promise((resolve) => {
    const correlationId = uuid();
    channel.assertQueue('', { exclusive: true }, (err, q) => {
      channel.consume(q.queue, (msg) => {
        if (msg.properties.correlationId === correlationId) {
          resolve(parseInt(msg.content.toString()));
        }
      }, { noAck: true });

      channel.sendToQueue('rpc_queue',
        Buffer.from(n.toString()),
        { correlationId, replyTo: q.queue }
      );
    });
  });
}

// ── Dead Letter Exchange (DLX) ──
// Messages are routed to DLX when:
// 1. Rejected with requeue=false
// 2. Expired (TTL)
// 3. Queue full (max-length reached)

// Setup main queue with DLX
channel.assertExchange('dlx.exchange', 'direct', { durable: true });
channel.assertQueue('dlx.queue', { durable: true });
channel.bindQueue('dlx.queue', 'dlx.exchange', 'failed');

channel.assertQueue('order-processing', {
  durable: true,
  deadLetterExchange: 'dlx.exchange',
  deadLetterRoutingKey: 'failed',
  messageTtl: 300000,       // 5 min TTL
  maxLength: 50000,          // max queue depth
});

// ── Message Acknowledgement Patterns ──

// Pattern 1: Auto-ack (fire and forget)
channel.consume(queue, (msg) => {
  processMessage(msg);  // no ack needed
}, { noAck: true });

// Pattern 2: Manual ack with error handling
channel.consume(queue, (msg) => {
  processMessage(msg)
    .then(() => {
      channel.ack(msg);              // success → acknowledge
    })
    .catch((err) => {
      if (isRetryable(err)) {
        channel.nack(msg, false, true); // retry → requeue
      } else {
        channel.nack(msg, false, false); // fail → DLX
      }
    });
}, { noAck: false });

// Pattern 3: Batch acknowledgement
channel.consume(queue, (msg) => {
  buffer.push(msg);
  if (buffer.length >= 100) {
    channel.ack(buffer[buffer.length - 1], true); // ack up to this msg
    buffer = [];
  }
}, { noAck: false });

// ── Consumer Cancellation & Recovery ──
channel.on('error', (err) => {
  console.error('Channel error:', err.message);
});
channel.on('close', () => {
  console.log('Channel closed, reconnecting...');
  setTimeout(connect, 2000);
});

# ── RabbitMQ Cluster (3 nodes) ──
version: "3.8"
services:
  rabbit1:
    image: rabbitmq:3-management
    hostname: rabbit1
    environment:
      RABBITMQ_ERLANG_COOKIE: "supersecretcookie"
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: admin123
    ports:
      - "5672:5672"
      - "15672:15672"

  rabbit2:
    image: rabbitmq:3-management
    hostname: rabbit2
    environment:
      RABBITMQ_ERLANG_COOKIE: "supersecretcookie"
    depends_on:
      - rabbit1

  rabbit3:
    image: rabbitmq:3-management
    hostname: rabbit3
    environment:
      RABBITMQ_ERLANG_COOKIE: "supersecretcookie"
    depends_on:
      - rabbit2

# ── Join cluster (run on rabbit2, rabbit3) ──
# rabbitmqctl stop_app
# rabbitmqctl join_cluster rabbit@rabbit1
# rabbitmqctl start_app

# ── Enable federation (cross-cluster messaging) ──
# rabbitmq-plugins enable rabbitmq_federation
# rabbitmq-plugins enable rabbitmq_federation_management

// ── Publisher Confirms (reliable publishing) ──
channel.confirmSelect((err) => {
  if (err) throw err;

  channel.on('ack', (msgId) => {
    console.log('Message confirmed:', msgId);
  });

  channel.on('nack', (msgId) => {
    console.log('Message rejected:', msgId);
    // Retry or log failure
  });
});

// ── Batch Publish with Confirms ──
async function publishBatch(channel, exchange, messages) {
  for (const msg of messages) {
    channel.publish(exchange, msg.routingKey, Buffer.from(JSON.stringify(msg.body)), {
      persistent: true,
      contentType: 'application/json',
    });
  }
  // Wait for all confirms
  await new Promise((resolve) => channel.waitForConfirms(resolve));
}