🐇 RabbitMQ Basics & Setup

What Is RabbitMQ?

RabbitMQ is an open-source message broker that has been quietly running production systems since 2007. It implements AMQP 0-9-1 (the Advanced Message Queuing Protocol), an open standard, so clients exist for virtually every language. It's written in Erlang — a language built for concurrent, distributed, fault-tolerant systems — which is a big part of why RabbitMQ is so dependable.

What sets RabbitMQ apart is its smart routing. Producers don't send messages straight to queues; they send them to an exchange, and the exchange decides — by rules you define — which queues receive a copy. That indirection is what lets one publish reach one worker, or ten services, or none, without the producer changing a line.

Setup with Docker

The fastest, cleanest way to get a broker for development is Docker. This one command pulls RabbitMQ with the management plugin and runs it:

# Port 5672 = AMQP (your apps connect here)
# Port 15672 = the web management dashboard
docker run -d --name rabbitmq \
  -p 5672:5672 -p 15672:15672 \
  rabbitmq:3-management

Give it a few seconds to boot, then open http://localhost:15672 and sign in:

Prefer a native install? Grab Erlang and the server from rabbitmq.com/download, then enable the dashboard with rabbitmq-plugins enable rabbitmq_management. On macOS, brew install rabbitmq works too. For learning, Docker keeps your machine clean.

The management dashboard

The UI at port 15672 is your window into the broker. You'll use it constantly to watch Queues (depth, message rates), inspect Exchanges and their bindings, view live Connections and Channels, and manage users under Admin. When something isn't working, this dashboard is the first place to look.

The RabbitMQ Model

Six terms unlock everything else. Read them once here, and the code below becomes obvious.

Your First Queue

Install the client, then write a publisher and a consumer that talk through a single queue.

npm install amqplib

Publisher — publisher.js

const amqp = require('amqplib');

async function publishMessage() {
  // 1. Open a connection (one per process) and a channel (per task).
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // 2. Declare the queue. assertQueue is idempotent: it creates the queue
  //    if missing, or does nothing if it already exists.
  const queue = 'tasks';
  await channel.assertQueue(queue, { durable: true }); // survive broker restart

  // 3. Build a message. Messages travel as raw bytes, so serialize to a Buffer.
  const message = {
    id: Math.floor(Math.random() * 1000),
    task: 'Process data',
    timestamp: new Date().toISOString(),
  };

  channel.sendToQueue(queue, Buffer.from(JSON.stringify(message)), {
    persistent: true, // write to disk so it isn't lost on restart
  });

  console.log(`[x] Sent: ${JSON.stringify(message)}`);

  // 4. Give the broker a moment to receive, then close cleanly.
  await channel.close();
  await connection.close();
}

publishMessage().catch(console.error);

Consumer — consumer.js

const amqp = require('amqplib');

async function consumeMessages() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const queue = 'tasks';
  await channel.assertQueue(queue, { durable: true }); // must match the publisher

  // Only hand this consumer one unacked message at a time (fair dispatch).
  channel.prefetch(1);

  console.log('[*] Waiting for messages. Press CTRL+C to exit.');

  channel.consume(queue, async (msg) => {
    if (msg === null) return; // queue was deleted / cancelled

    const content = JSON.parse(msg.content.toString());
    console.log(`[x] Received: ${JSON.stringify(content)}`);

    // Simulate real work, then acknowledge so the broker deletes it.
    await new Promise((r) => setTimeout(r, 500));
    console.log(`[x] Done task ${content.id}`);
    channel.ack(msg); // manual ack — the message is gone only after this
  }, { noAck: false }); // manual acknowledgement mode
}

consumeMessages().catch(console.error);

Terminal 1:  node consumer.js
             [*] Waiting for messages. Press CTRL+C to exit.

Terminal 2:  node publisher.js
             [x] Sent: {"id":742,"task":"Process data",...}

Terminal 1:  [x] Received: {"id":742,...}
             [x] Done task 742

Start the consumer first, then run the publisher. Watch the tasks queue in the management UI — you'll see the message count tick up and back down in real time.

Reliability: Durability, Persistence, Acks

By default, a message can vanish in three ways: the broker restarts, the process crashes mid-work, or the message is acked before the work finishes. Three settings close those gaps — and you need all three for end-to-end safety.

ack vs nack

channel.consume(queue, async (msg) => {
  try {
    await doWork(JSON.parse(msg.content.toString()));
    channel.ack(msg);                    // success → broker deletes it
  } catch (err) {
    console.error('Processing failed:', err);
    // nack(msg, allUpTo, requeue). Set requeue=false to send it onward
    // to a dead-letter queue instead of looping forever.
    channel.nack(msg, false, false);
  }
}, { noAck: false });

Fair Dispatch with Prefetch

By default RabbitMQ round-robins messages to consumers the instant they connect — even if one worker is already buried under slow tasks. The result: a fast, idle worker sits waiting while a slow worker's inbox overflows. prefetch fixes this.

// "Don't give me a new message until I've acked my current one."
channel.prefetch(1);

