Construyendo un microservicio orientado a eventos con Node.js

Tutorial paso a paso de un microservicio orientado a eventos con Node.js, RabbitMQ y TypeScript: publicación, consumo, colas muertas e idempotencia.

Publicado el 5 de agosto de 2022Actualizado el 9 de septiembre de 20265 min de lectura

tutorials event-driven microservices nodejs rabbitmq typescript

Los microservicios orientados a eventos se comunican mediante eventos en lugar de llamadas directas a APIs. El servicio A publica un evento ("se realizó un pedido") y cualquier servicio interesado en ese evento lo procesa de forma independiente. Esto desacopla los servicios: el servicio de pedidos no necesita conocer el servicio de inventario, el servicio de notificaciones ni ningún otro consumidor. Se pueden añadir nuevos consumidores sin modificar el productor.

Construiremos un sistema práctico orientado a eventos usando Node.js, TypeScript y RabbitMQ. El sistema procesa eventos de pedidos: cuando se realiza un pedido, consumidores separados actualizan el inventario, envían correos de notificación y registran analíticas.

Configurando RabbitMQ

RabbitMQ es un broker de mensajes que enruta eventos desde los productores hasta los consumidores a través de exchanges y colas.

ymlyaml

# docker-compose.yml
services:
  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"    # AMQP protocol
      - "15672:15672"  # Management UI
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest
    volumes:
      - rabbitmq_data:/var/lib/rabbitmq
 
volumes:
  rabbitmq_data:

tstypescript

// lib/rabbitmq.ts — connection wrapper
import amqp, { type Connection, type Channel } from 'amqplib';
 
let connection: Connection | null = null;
let channel: Channel | null = null;
 
export async function getChannel(): Promise<Channel> {
  if (channel) return channel;
 
  connection = await amqp.connect(
    process.env.RABBITMQ_URL ?? 'amqp://guest:guest@localhost:5672'
  );
 
  connection.on('error', (err) => {
    console.error('RabbitMQ connection error:', err);
    channel = null;
    connection = null;
  });
 
  channel = await connection.createChannel();
 
  // Prefetch: process one message at a time per consumer
  await channel.prefetch(1);
 
  return channel;
}
 
export async function closeConnection(): Promise<void> {
  if (channel) await channel.close();
  if (connection) await connection.close();
  channel = null;
  connection = null;
}

Definiendo eventos

Los eventos deben ser autocontenidos: cada consumidor debe poder procesar el evento sin hacer llamadas adicionales a APIs.

tstypescript

// events/types.ts
interface BaseEvent {
  eventId: string;       // Unique ID for idempotency
  eventType: string;     // Event name
  timestamp: string;     // ISO 8601
  version: number;       // Schema version
  source: string;        // Which service published this
}
 
interface OrderPlacedEvent extends BaseEvent {
  eventType: 'order.placed';
  data: {
    orderId: string;
    customerId: string;
    customerEmail: string;
    items: {
      productId: string;
      quantity: number;
      unitPrice: number;
    }[];
    total: number;
    currency: string;
  };
}
 
interface OrderCancelledEvent extends BaseEvent {
  eventType: 'order.cancelled';
  data: {
    orderId: string;
    customerId: string;
    reason: string;
  };
}
 
type OrderEvent = OrderPlacedEvent | OrderCancelledEvent;

tstypescript

// ❌ Anemic events — consumers need to call back to the producer
interface BadOrderEvent {
  orderId: string;  // Consumer must call GET /orders/123 for details
}
// This creates coupling: the consumer depends on the producer's API
 
// ✅ Rich events — self-contained with all necessary data
interface GoodOrderEvent extends BaseEvent {
  eventType: 'order.placed';
  data: {
    orderId: string;
    customerId: string;
    customerEmail: string;
    items: { productId: string; quantity: number; unitPrice: number }[];
    total: number;
  };
}
// Consumer has everything it needs — no callbacks required

Publicando eventos

El productor publica eventos en un exchange de RabbitMQ. Usar un exchange de tipo topic permite a los consumidores suscribirse a patrones de eventos específicos.

tstypescript

// events/publisher.ts
import { getChannel } from '../lib/rabbitmq';
import crypto from 'crypto';
 
const EXCHANGE_NAME = 'order_events';
 
