RabbitMQ with Node.js (amqplib) — Complete Guide
In this tutorial, you will learn about RabbitMQ with Node.js (amqplib). We cover key concepts, practical examples, and best practices to help you master this topic.
Use amqplib to integrate RabbitMQ with Node.js for building scalable messaging systems with exchanges, queues, acknowledgements, and connection management.
What You Learn
You will learn how to connect to RabbitMQ with amqplib, publish and consume messages, work with exchanges, handle reconnections, and build event-driven applications.
Why It Matters
Node.js excels at I/O-bound workloads, making it a natural fit for messaging applications. amqplib is the most popular RabbitMQ client for Node.js, providing both callback and promise-based APIs for building real-time systems.
Real-World Use
Doda Browser uses Node.js services with amqplib for real-time notification delivery. When a malware scan completes, a Node.js consumer receives the result and pushes a notification to the browser client via Websocket.
Connecting with amqplib
const amqp = require('amqplib');
async function connect() {
const connection = await amqp.connect('amqp://guest:guest@localhost:5672');
const channel = await connection.createChannel();
console.log('Connected to RabbitMQ');
console.log('Connection:', connection.connection.serverProperties.product);
await channel.close();
await connection.close();
}
connect().catch(console.error);
Expected output:
Connected to RabbitMQ
Connection: RabbitMQ
Publishing Messages
const amqp = require('amqplib');
async function publish() {
const conn = await amqp.connect('amqp://localhost');
const ch = await conn.createChannel();
const exchange = 'nodejs_exchange';
const queue = 'nodejs_queue';
const routingKey = 'nodejs.task';
await ch.assertExchange(exchange, 'topic', { durable: true });
await ch.assertQueue(queue, { durable: true });
await ch.bindQueue(queue, exchange, routingKey);
const message = {
action: 'process_file',
file: '/data/report.pdf',
priority: 'high'
};
ch.publish(exchange, routingKey, Buffer.from(JSON.stringify(message)), {
persistent: true,
contentType: 'application/json',
messageId: 'msg_001'
});
console.log('Published:', message);
await conn.close();
}
publish().catch(console.error);
Expected output:
Published: { action: 'process_file', file: '/data/report.pdf', priority: 'high' }
Consuming Messages
const amqp = require('amqplib');
async function consume() {
const conn = await amqp.connect('amqp://localhost');
const ch = await conn.createChannel();
const queue = 'nodejs_queue';
await ch.assertQueue(queue, { durable: true });
await ch.prefetch(1);
console.log('Waiting for messages...');
ch.consume(queue, async (msg) => {
if (msg !== null) {
const data = JSON.parse(msg.content.toString());
console.log('Message ID:', msg.properties.messageId);
console.log('Action:', data.action);
console.log('File:', data.file);
// Simulate processing
await new Promise(resolve => setTimeout(resolve, 500));
console.log('Processing complete');
ch.ack(msg);
}
});
}
consume().catch(console.error);
Expected output:
Waiting for messages...
Message ID: msg_001
Action: process_file
File: /data/report.pdf
Processing complete
Connection Management with Reconnection
const amqp = require('amqplib');
class RabbitMQClient {
constructor(url = 'amqp://localhost') {
this.url = url;
this.conn = null;
this.ch = null;
this.reconnectDelay = 1000;
}
async connect() {
try {
this.conn = await amqp.connect(this.url);
this.ch = await this.conn.createChannel();
await this.ch.assertQueue('nodejs_robust', { durable: true });
await this.ch.prefetch(1);
console.log('Connected');
this.conn.on('close', () => {
console.log('Connection closed, reconnecting...');
setTimeout(() => this.connect(), this.reconnectDelay);
this.reconnectDelay = Math.min(this.reconnectDelay * 2, 30000);
});
this.reconnectDelay = 1000;
} catch (err) {
console.error('Connection failed:', err.message);
setTimeout(() => this.connect(), this.reconnectDelay);
this.reconnectDelay = Math.min(this.reconnectDelay * 2, 30000);
}
}
async publish(message) {
if (!this.ch) {
console.error('Channel not available');
return;
}
const published = this.ch.publish(
'', 'nodejs_robust',
Buffer.from(message),
{ persistent: true }
);
console.log(published ? 'Published' : 'Channel full');
}
}
const client = new RabbitMQClient();
client.connect();
setTimeout(() => client.publish('hello'), 2000);
Expected output:
Connected
Published
Working with Direct Exchange
const amqp = require('amqplib');
async function directExchangeDemo() {
const conn = await amqp.connect('amqp://localhost');
const ch = await conn.createChannel();
const exchange = 'direct_logs';
await ch.assertExchange(exchange, 'direct', { durable: true });
// Create queues and bind
const queues = ['error_queue', 'warning_queue', 'info_queue'];
for (const q of queues) {
await ch.assertQueue(q, { durable: true });
const severity = q.split('_')[0];
await ch.bindQueue(q, exchange, severity);
}
// Publish messages
const logs = [
{ severity: 'error', message: 'Disk failure' },
{ severity: 'warning', message: 'High memory usage' },
{ severity: 'info', message: 'User login' }
];
for (const log of logs) {
ch.publish(exchange, log.severity, Buffer.from(log.message));
console.log(`Sent [${log.severity}]: ${log.message}`);
}
// Consume from error queue
const errorMsg = await ch.get('error_queue', { noAck: true });
if (errorMsg) {
console.log('Error queue got:', errorMsg.content.toString());
}
await conn.close();
}
directExchangeDemo().catch(console.error);
Expected output:
Sent [error]: Disk failure
Sent [warning]: High memory usage
Sent [info]: User login
Error queue got: Disk failure
Common Mistakes
1. Not Awaiting Channel Operations
amqplib is promise-based. Forgetting await on assertQueue or bindQueue causes race conditions where queues are used before they exist.
2. Not Handling Connection Drops
Without reconnection logic, the Node.js Process loses RabbitMQ connectivity permanently. Implement auto-reconnect with exponential backoff.
3. Not Setting prefetch
Without ch.prefetch(1), Node.js consumers buffer all messages, causing memory growth. Always set prefetch for reliable consumers.
4. Creating Too Many Channels
Each channel has overhead. Reuse channels within a connection. One channel per logical operation is a good rule of thumb.
5. Ignoring publish Confirms
ch.publish() returns a boolean but does not confirm delivery. Use ch.confirmSelect() and ch.waitForConfirms() for reliable publishing.
Practice Questions
1. What is amqplib?
amqplib is the most popular RabbitMQ client library for Node.js. It provides both promise-based and callback-based APIs.
2. How do you set QoS prefetch in Node.js?
Call channel.prefetch(count) before consuming. prefetch(1) ensures one message at a time per consumer.
3. How do you handle reconnection in amqplib?
Listen for the 'close' event on the connection object and recreate both connection and channel with exponential backoff.
4. What is the difference between assertQueue and checkQueue?
assertQueue creates the queue if it does not exist. checkQueue fails if the queue does not exist. Use assertQueue for producers and checkQueue for consumers that expect existing queues.
Challenge
Build a Node.js service using amqplib that implements an RPC pattern: a client publishes scan requests with a correlation ID and reply-to queue, a worker consumes and processes, and the client receives the response asynchronously.
FAQ
Mini Project: Notification Service
const amqp = require('amqplib');
class NotificationService {
constructor() {
this.conn = null;
this.ch = null;
}
async connect() {
this.conn = await amqp.connect('amqp://localhost');
this.ch = await this.conn.createChannel();
await this.ch.assertExchange('notifications', 'topic', { durable: true });
await this.ch.prefetch(1);
console.log('Notification service ready');
}
async listen(severities) {
const q = await this.ch.assertQueue('', { exclusive: true });
for (const severity of severities) {
await this.ch.bindQueue(q.queue, 'notifications', severity);
}
console.log(`Listening for: ${severities.join(', ')}`);
this.ch.consume(q.queue, (msg) => {
if (msg !== null) {
const notification = JSON.parse(msg.content.toString());
console.log(`[${msg.fields.routingKey}] ${notification.message}`);
this.ch.ack(msg);
}
});
}
async send(severity, message) {
this.ch.publish(
'notifications',
severity,
Buffer.from(JSON.stringify({ message, timestamp: Date.now() })),
{ persistent: true }
);
console.log(`Sent [${severity}]: ${message}`);
}
}
async function demo() {
const service = new NotificationService();
await service.connect();
await service.listen(['error', 'warning', 'info']);
await service.send('error', 'Database connection failed');
await service.send('warning', 'Memory usage 85%');
await service.send('info', 'User logged in');
}
demo().catch(console.error);
Expected output:
Notification service ready
Listening for: error, warning, info
Sent [error]: Database connection failed
[error] Database connection failed
Sent [warning]: Memory usage 85%
[warning] Memory usage 85%
Sent [info]: User logged in
[info] User logged in
What's Next
Now that you understand RabbitMQ with Node.js, explore RabbitMQ security for production hardening, then build the mini project: notification system to apply everything you learned.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro