Message Queues with RabbitMQ

Reviewed & published by Brayan K

A message queue lets one service hand off work to another without waiting for it. The producer drops a message into a broker like RabbitMQ, and a consumer picks it up whenever it is ready — so the two services are decoupled in time, speed, and availability.

Part of the free Node.js course at LearnCodingFast — hands-on lessons with worked examples and the output they print, plus practice exercises and a quick quiz.

In this lesson you'll learn the core AMQP concepts — connections, channels, queues, exchanges, and acknowledgements — and use the amqplib client to build a durable work queue and a pub/sub broadcast in Node.js.

What You'll Learn in This Lesson

• Why queues decouple producers from consumers
• Core AMQP concepts: connection, channel, queue, exchange
• Connecting with amqplib and creating a channel
• assertQueue, sendToQueue, and consume with ack/nack
• Building a fair-dispatch work queue with prefetch
• Pub/sub broadcasting with a fanout exchange

1️⃣ Connect, Channel, and Declare a Queue

Every amqplib program starts the same way: open a connection to the broker, create a lightweight channel over it, and assert the queue you'll use. A channel is where nearly all work happens — you rarely touch the raw connection again. assertQueue is idempotent, so it's safe to call on every startup.

// A broker is not running here, so amqplib cannot connect and none of the
// blocks above can execute. What a queue GUARANTEES needs no broker, and
// that is the part worth being sure of: fair dispatch with prefetch,
// acknowledgement, and redelivery of anything a worker did not ack.
// Here is that contract in about thirty lines, so you can watch it hold.

class Queue {
  constructor() { this.ready = []; this.unacked = new Map(); this.nextTag = 1; }
  send(job) { this.ready.push(job); }

  // prefetch(1): a worker is only ever holding one unacked message, which is
  // what stops a slow job piling up while another worker sits idle.
  deliverTo(worker) {
    if (this.ready.length === 0) return null;
    if ([...this.unacked.values()].some((u) => u.worker === worker)) return null;
    const tag = this.nextTag++;
    const job = this.ready.shift();
    this.unacked.set(tag, { job, worker });
    return { tag, job };
  }

  ack(tag) { this.unacked.delete(tag); }

  // A worker that dies without acking: the broker puts its job back at the
  // front of the queue. Nothing is lost, and nothing needed to be persisted
  // for that to be true.
  workerDied(worker) {
    for (const [tag, u] of [...this.unacked]) {
      if (u.worker === worker) { this.unacked.delete(tag); this.ready.unshift(u.job); }
    }
  }
}

const q = new Queue();
for (const id of [1, 2, 3]) q.send({ id, type: "resize-image" });

const a = q.deliverTo("worker-A");
const b = q.deliverTo("worker-B");
console.log("A got job", a.job.id, "| B got job", b.job.id);

const again = q.deliverTo("worker-A");
console.log("A asks for another before acking:", again);

q.ack(a.tag);
const third = q.deliverTo("worker-A");
console.log("after A acks, A got job", third.job.id);

console.log("B crashes before acking job", b.job.id);
q.workerDied("worker-B");
const redelivered = q.deliverTo("worker-B");
console.log("B restarts and is handed job", redelivered.job.id, "again");
console.log("jobs still waiting:", q.ready.length, "| unacked:", q.unacked.size);

// The second line is the one to sit with. A asked for more work and got
// null, not job 2 — that is prefetch(1) refusing to hand a worker a second
// message while the first is unacked. And job 2 came back after B died
// because it was never acked. Both are the reason you ack AFTER the work is
// done, not when the message arrives.

// install: npm install amqplib
// A message queue lets one service (a PRODUCER) hand off work to another
// service (a CONSUMER) without waiting for it. The broker (RabbitMQ) holds
// the messages in between, so the two sides never have to be up at once.
import amqp from "amqplib";

// 1) Open a TCP connection to the broker.
const connection = await amqp.connect("amqp://localhost");

// 2) Open a CHANNEL. Almost every AMQP operation happens on a channel,
//    a lightweight virtual connection multiplexed over the real one.
const channel = await connection.createChannel();

// 3) Declare the queue we will use. assertQueue is idempotent: it creates
//    the queue if it is missing and does nothing if it already exists.
//    durable: true means the queue definition survives a broker restart.
await channel.assertQueue("tasks", { durable: true });

