Message Queues and Event-Driven Architecture

Challenge

Decouple services, handle high-volume async operations, and ensure reliable message delivery.

Explanation

Message queues enable asynchronous communication between services, improving scalability and reliability.

RabbitMQ Basics

const amqp = require('amqplib');

// Connect to RabbitMQ
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();

// Declare queue
await channel.assertQueue('task_queue', { durable: true });

// Publish message
channel.sendToQueue('task_queue', Buffer.from('Task data'), {
  persistent: true
});

// Consume messages
channel.consume('task_queue', async (msg) => {
  console.log('Received:', msg.content.toString());

  // Process message
  // ...

  // Acknowledge after processing
  channel.ack(msg);
});

await connection.close();

Work Queues (Task Distribution)

// Publisher (send tasks)
async function publishTask(task) {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  await channel.assertQueue('tasks', { durable: true });

  channel.sendToQueue('tasks', Buffer.from(JSON.stringify(task)), {
    persistent: true
  });

  console.log('Task published:', task);
  await connection.close();
}

// Worker (process tasks)
async function startWorker() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  await channel.assertQueue('tasks', { durable: true });

  // Process one task at a time
  channel.prefetch(1);

  channel.consume('tasks', async (msg) => {
    const task = JSON.parse(msg.content.toString());
    console.log('Processing:', task);

    // Simulate work
    await processTask(task);

    channel.ack(msg);
  });

  console.log('Worker started, waiting for tasks...');
}

// Usage
publishTask({ type: 'email', to: 'user@example.com' });
await startWorker();

Redis as Message Queue

const redis = require('redis');

const publisher = redis.createClient();
const subscriber = redis.createClient();

// Publish-Subscribe
publisher.on('connect', () => {
  publisher.publish('notifications', JSON.stringify({
    type: 'user_signup',
    userId: 123
  }));
});

subscriber.on('connect', async () => {
  await subscriber.subscribe('notifications');
});

subscriber.on('message', (channel, message) => {
  const event = JSON.parse(message);
  console.log(`Received on ${channel}:`, event);
});

// Message Queues with Lists
const client = redis.createClient();

// Push to queue
await client.lpush('queue:tasks', JSON.stringify({ task: 'process' }));

// Pop from queue
const task = await client.rpop('queue:tasks');
console.log(task);

// Blocking pop (wait for message)
const [key, task] = await client.blpop('queue:tasks', 0);

Bull: Node.js Queue Library

const Queue = require('bull');

// Create queue
const emailQueue = new Queue('emails', 'redis://localhost:6379');

// Add job
const job = await emailQueue.add(
  { email: 'user@example.com', subject: 'Welcome' },
  { attempts: 3, backoff: { type: 'exponential', delay: 2000 } }
);

// Process jobs
emailQueue.process(async (job) => {
  console.log('Sending email:', job.data);

  // Send email
  await sendEmail(job.data);

  // Update progress
  job.progress(50);

  return { success: true };
});

// Listen to events
emailQueue.on('completed', (job) => {
  console.log('Job completed:', job.id);
});

emailQueue.on('failed', (job, err) => {
  console.log('Job failed:', job.id, err.message);
});

// Get job status
const status = await job.getState();
const progress = await job.progress();

Pub/Sub Pattern

const EventEmitter = require('events');

class EventBus extends EventEmitter {
  publishAsync(event, data) {
    setImmediate(() => {
      this.emit(event, data);
    });
  }
}

const eventBus = new EventBus();

// Multiple subscribers
eventBus.on('order:created', async (order) => {
  console.log('Sending confirmation email...');
  await sendEmail(order);
});

eventBus.on('order:created', async (order) => {
  console.log('Updating inventory...');
  await updateInventory(order);
});

eventBus.on('order:created', async (order) => {
  console.log('Recording analytics...');
  await recordAnalytics(order);
});

// Publish event
eventBus.publishAsync('order:created', {
  id: 1,
  userId: 123,
  items: [...]
});

Error Handling and Retry

const queue = new Queue('tasks', 'redis://localhost');

queue.process(async (job) => {
  try {
    await processTask(job.data);
  } catch (error) {
    // Exponential backoff retry
    if (job.attemptsMade < 3) {
      throw error; // Retry
    } else {
      // Move to dead letter queue
      await queue.add(job.data, {
        attempts: 0
      });
      throw error;
    }
  }
});

queue.on('failed', (job, error) => {
  console.error(`Job ${job.id} failed:`, error.message);
  // Alert monitoring system
  alertMonitoring({
    jobId: job.id,
    error: error.message,
    attempts: job.attemptsMade
  });
});

Best Practices

// 1. Always acknowledge after processing
channel.consume(queue, (msg) => {
  process(msg);
  channel.ack(msg); // Don't forget!
});

// 2. Set timeouts
queue.process(async (job) => {
  return Promise.race([
    processTask(job.data),
    timeout(30000) // 30 second timeout
  ]);
});

// 3. Use dead letter queues
const dlq = new Queue('dlq');
queue.on('failed', (job) => {
  dlq.add(job.data, { retries: 0 });
});

// 4. Monitor queue depth
setInterval(async () => {
  const count = await queue.count();
  metrics.recordQueueDepth(count);
}, 60000);