Message Queues - The Imperial Messenger System

Welcome, system architect! Consul Caesar.js is now introducing you to the world of Message Queues - a communication system between services that works like an army of Imperial Roman messengers. When one legion needs to send an order to another, it sends a messenger - the message goes into a queue and waits for the recipient.

What Are Message Queues?

Message Queues are a mechanism for asynchronous communication between services. Instead of a direct (synchronous) connection, the producer sends a message to a queue, and the consumer picks it up when ready.

Roman Analogy

1SYNCHRONOUS communication (without queue):
2┌─────────┐         ┌─────────┐
3│ Legat   │ ──────► │ Centurion│
4│ (waits) │ ◄────── │          │
5└─────────┘         └─────────┘
6The Legat must wait for a response!
7
8ASYNCHRONOUS communication (with queue):
9┌─────────┐    ┌─────────┐    ┌─────────────┐
10│ Legat   │ ──►│ Messenger│ ──►│ Centurion   │
11│ (keeps  │    │ (queue)  │    │ (processes) │
12│ working)│    └─────────┘    └─────────────┘
13└─────────┘
14The Legat can do other things!

RabbitMQ - Basics

RabbitMQ is one of the most popular message brokers. It implements the AMQP (Advanced Message Queuing Protocol) protocol.

Installation and Configuration in NestJS

1# Package installation
2npm install @nestjs/microservices amqplib amqp-connection-manager
3
4# Docker for RabbitMQ
5docker run -d --hostname rabbit --name rabbitmq \
6  -p 5672:5672 -p 15672:15672 \
7  rabbitmq:3-management

Module Configuration

1// app.module.ts
2import { Module } from '@nestjs/common';
3import { ClientsModule, Transport } from '@nestjs/microservices';
4
5@Module({
6  imports: [
7    ClientsModule.register([
8      {
9        name: 'ORDERS_SERVICE',
10        transport: Transport.RMQ,
11        options: {
12          urls: ['amqp://localhost:5672'],
13          queue: 'orders_queue',
14          queueOptions: {
15            durable: true, // Queue survives broker restart
16          },
17        },
18      },
19    ]),
20  ],
21})
22export class AppModule {}

Message Producer

1// orders/orders.service.ts
2import { Injectable, Inject } from '@nestjs/common';
3import { ClientProxy } from '@nestjs/microservices';
4
5interface OrderCreatedEvent {
6  orderId: string;
7  items: Array<{ productId: string; quantity: number }>;
8  totalAmount: number;
9  userId: string;
10  timestamp: Date;
11}
12
13@Injectable()
14export class OrdersService {
15  constructor(
16    @Inject('ORDERS_SERVICE') private readonly client: ClientProxy,
17  ) {}
18
19  async createOrder(orderData: CreateOrderDto): Promise<Order> {
20    // 1. Save order in database
21    const order = await this.ordersRepository.create(orderData);
22
23    // 2. Send event to queue (asynchronously!)
24    const event: OrderCreatedEvent = {
25      orderId: order.id,
26      items: order.items,
27      totalAmount: order.total,
28      userId: order.userId,
29      timestamp: new Date(),
30    };
31
32    // emit() - fire-and-forget (doesn't wait for response)
33    this.client.emit('order_created', event);
34
35    // We can also use send() if we need a response
36    // const result = await firstValueFrom(
37    //   this.client.send('process_order', event)
38    // );
39
40    return order;
41  }
42
43  async cancelOrder(orderId: string): Promise<void> {
44    const order = await this.ordersRepository.findById(orderId);
45    order.status = 'cancelled';
46    await this.ordersRepository.save(order);
47
48    // Notify other services about cancellation
49    this.client.emit('order_cancelled', {
50      orderId,
51      reason: 'User requested cancellation',
52      timestamp: new Date(),
53    });
54  }
55}

Message Consumer