export async function setupPublisher(): Promise<void> {
  const channel = await getChannel();
 
  // Topic exchange: routes messages based on routing key pattern
  await channel.assertExchange(EXCHANGE_NAME, 'topic', {
    durable: true,  // Survives broker restart
  });
}
 
export async function publishEvent(
  event: OrderEvent
): Promise<void> {
  const channel = await getChannel();
 
  const message = Buffer.from(JSON.stringify(event));
  const routingKey = event.eventType; // e.g., "order.placed"
 
  channel.publish(EXCHANGE_NAME, routingKey, message, {
    persistent: true,     // Survives broker restart
    contentType: 'application/json',
    messageId: event.eventId,
    timestamp: Date.now(),
  });
 
  console.log(`Published ${event.eventType}: ${event.eventId}`);
}
 
// Usage in the order service
async function placeOrder(order: Order): Promise<void> {
  // Save order to database first
  await db.query(
    'INSERT INTO orders (id, customer_id, total) VALUES ($1, $2, $3)',
    [order.id, order.customerId, order.total]
  );
 
  // Then publish the event
  await publishEvent({
    eventId: crypto.randomUUID(),
    eventType: 'order.placed',
    timestamp: new Date().toISOString(),
    version: 1,
    source: 'order-service',
    data: {
      orderId: order.id,
      customerId: order.customerId,
      customerEmail: order.customerEmail,
      items: order.items,
      total: order.total,
      currency: 'USD',
    },
  });
}

Consumiendo eventos

Cada consumidor vincula su propia cola al exchange con un patrón de routing key. Varios consumidores pueden procesar el mismo evento de forma independiente.

tstypescript

// consumers/inventory-consumer.ts
import { getChannel } from '../lib/rabbitmq';
 
const EXCHANGE_NAME = 'order_events';
const QUEUE_NAME = 'inventory_order_events';
 
export async function startInventoryConsumer(): Promise<void> {
  const channel = await getChannel();
 
  // Create a durable queue for this consumer
  await channel.assertQueue(QUEUE_NAME, {
    durable: true,
    deadLetterExchange: 'dlx_order_events',  // Failed messages go here
  });
 
  // Bind queue to exchange with routing key pattern
  await channel.bindQueue(QUEUE_NAME, EXCHANGE_NAME, 'order.*');
 
  console.log('Inventory consumer listening for order events...');
 
  channel.consume(QUEUE_NAME, async (msg) => {
    if (!msg) return;
 
    try {
      const event: OrderEvent = JSON.parse(msg.content.toString());
 
      switch (event.eventType) {
        case 'order.placed':
          await handleOrderPlaced(event);
          break;
        case 'order.cancelled':
          await handleOrderCancelled(event);
          break;
        default:
          console.warn(`Unknown event type: ${event.eventType}`);
      }
 
      // Acknowledge: message processed successfully
      channel.ack(msg);
    } catch (error) {
      console.error('Failed to process message:', error);
      // Reject and send to dead letter queue
      channel.nack(msg, false, false);
    }
  });
}
 
async function handleOrderPlaced(event: OrderPlacedEvent): Promise<void> {
  for (const item of event.data.items) {
    await db.query(
      'UPDATE products SET stock = stock - $1 WHERE id = $2 AND stock >= $1',
      [item.quantity, item.productId]
    );
  }
  console.log(`Inventory updated for order ${event.data.orderId}`);
}
 
async function handleOrderCancelled(event: OrderCancelledEvent): Promise<void> {
  // Restore inventory from the cancelled order
  const order = await db.query(
    'SELECT items FROM orders WHERE id = $1',
    [event.data.orderId]
  );
 
  for (const item of order.rows[0].items) {
    await db.query(
      'UPDATE products SET stock = stock + $1 WHERE id = $2',
      [item.quantity, item.productId]
    );
  }
}

Procesamiento idempotente

Los mensajes pueden entregarse más de una vez: reintentos de red, reinicios del broker, consumidores que fallan antes de confirmar. Cada consumidor debe manejar mensajes duplicados de forma segura.

tstypescript

// lib/idempotency.ts
async function isProcessed(
  eventId: string,
  consumerName: string,
  db: Database
): Promise<boolean> {
  const result = await db.query(
    `SELECT 1 FROM processed_events
     WHERE event_id = $1 AND consumer = $2`,
    [eventId, consumerName]
  );
  return result.rows.length > 0;
}
 
