Message queues enable asynchronous communication between services, making your system more scalable and resilient. RabbitMQ is a popular message broker that implements AMQP (Advanced Message Queuing Protocol).
Why Use Message Queues?
Decoupling: Services communicate without knowing about each other
Scalability: Process messages at your own pace
Reliability: Messages aren't lost if services are down
Flexibility: Add consumers without changing producers
RabbitMQ Concepts
Producer: Sends messages
Queue: Stores messages
Consumer: Receives and processes messages
Exchange: Routes messages to queues
Basic Setup
typescript
import amqp from 'amqplib';
class RabbitMQConnection {
private connection: any;
private channel: any;
async connect(url: string) {
this.connection = await amqp.connect(url);
this.channel = await this.connection.createChannel();
}
getChannel() {
return this.channel;
}
async close() {
await this.channel.close();
await this.connection.close();
}
}Publishing Messages
typescript
class MessagePublisher {
private channel: any;
constructor(channel: any) {
this.channel = channel;
}
async publishMessage(queue: string, message: any) {
await this.channel.assertQueue(queue, { durable: true });
this.channel.sendToQueue(
queue,
Buffer.from(JSON.stringify(message)),
{ persistent: true }
);
}
}
// Usage
const connection = new RabbitMQConnection();
await connection.connect('amqp://localhost');
const publisher = new MessagePublisher(connection.getChannel());
await publisher.publishMessage('tasks', { type: 'email', to: 'user@example.com' });Consuming Messages
typescript
class MessageConsumer {
private channel: any;
constructor(channel: any) {
this.channel = channel;
}
async consumeMessages(queue: string, handler: (message: any) => Promise<void>) {
await this.channel.assertQueue(queue, { durable: true });
this.channel.prefetch(1); // Process one message at a time
this.channel.consume(queue, async (msg: any) => {
if (msg) {
try {
const content = JSON.parse(msg.content.toString());
await handler(content);
this.channel.ack(msg);
} catch (error) {
console.error('Error processing message:', error);
this.channel.nack(msg, false, true); // Requeue message
}
}
});
}
}Exchange Types
Direct Exchange
typescript
async function setupDirectExchange() {
await channel.assertExchange('logs', 'direct', { durable: true });
const queue = await channel.assertQueue('', { exclusive: true });
await channel.bindQueue(queue.queue, 'logs', 'error');
}Topic Exchange
typescript
async function setupTopicExchange() {
await channel.assertExchange('topics', 'topic', { durable: true });
await channel.bindQueue(queue.queue, 'topics', '*.error');
await channel.bindQueue(queue.queue, 'tasks.*');
}Best Practices
- 1Use durable queues to survive broker restarts
- 1Set prefetch to control message delivery rate
- 1Handle errors gracefully with proper ack/nack
- 1Use dead letter queues for failed messages
Conclusion
RabbitMQ provides robust messaging for distributed systems. Start with simple queues and gradually adopt more complex patterns as needed.