RabbitMQ — Node.js

RabbitMQ with Node.js

RabbitMQ is a battle-tested open-source message broker that implements the AMQP 0-9-1 protocol. It sits between producers and consumers, routing messages through a system of exchanges and queues. Unlike simpler in-process event emitters, RabbitMQ is a durable external process: messages survive application restarts, multiple services can consume from the same queue, and the broker tracks acknowledgements. This page covers the core model — exchanges, queues, bindings, and routing keys — then works through a complete Node.js example using the amqplib client.

Core model: exchanges, queues, bindings

Text

Producer                  RabbitMQ broker                    Consumer(s)
   │                                                                │
   │  publish(exchange,      ┌─────────────┐    ┌──────────────┐   │
   │  routingKey, msg)       │   Exchange  │    │    Queue     │   │
   └────────────────────────►│             ├───►│  (durable)   ├───►  worker-1
                             │  (routing   │    │              │
                             │   logic)    │    └──────────────┘   ►  worker-2
                             └─────────────┘
                               │         │
                        Binding: routingKey matches?
                        Yes → message goes to queue
                        No  → message dropped (or returned)

Producers publish to exchanges, not queues — the exchange routes to queues via bindings; this indirection is what makes routing flexible

The key insight is that producers **never publish directly to a queue**. They publish to an **exchange** with a **routing key**. The exchange applies its routing logic and forwards matching messages to one or more queues. Consumers subscribe to queues, not exchanges. This separation means routing rules are decoupled from both producer and consumer — you can add a new queue bound to an existing exchange without changing any producer code. Three exchange types cover most use cases: **direct** (exact routing key match), **fanout** (broadcast to all bound queues), and **topic** (pattern matching with `*` and `#` wildcards).

Exchange types

Exchange type	Routing logic	Use when
direct	Route to queue whose binding key exactly matches routing key	Work queues, task distribution
fanout	Broadcast to ALL bound queues (routing key ignored)	Notifications to multiple services
topic	`*` matches one word, `#` matches zero or more words in the key	Event routing by category (e.g. `order.#`, `*.created`)
headers	Match on message header attributes instead of routing key	Complex routing conditions (rarely used)
(default / nameless)	Routes to queue whose name equals the routing key	Simplest case — direct to a named queue

Installing amqplib

Bash

npm install amqplib
npm install --save-dev @types/amqplib

Producer — publishing a message

import amqp from 'amqplib'

async function publishJob(payload: object) {
  const connection = await amqp.connect(process.env.RABBITMQ_URL ?? 'amqp://localhost')
  const channel = await connection.createChannel()

  const queue = 'jobs'
  // assertQueue is idempotent: creates the queue if it doesn't exist
  await channel.assertQueue(queue, { durable: true })   // survives broker restart

  const message = Buffer.from(JSON.stringify(payload))

  // persistent: 2 → message is written to disk, not just memory
  channel.sendToQueue(queue, message, { persistent: true })

  console.log('Published:', payload)

  // Give the channel time to flush before closing
  await channel.close()
  await connection.close()
}

Always assertQueue before sending — it's idempotent and ensures the queue exists; without it, messages published to a non-existent queue are silently dropped

`assertQueue` declares the queue if it doesn't exist or verifies it matches the existing queue's properties. Calling it on both producer and consumer with the same options prevents the race condition where a producer starts before the consumer has created the queue. The two critical durability options work together: **`durable: true`** on the queue declaration tells RabbitMQ to persist the queue definition across restarts; **`persistent: true`** on the message tells it to write the message to disk. You need both — a durable queue with a non-persistent message will lose the message on restart, and vice versa. Accept some throughput reduction for the write guarantee in any work queue you care about.

Consumer — processing messages

import amqp from 'amqplib'

async function startWorker() {
  const connection = await amqp.connect(process.env.RABBITMQ_URL ?? 'amqp://localhost')
  const channel = await connection.createChannel()

  const queue = 'jobs'
  await channel.assertQueue(queue, { durable: true })

  // prefetch(1): don't give this worker a second message until it acks the first.
  // Without this, RabbitMQ round-robins messages regardless of how busy each worker is.
  channel.prefetch(1)

  console.log('Worker waiting for messages...')

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

    try {
      const payload = JSON.parse(msg.content.toString())
      await processJob(payload) // your business logic

      channel.ack(msg)          // ✅ success — remove from queue
    } catch (err) {
      console.error('Job failed:', err)
      // requeue: false → send to dead-letter queue instead of infinite retry loop
      channel.nack(msg, false, false)
    }
  })
}