async function markProcessed(
  eventId: string,
  consumerName: string,
  db: Database
): Promise<void> {
  await db.query(
    `INSERT INTO processed_events (event_id, consumer, processed_at)
     VALUES ($1, $2, NOW())
     ON CONFLICT (event_id, consumer) DO NOTHING`,
    [eventId, consumerName]
  );
}
 
// Idempotent consumer wrapper
async function processIdempotently(
  event: BaseEvent,
  consumerName: string,
  handler: (event: BaseEvent) => Promise<void>,
  db: Database
): Promise<void> {
  if (await isProcessed(event.eventId, consumerName, db)) {
    console.log(`Skipping duplicate event: ${event.eventId}`);
    return;
  }
 
  await handler(event);
  await markProcessed(event.eventId, consumerName, db);
}

tstypescript

// ❌ Non-idempotent: processes duplicates, corrupts data
async function handleOrderPlaced(event: OrderPlacedEvent) {
  await db.query(
    'UPDATE accounts SET balance = balance + $1 WHERE id = $2',
    [event.data.total, event.data.customerId]
  );
  // If this event is delivered twice, balance is credited twice!
}
 
// ✅ Idempotent: safely handles duplicate delivery
async function handleOrderPlaced(event: OrderPlacedEvent) {
  await processIdempotently(
    event,
    'payment-consumer',
    async (e) => {
      const orderEvent = e as OrderPlacedEvent;
      await db.query(
        'UPDATE accounts SET balance = balance + $1 WHERE id = $2',
        [orderEvent.data.total, orderEvent.data.customerId]
      );
    },
    db
  );
}

Colas de mensajes muertos (Dead Letter Queues)

Los mensajes que fallan al procesarse no deberían reintentarse indefinidamente. Las colas de mensajes muertos capturan los mensajes fallidos para su inspección y reprocesamiento manual.

tstypescript

// Setup dead letter exchange and queue
async function setupDeadLetterQueue(): Promise<void> {
  const channel = await getChannel();
 
  // Dead letter exchange
  await channel.assertExchange('dlx_order_events', 'fanout', {
    durable: true,
  });
 
  // Dead letter queue — stores failed messages
  await channel.assertQueue('dead_letter_order_events', {
    durable: true,
  });
 
  await channel.bindQueue(
    'dead_letter_order_events',
    'dlx_order_events',
    ''
  );
}
 
// Monitoring: check dead letter queue depth
async function getDeadLetterCount(): Promise<number> {
  const channel = await getChannel();
  const info = await channel.checkQueue('dead_letter_order_events');
  return info.messageCount;
}
 
// Reprocess dead letters (manual intervention)
async function reprocessDeadLetters(limit: number = 10): Promise<void> {
  const channel = await getChannel();
 
  for (let i = 0; i < limit; i++) {
    const msg = await channel.get('dead_letter_order_events');
    if (!msg) break;
 
    // Republish to the original exchange for retry
    channel.publish(
      'order_events',
      msg.fields.routingKey,
      msg.content,
      { persistent: true }
    );
 
    channel.ack(msg);
    console.log(`Requeued dead letter: ${msg.properties.messageId}`);
  }
}

Conclusiones clave

Publica eventos ricos: incluye todos los datos que los consumidores necesitan; los eventos anémicos que requieren callbacks crean acoplamiento entre servicios
Usa exchanges de tipo topic para un enrutamiento flexible: los consumidores se suscriben a patrones como order.* y reciben automáticamente los nuevos tipos de eventos
Cada consumidor debe ser idempotente: registra los IDs de eventos procesados para manejar de forma segura las entregas duplicadas
Confirma después de procesar, no antes: si el consumidor falla después del ack pero antes de procesar, el mensaje se pierde
Las colas de mensajes muertos evitan reintentos infinitos: los mensajes fallidos se capturan para su inspección en lugar de bloquear la cola
Prefetch de un mensaje a la vez: esto garantiza que los consumidores lentos no almacenen en búfer mensajes que no pueden procesar, evitando problemas de memoria

Construyendo un microservicio orientado a eventos con Node.js

Configurando RabbitMQ

Definiendo eventos

Publicando eventos

Consumiendo eventos

Procesamiento idempotente

Colas de mensajes muertos (Dead Letter Queues)

Conclusiones clave

Publicaciones relacionadas

Construir una herramienta CLI con Node.js desde cero

Construyendo un servidor GraphQL desde cero

Cómo construir un acortador de URLs desde cero