console.log("Connected and queue 'tasks' is ready");

Note durable: true on the queue — that's what lets the queue itself survive a broker restart. Durability is opt-in because not every queue needs to outlive a crash, and durable queues cost a little extra disk work.

2️⃣ Produce and Consume (with Acknowledgements)

A producer drops bytes onto the queue with sendToQueue and moves on. A consumer registers a callback with consume. The crucial part is acknowledgement: after you finish a message you call ack so the broker removes it. If your consumer crashes before acking, the broker redelivers the message — that's at-least-once delivery.

// PRODUCER: send a job into the queue and exit.
// persistent: true marks the message deliveryMode 2 so it is written to disk;
// combined with a durable queue, the job survives a broker restart.
const job = { id: 42, type: "resize-image", file: "cat.png" };

channel.sendToQueue(
  "tasks",
  Buffer.from(JSON.stringify(job)), // AMQP payloads are raw bytes
  { persistent: true }
);
console.log("Sent job", job.id);

// CONSUMER (runs in a separate process / worker):
// prefetch(1) = fair dispatch: only hand this worker one unacked message at a
// time, so a slow job does not pile up while another worker sits idle.
await channel.prefetch(1);

await channel.consume("tasks", async (msg) => {
  if (msg === null) return;            // consumer was cancelled by the broker
  const data = JSON.parse(msg.content.toString());
  try {
    await doWork(data);                // your actual processing
    channel.ack(msg);                  // success: remove it from the queue
  } catch (err) {
    // requeue: false sends it to a dead-letter queue instead of looping forever
    channel.nack(msg, false, false);
  }
});

prefetch(1) gives you fair dispatch: the broker won't hand a worker a second message until it acks the first, so a slow job never piles up on one worker while another sits idle. On failure, nack(msg, false, false) rejects the message without requeuing it, so it can route to a dead-letter queue instead of looping forever.

• • sendToQueue(queue, buffer, opts) — publish bytes to a queue.
• • consume(queue, cb) — register a handler that runs per message.
• • ack(msg) — success; remove the message from the queue.
• • nack(msg, allUpTo, requeue) — failure; requeue or dead-letter.

3️⃣ Pub/Sub with a Fanout Exchange

A work queue makes consumers compete for messages. Pub/sub does the opposite: every subscriber gets its own copy. To do that, producers publish to an exchange instead of a queue, and each subscriber binds its own private queue to that exchange. A fanout exchange copies each message to every bound queue.

// PUB/SUB with a fanout exchange: one published message is copied to EVERY
// subscriber. This is different from a work queue, where workers COMPETE for
// each message. With fanout, everyone gets their own copy.

// --- Publisher ---
await channel.assertExchange("logs", "fanout", { durable: false });
channel.publish("logs", "", Buffer.from("user signed up")); // "" = no routing key

// --- Subscriber (run once per subscriber process) ---
// An exclusive queue with a server-generated name is private to this
// connection and auto-deletes when the subscriber disconnects.
const { queue } = await channel.assertQueue("", { exclusive: true });
await channel.bindQueue(queue, "logs", "");  // bind it to the fanout exchange

await channel.consume(queue, (msg) => {
  if (msg) console.log("[subscriber] got:", msg.content.toString());
}, { noAck: true }); // noAck: true = fire-and-forget, fine for broadcast logs

Each subscriber declares an exclusive queue with an empty name, so the broker generates a unique name and auto-deletes it when the subscriber disconnects. Bind it to the exchange and every broadcast lands in it. Swap fanout for topic when subscribers only want messages matching a routing-key pattern.

Your turn. Fill in the two ___ blanks so the work queue and its messages both survive a broker restart.

// 🎯 YOUR TURN — fill in the ___ blanks to finish a durable work queue.
import amqp from "amqplib";

const connection = await amqp.connect("amqp://localhost");
const channel = await connection.createChannel();

// Make the queue survive a broker restart:
await channel.assertQueue("emails", { durable: ___ }); // 👉 true or false?

// Send a job that is written to disk:
channel.sendToQueue(
  "emails",
  Buffer.from(JSON.stringify({ to: "a@b.com" })),
  { persistent: ___ }                  // 👉 which flag makes it survive a crash?
);

// ✅ Both blanks should be true: a durable queue AND persistent messages are
//    BOTH required for the job to survive a broker restart.