With prefetch(1), the broker holds back new messages from a worker until it acks the one it has. Work naturally flows to whoever is free, and no single consumer hoards a backlog. For high-throughput pipelines a small batch (say prefetch(10)) can improve throughput — measure and tune for your workload.

Exchanges & Routing

Work queues get you far, but the real power is in exchanges. Three types cover nearly everything.

Fanout — broadcast to all

Ignores routing keys entirely; every bound queue gets a copy. Perfect for pub/sub.

// Publisher — broadcast a log line to everyone listening.
const ex = 'logs';
await channel.assertExchange(ex, 'fanout', { durable: false });
channel.publish(ex, '', Buffer.from(JSON.stringify({ msg: 'System update done' })));
//                   ^^ empty routing key — fanout ignores it anyway

// Consumer — each gets its own temporary, auto-deleting queue.
await channel.assertExchange(ex, 'fanout', { durable: false });
const { queue } = await channel.assertQueue('', { exclusive: true });
await channel.bindQueue(queue, ex, '');
channel.consume(queue, (msg) => {
  console.log('[x] Broadcast:', msg.content.toString());
  channel.ack(msg);
});

Direct — exact routing-key match

Delivers to queues whose binding key exactly equals the message's routing key. The classic use is severity-based log routing.

// Publisher — the severity IS the routing key.
const ex = 'logs_direct';
await channel.assertExchange(ex, 'direct', { durable: false });
const severity = process.argv[2] || 'info'; // 'info' | 'warning' | 'error'
channel.publish(ex, severity, Buffer.from(`a ${severity} message`));

// Consumer — subscribe only to the severities you care about.
await channel.assertExchange(ex, 'direct', { durable: false });
const { queue } = await channel.assertQueue('', { exclusive: true });
for (const sev of process.argv.slice(2)) {
  await channel.bindQueue(queue, ex, sev); // bind once per severity
}
channel.consume(queue, (msg) => {
  console.log(`[x] ${msg.fields.routingKey}: ${msg.content.toString()}`);
  channel.ack(msg);
});

Topic — wildcard pattern match

Routing keys are dot-delimited words; bindings use two wildcards: * matches exactly one word, # matches zero or more.

Practice & Quiz

// producer.js — send 10 tasks with varied "work" size.
const amqp = require('amqplib');
(async () => {
  const conn = await amqp.connect('amqp://localhost');
  const ch = await conn.createChannel();
  await ch.assertQueue('work', { durable: true });
  for (let i = 1; i <= 10; i++) {
    const dots = '.'.repeat(i % 4);        // 0-3 "seconds" of work
    ch.sendToQueue('work', Buffer.from(`task ${i}${dots}`), { persistent: true });
    console.log(`[x] Sent task ${i}`);
  }
  await ch.close(); await conn.close();
})();

// worker.js — run this in TWO terminals.
const amqp = require('amqplib');
(async () => {
  const conn = await amqp.connect('amqp://localhost');
  const ch = await conn.createChannel();
  await ch.assertQueue('work', { durable: true });
  ch.prefetch(1);                          // fair dispatch
  ch.consume('work', async (msg) => {
    const body = msg.content.toString();
    const secs = (body.match(/\./g) || []).length;
    console.log(`[x] ${body}`);
    await new Promise((r) => setTimeout(r, secs * 1000));
    console.log(`[x] done: ${body}`);
    ch.ack(msg);                           // ack only after the work
  }, { noAck: false });
})();
// Kill one worker mid-task (Ctrl+C): its unacked message reappears
// on the other worker — proof that manual acks make work safe.

// Consumer A (errors only):   node consumer.js error
// Consumer B (everything):     node consumer.js info warning error
// Publish:                     node publisher.js error
//                              node publisher.js info
//
// Result: A prints only the 'error' line; B prints both.
// Each severity you pass on the consumer becomes a bindQueue call,
// so B has three bindings and A has one.

Best Practices & Pitfalls

✅ Do

• Open one connection per process and many channels on it — channels are cheap, connections are not
• Declare queues and exchanges with assert* on startup so infrastructure is self-creating
• Use durable queues + persistent messages + manual acks for anything you can't afford to lose
• Set prefetch to get fair dispatch across competing consumers
• Configure a dead-letter exchange and reconnect logic before going to production

❌ Don't

• Open a new connection per message — it will crush the broker under load
• Use noAck: true for important work — a crash then loses the message
• Ship the default guest/guest user to production
• Send huge payloads; store big data externally and pass a reference
• Requeue a failing message endlessly with no dead-letter escape hatch

Summary

📚 Additional Resources

• RabbitMQ — Official Tutorials (JavaScript)
• amqplib — npm package & API docs
• CloudAMQP — 13 Common RabbitMQ Mistakes
• RabbitMQ — Production Checklist

🚀 What's Next?

You can now move messages reliably and route them with intent. In the final lesson of this trio, Event-Driven Architecture, you'll stitch several services together with topic exchanges so a single order ripples through payment, inventory, and notifications — no service calling another directly.