1// notifications/notifications.controller.ts
2import { Controller } from '@nestjs/common';
3import { EventPattern, Payload, Ctx, RmqContext } from '@nestjs/microservices';
4
5@Controller()
6export class NotificationsController {
7  constructor(private readonly notificationService: NotificationService) {}
8
9  @EventPattern('order_created')
10  async handleOrderCreated(
11    @Payload() data: OrderCreatedEvent,
12    @Ctx() context: RmqContext,
13  ): Promise<void> {
14    const channel = context.getChannelRef();
15    const originalMsg = context.getMessage();
16
17    try {
18      console.log('Received order_created event:', data.orderId);
19
20      // Send email notification
21      await this.notificationService.sendOrderConfirmation(data);
22
23      // Send push notification
24      await this.notificationService.sendPushNotification(
25        data.userId,
26        'Order accepted!',
27        \`Your order #\${data.orderId} has been accepted.\`
28      );
29
30      // Acknowledge message processing (ACK)
31      channel.ack(originalMsg);
32    } catch (error) {
33      console.error('Processing error:', error);
34
35      // Reject message and put back in queue (NACK)
36      channel.nack(originalMsg, false, true);
37    }
38  }
39
40  @EventPattern('order_cancelled')
41  async handleOrderCancelled(
42    @Payload() data: OrderCancelledEvent,
43    @Ctx() context: RmqContext,
44  ): Promise<void> {
45    const channel = context.getChannelRef();
46    const originalMsg = context.getMessage();
47
48    try {
49      await this.notificationService.sendCancellationEmail(data);
50      channel.ack(originalMsg);
51    } catch (error) {
52      channel.nack(originalMsg, false, true);
53    }
54  }
55}

Advanced Message Queue Patterns

1. Dead Letter Queue (DLQ)

A queue for messages that couldn't be processed:

1// config/rabbitmq.config.ts
2export const rabbitMQConfig = {
3  queue: 'orders_queue',
4  queueOptions: {
5    durable: true,
6    arguments: {
7      'x-dead-letter-exchange': 'dlx.exchange',
8      'x-dead-letter-routing-key': 'dlx.orders',
9      'x-message-ttl': 86400000, // 24h
10      'x-max-retries': 3,
11    },
12  },
13};
14
15// Dead Letter Queue handler
16@Controller()
17export class DeadLetterHandler {
18  @EventPattern('dlx.orders')
19  async handleDeadLetter(
20    @Payload() data: any,
21    @Ctx() context: RmqContext,
22  ): Promise<void> {
23    const channel = context.getChannelRef();
24    const originalMsg = context.getMessage();
25
26    // Log failed messages for monitoring
27    await this.alertService.sendAlert({
28      type: 'DEAD_LETTER',
29      queue: 'orders_queue',
30      message: data,
31      headers: originalMsg.properties.headers,
32    });
33
34    // Save to database for later analysis
35    await this.failedMessagesRepository.save({
36      payload: data,
37      error: originalMsg.properties.headers['x-death'],
38      createdAt: new Date(),
39    });
40
41    channel.ack(originalMsg);
42  }
43}

2. Retry with Exponential Backoff

1// common/retry.decorator.ts
2import { Injectable } from '@nestjs/common';
3
4@Injectable()
5export class MessageRetryService {
6  private readonly maxRetries = 5;
7  private readonly baseDelay = 1000; // 1 second
8
9  async processWithRetry<T>(
10    handler: () => Promise<T>,
11    context: RmqContext,
12  ): Promise<T> {
13    const channel = context.getChannelRef();
14    const originalMsg = context.getMessage();
15    const retryCount = this.getRetryCount(originalMsg);
16
17    try {
18      const result = await handler();
19      channel.ack(originalMsg);
20      return result;
21    } catch (error) {
22      if (retryCount < this.maxRetries) {
23        // Calculate delay with exponential backoff
24        const delay = this.baseDelay * Math.pow(2, retryCount);
25
26        console.log(\`Retry \${retryCount + 1}/\${this.maxRetries} in \${delay}ms\`);
27
28        // Reject and schedule reprocessing
29        await this.scheduleRetry(originalMsg, delay, retryCount + 1);
30        channel.ack(originalMsg); // ACK original, new one goes to queue
31      } else {
32        // Exceeded limit - send to DLQ
33        channel.nack(originalMsg, false, false);
34      }
35      throw error;
36    }
37  }
38
39  private getRetryCount(msg: any): number {
40    return msg.properties.headers?.['x-retry-count'] || 0;
41  }
42
43  private async scheduleRetry(
44    originalMsg: any,
45    delay: number,
46    retryCount: number,
47  ): Promise<void> {
48    // Use delayed message plugin or separate queue with TTL
49    await this.rabbitClient.emit('orders_queue_delayed', {
50      payload: JSON.parse(originalMsg.content.toString()),
51      headers: {
52        'x-retry-count': retryCount,
53        'x-delay': delay,
54      },
55    });
56  }
57}

3. Saga Pattern - Distributed Transactions

1// sagas/order.saga.ts
2import { Injectable } from '@nestjs/common';
3
4interface SagaStep {
5  name: string;
6  execute: () => Promise<void>;
7  compensate: () => Promise<void>;
8}
9
10@Injectable()
11export class OrderSaga {
12  private executedSteps: SagaStep[] = [];
13
14  async execute(orderId: string): Promise<void> {
15    const steps: SagaStep[] = [
16      {
17        name: 'reserve_inventory',
18        execute: () => this.inventoryService.reserve(orderId),
19        compensate: () => this.inventoryService.release(orderId),
20      },
21      {
22        name: 'process_payment',
23        execute: () => this.paymentService.charge(orderId),
24        compensate: () => this.paymentService.refund(orderId),
25      },
26      {
27        name: 'create_shipment',
28        execute: () => this.shippingService.createShipment(orderId),
29        compensate: () => this.shippingService.cancelShipment(orderId),
30      },
31      {
32        name: 'send_confirmation',
33        execute: () => this.notificationService.sendConfirmation(orderId),
34        compensate: () => this.notificationService.sendCancellation(orderId),
35      },
36    ];
37
38    try {
39      for (const step of steps) {
40        console.log(\`Executing step: \${step.name}\`);
41        await step.execute();
42        this.executedSteps.push(step);
43      }
44
45      // All steps completed - emit success
46      this.eventBus.emit('order_saga_completed', { orderId });
47    } catch (error) {
48      console.error('Saga failed, starting compensation...');
49      await this.compensate();
50
51      this.eventBus.emit('order_saga_failed', {
52        orderId,
53        error: error.message,
54        failedStep: this.executedSteps.length,
55      });
56
57      throw error;
58    }
59  }
60
61  private async compensate(): Promise<void> {
62    // Execute compensation in reverse order
63    for (const step of this.executedSteps.reverse()) {
64      try {
65        console.log(\`Compensating step: \${step.name}\`);
66        await step.compensate();
67      } catch (compensationError) {
68        console.error(\`Compensation failed for \${step.name}:\`, compensationError);
69        // Log to alerts - requires manual intervention
70      }
71    }
72  }
73}

Apache Kafka - For Large Scale

Kafka is a data streaming platform, ideal for large volumes:

1// kafka/kafka.module.ts
2import { Module } from '@nestjs/common';
3import { ClientsModule, Transport } from '@nestjs/microservices';
4
5@Module({
6  imports: [
7    ClientsModule.register([
8      {
9        name: 'KAFKA_SERVICE',
10        transport: Transport.KAFKA,
11        options: {
12          client: {
13            clientId: 'orders-service',
14            brokers: ['localhost:9092'],
15          },
16          consumer: {
17            groupId: 'orders-consumer-group',
18          },
19        },
20      },
21    ]),
22  ],
23})
24export class KafkaModule {}
25
26// Kafka Producer
27@Injectable()
28export class EventProducerService {
29  constructor(
30    @Inject('KAFKA_SERVICE') private readonly kafka: ClientProxy,
31  ) {}
32
33  async publishOrderEvent(event: OrderEvent): Promise<void> {
34    await this.kafka.emit('orders.events', {
35      key: event.orderId, // Partitioning key
36      value: event,
37      headers: {
38        'event-type': event.type,
39        'event-version': '1.0',
40        'correlation-id': event.correlationId,
41      },
42    });
43  }
44}
45
46// Kafka Consumer
47@Controller()
48export class OrderEventsConsumer {
49  @EventPattern('orders.events')
50  async handleOrderEvent(
51    @Payload() message: any,
52    @Ctx() context: KafkaContext,
53  ): Promise<void> {
54    const { offset } = context.getMessage();
55    const partition = context.getPartition();
56    const topic = context.getTopic();
57
58    console.log(\`Processing message from \${topic}[\${partition}] offset \${offset}\`);
59
60    // Process event
61    await this.processEvent(message.value);
62
63    // Kafka automatically commits offset after successful processing
64  }
65}

Queue Monitoring

1// monitoring/queue-monitor.service.ts
2@Injectable()
3export class QueueMonitorService {
4  private readonly metrics = new Map<string, QueueMetrics>();
5
6  @Cron('*/30 * * * * *') // Every 30 seconds
7  async collectMetrics(): Promise<void> {
8    const queues = ['orders_queue', 'notifications_queue', 'payments_queue'];
9
10    for (const queue of queues) {
11      const stats = await this.rabbitMQAdmin.getQueueStats(queue);
12
13      const metrics: QueueMetrics = {
14        queue,
15        messageCount: stats.messages,
16        consumerCount: stats.consumers,
17        publishRate: stats.message_stats?.publish_details?.rate || 0,
18        deliverRate: stats.message_stats?.deliver_details?.rate || 0,
19        timestamp: new Date(),
20      };
21
22      this.metrics.set(queue, metrics);
23
24      // Alert if queue is growing too fast
25      if (metrics.messageCount > 10000) {
26        await this.alertService.send({
27          severity: 'WARNING',
28          message: \`Queue \${queue} has \${metrics.messageCount} pending messages\`,
29        });
30      }
31
32      // Alert if no consumers
33      if (metrics.consumerCount === 0) {
34        await this.alertService.send({
35          severity: 'CRITICAL',
36          message: \`Queue \${queue} has no active consumers!\`,
37        });
38      }
39    }
40  }
41}

Best Practices

1. Idempotency

1// Always check if message has already been processed
2@EventPattern('order_created')
3async handleOrder(@Payload() data: OrderEvent): Promise<void> {
4  const alreadyProcessed = await this.cache.get(\`processed:\${data.eventId}\`);
5
6  if (alreadyProcessed) {
7    console.log(\`Event \${data.eventId} already processed, skipping\`);
8    return;
9  }
10
11  await this.processOrder(data);
12  await this.cache.set(\`processed:\${data.eventId}\`, true, 86400); // 24h TTL
13}

2. Message Versioning

1interface MessageEnvelope<T> {
2  version: string;
3  timestamp: Date;
4  correlationId: string;
5  payload: T;
6}
7
8// Handler supporting different versions
9@EventPattern('order_created')
10async handleOrder(@Payload() envelope: MessageEnvelope<any>): Promise<void> {
11  switch (envelope.version) {
12    case '1.0':
13      await this.handleV1(envelope.payload);
14      break;
15    case '2.0':
16      await this.handleV2(envelope.payload);
17      break;
18    default:
19      throw new Error(\`Unsupported message version: \${envelope.version}\`);
20  }
21}

Message Queues are the foundation of scalable distributed systems. Just as the messengers of the Roman Empire ensured communication between legions across vast distances, message queues ensure reliable communication between microservices!