Mini-Challenge: Add a Second Worker

Start two copies of the consumer from section 2, both consuming the tasks queue. Publish ten jobs from the producer. Because both workers share the queue and use prefetch(1), the jobs are divided between them — a fast worker grabs more than a slow one. Confirm that each job is handled by exactly one worker, never both. That's the competing-consumers pattern that lets you scale throughput just by adding processes.

Common Errors (and the fix)

• Forgetting to ack — Without channel.ack(msg), messages stay unacked, pile up, and get redelivered when the consumer reconnects. Always ack on success or nack on failure.
• Expecting durability for free — A durable queue alone is not enough; the messages must also be published with persistent: true. You need both.
• Publishing strings, not buffers — sendToQueue expects a Buffer. Wrap objects with Buffer.from(JSON.stringify(obj)).
• Confusing work queues with pub/sub — Multiple consumers on one queue share messages. For everyone to get a copy you need an exchange and one queue per subscriber.
• Requeuing a poison message forever — A message that always fails will loop endlessly if you nack with requeue true. Send it to a dead-letter queue instead.

Pro Tips

• 💡 One channel per task: channels are cheap, but they are not thread-safe to share across concurrent publishers. Use a channel per worker rather than one global channel for everything.
• 💡 Turn on publisher confirms for critical messages so the broker explicitly confirms each one was persisted, closing the small window where a published message could be lost.
• 💡 Set a prefetch that matches how heavy your jobs are. prefetch(1) is safest for slow jobs; a higher value improves throughput for fast ones.

📋 Quick Reference — amqplib

🎉 Lesson Complete!

• ✅ Queues decouple producers from consumers in time and speed
• ✅ You work over a channel, not the raw connection
• ✅ assertQueue declares idempotently; sendToQueue publishes bytes
• ✅ ack/nack give at-least-once delivery and safe retries
• ✅ prefetch(1) gives fair dispatch across workers
• ✅ A fanout exchange broadcasts to every subscriber (pub/sub)

Practice quiz

Continue this course

• Previous: Capstone: A REST API
• Next: gRPC Services — Define services with Protocol Buffers and build fast unary and streaming RPCs
• Quick reference: Node.js cheat sheet

Frequently asked questions

What is the difference between a queue and an exchange in RabbitMQ?

A queue is the buffer that actually stores messages until a consumer reads them. An exchange is a router: producers publish to an exchange, never directly to a queue, and the exchange decides which bound queues receive a copy based on its type and routing keys. A direct exchange routes by an exact routing-key match, a fanout exchange copies to every bound queue (broadcast), and a topic exchange matches wildcard patterns. In amqplib you assertExchange, assertQueue, and bindQueue to connect them. The default nameless exchange routes a message to the queue whose name equals the routing key, which is why sendToQueue feels like publishing straight to a queue.

What is the difference between ack and nack?

channel.ack(msg) tells the broker the message was processed successfully, so it can be removed from the queue and never redelivered. channel.nack(msg, allUpTo, requeue) signals failure: if requeue is true the message goes back on the queue to be tried again, and if requeue is false it is discarded or routed to a dead-letter exchange. Until you ack or nack, the broker considers the message in-flight, and if your consumer disconnects it will be redelivered. This manual acknowledgement is what gives you at-least-once delivery and lets you safely retry failed work.

How do I make sure messages are not lost if RabbitMQ restarts?

You need three things together. First, declare the queue with durable: true so the queue definition survives a restart. Second, publish messages with persistent: true (AMQP deliveryMode 2) so the bytes are written to disk rather than only held in memory. Third, use manual acknowledgements so a message is only removed after a consumer confirms it. Even with all three, there is a tiny window during which a freshly published message may not yet be flushed to disk; for stronger guarantees enable publisher confirms, which have the broker explicitly confirm each persisted message.

When should I use a work queue versus pub/sub?

Use a work queue (the competing-consumers pattern) when each message represents a unit of work that should be handled exactly once by one of several interchangeable workers, such as resizing an image or sending an email. Add more workers and they share the load, and prefetch(1) keeps the dispatch fair. Use pub/sub (a fanout or topic exchange with one queue per subscriber) when every interested service needs its own copy of an event, such as broadcasting that a user signed up so the email, analytics, and audit services all react independently. Work queues divide work; pub/sub multiplies notifications.