RabbitMQ with Node.js

Node.js applications connect to RabbitMQ using the amqplib library, which provides a full AMQP 0-9-1 implementation for both callback-based and promise/async-await styles. Its lightweight design and non-blocking I/O make it a natural fit for Node.js event-driven applications.

Installing amqplib

npm install amqplib

Connecting to RabbitMQ

const amqp = require('amqplib');

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

For custom credentials and vhost:

const connection = await amqp.connect({
  protocol: 'amqp',
  hostname: 'rabbitmq.example.com',
  port: 5672,
  username: 'myuser',
  password: 'mypassword',
  vhost: '/production'
});

Publishing a Message

const amqp = require('amqplib');

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

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

  const message = JSON.stringify(order);
  channel.sendToQueue(queue, Buffer.from(message), {
    persistent: true,          // delivery_mode = 2
    contentType: 'application/json'
  });

  console.log(`Order published: ${message}`);
  await channel.close();
  await connection.close();
}

publishOrder({ orderId: 1001, product: 'Keyboard', qty: 2 });

Consuming Messages

const amqp = require('amqplib');

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

  const queue = 'orders';
  await channel.assertQueue(queue, { durable: true });
  await channel.prefetch(1);  // fair dispatch

  console.log('Waiting for orders...');

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

    const order = JSON.parse(msg.content.toString());
    console.log(`Processing order ${order.orderId}`);

    try {
      await processOrder(order);
      channel.ack(msg);             // success
    } catch (err) {
      console.error(`Failed: ${err.message}`);
      channel.nack(msg, false, true); // requeue on error
    }
  });
}

startConsumer().catch(console.error);

Working with Exchanges

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

  // Declare a topic exchange
  await ch.assertExchange('app-events', 'topic', { durable: true });

  // Declare a queue and bind with a pattern
  const q = await ch.assertQueue('order-events', { durable: true });
  await ch.bindQueue(q.queue, 'app-events', 'order.#');

  // Publish to the exchange
  ch.publish(
    'app-events',
    'order.placed',
    Buffer.from(JSON.stringify({ orderId: 2001 })),
    { persistent: true, contentType: 'application/json' }
  );

  console.log('Event published to topic exchange');
}

Publisher Confirms

async function publishWithConfirm(message) {
  const conn = await amqp.connect('amqp://localhost');
  const ch = await conn.createConfirmChannel();  // confirm channel

  await ch.assertQueue('critical-tasks', { durable: true });

  return new Promise((resolve, reject) => {
    ch.sendToQueue(
      'critical-tasks',
      Buffer.from(message),
      { persistent: true },
      (err, ok) => {
        if (err) {
          reject(new Error(`Message nacked: ${err.message}`));
        } else {
          resolve('Message confirmed by broker');
        }
      }
    );
  });
}

publishWithConfirm('{"task":"send-invoice"}')
  .then(console.log)
  .catch(console.error);

Fanout Exchange for Pub/Sub

// Publisher
async function publishNews(newsItem) {
  const conn = await amqp.connect('amqp://localhost');
  const ch = await conn.createChannel();
  await ch.assertExchange('news', 'fanout', { durable: false });
  ch.publish('news', '', Buffer.from(JSON.stringify(newsItem)));
  await ch.close();
  await conn.close();
}

// Subscriber
async function subscribeToNews() {
  const conn = await amqp.connect('amqp://localhost');
  const ch = await conn.createChannel();
  await ch.assertExchange('news', 'fanout', { durable: false });

  // Exclusive, auto-delete temporary queue
  const q = await ch.assertQueue('', { exclusive: true });
  await ch.bindQueue(q.queue, 'news', '');

  ch.consume(q.queue, (msg) => {
    if (msg) {
      console.log(`News received: ${msg.content.toString()}`);
      ch.ack(msg);
    }
  });
}

Error Handling and Connection Recovery

amqplib does not include automatic reconnection. Implement it manually:

async function connectWithRetry(url, retries = 5, delay = 5000) {
  for (let i = 0; i < retries; i++) {
    try {
      const conn = await amqp.connect(url);
      console.log('Connected to RabbitMQ');

      conn.on('error', (err) => {
        console.error('Connection error:', err.message);
      });
      conn.on('close', () => {
        console.warn('Connection closed, reconnecting...');
        setTimeout(() => connectWithRetry(url, retries, delay), delay);
      });

      return conn;
    } catch (err) {
      console.warn(`Connection attempt ${i + 1} failed. Retrying in ${delay}ms...`);
      await new Promise(r => setTimeout(r, delay));
    }
  }
  throw new Error('Could not connect to RabbitMQ after maximum retries');
}

Using amqp-connection-manager (Recommended for Production)

The amqp-connection-manager package wraps amqplib with automatic reconnection and channel rebuilding:

npm install amqp-connection-manager

const amqpConnectionManager = require('amqp-connection-manager');

const connection = amqpConnectionManager.connect(['amqp://localhost']);

const channelWrapper = connection.createChannel({
  setup: (channel) => Promise.all([
    channel.assertQueue('orders', { durable: true }),
    channel.prefetch(1)
  ])
});

// Publish (queued internally if connection is down, sent when reconnected)
channelWrapper.sendToQueue('orders',
  Buffer.from(JSON.stringify({ orderId: 3001 })),
  { persistent: true }
);

Summary

Node.js connects to RabbitMQ using amqplib with async/await. Use createChannel() for standard operations and createConfirmChannel() for publisher confirms. Declare queues with assertQueue, exchanges with assertExchange, and bindings with bindQueue. Always call prefetch(1) for fair dispatch and ack() / nack() for manual acknowledgements. For production, use amqp-connection-manager to handle reconnection automatically so your service recovers from broker restarts without manual intervention.