async function processJob(payload: unknown) {
  // ... do the actual work
  console.log('Processing:', payload)
}

Ack only after processing completes — if you ack on receipt and then crash, the work is permanently lost; nack with requeue:false for permanently failing messages

The acknowledgement model is everything. The default delivery mode is **at-least-once**: RabbitMQ holds a message as unacknowledged until it receives an `ack`. If the consumer crashes before acking, the broker redelivers to another worker. This is the safety net — but it only works if you **ack after processing and persisting the result**, not on receipt. If you call `channel.ack(msg)` as the first line of your handler, any crash between that and the actual work causes permanent data loss. On the failure path, `channel.nack(msg, false, false)` (requeue=false) routes the message to the **dead-letter exchange** (if configured) rather than immediately requeueing, preventing a single bad message from looping forever and starving other messages. Set up a DLQ and alerting on it so permanently failing messages are visible.

Persistent connection with reconnection

// In production, don't open/close per message — maintain one long-lived connection.
// amqplib doesn't auto-reconnect; handle connection errors explicitly:

import amqp, { Connection, Channel } from 'amqplib'

class RabbitMQClient {
  private connection: Connection | null = null
  private channel: Channel | null = null

  async connect(): Promise<void> {
    this.connection = await amqp.connect(process.env.RABBITMQ_URL!)
    this.channel = await this.connection.createChannel()

    this.connection.on('error', (err) => {
      console.error('RabbitMQ connection error:', err)
      this.reconnect()
    })
    this.connection.on('close', () => {
      console.warn('RabbitMQ connection closed — reconnecting')
      this.reconnect()
    })
  }

  private reconnect(): void {
    setTimeout(() => this.connect(), 5000)   // simple backoff; use exponential in prod
  }

  getChannel(): Channel {
    if (!this.channel) throw new Error('RabbitMQ not connected')
    return this.channel
  }
}

export const rabbit = new RabbitMQClient()

Fanout exchange — broadcasting events

// Producer: publish to a fanout exchange — all bound queues receive a copy
const exchange = 'user.events'
await channel.assertExchange(exchange, 'fanout', { durable: true })

channel.publish(exchange, '', Buffer.from(JSON.stringify({ type: 'user.created', userId: '123' })))
// routing key '' is ignored for fanout

// Consumer: create an exclusive, auto-delete queue and bind it
const q = await channel.assertQueue('', { exclusive: true })  // server-generated name
await channel.bindQueue(q.queue, exchange, '')

channel.consume(q.queue, (msg) => {
  if (!msg) return
  const event = JSON.parse(msg.content.toString())
  // handle event...
  channel.ack(msg)
})

Fanout queues are typically exclusive and auto-delete — each consumer gets its own transient queue bound to the exchange; only messages published while connected are received

The fanout pattern models **pub/sub**: each subscriber gets its own queue, and the exchange delivers a copy of every message to each. Using `exclusive: true` + empty string name lets RabbitMQ generate a unique queue name per consumer. The queue is automatically deleted when the consumer disconnects. This is appropriate for notifications (email service, analytics service) where each needs its own copy. For **work queues** (competing consumers processing jobs exactly once), use a shared named queue — multiple workers bind to the same queue and RabbitMQ distributes messages round-robin. The pattern you choose depends entirely on delivery semantics: one copy per consumer (fanout), or one copy total (work queue).

Key options summary

Option	Where set	What it does
`durable: true`	assertQueue / assertExchange	Survives broker restart
`persistent: true`	sendToQueue / publish	Message written to disk (needs durable queue too)
`prefetch(N)`	channel	Max unacked messages per worker — prevents overloading slow workers
`exclusive: true`	assertQueue	Only this connection can use the queue; auto-deleted on disconnect
`autoDelete: true`	assertQueue	Queue deleted when last consumer disconnects
`deadLetterExchange`	assertQueue args	Where nacked / expired messages go

For high-throughput ordered event streams and log replay, see [Kafka](/nodejs/kafka).