Colas de mensajes en la práctica

// Point-to-point: one message, one consumer
// Use case: task distribution (send email, process payment)
// Each message is processed by exactly ONE worker
 
// Pub/Sub: one message, many consumers
// Use case: event broadcasting (order placed → notify analytics, inventory, email)
// Each message is delivered to ALL subscribers

// ❌ Synchronous chain — one failure breaks everything
app.post("/api/orders", async (req, res) => {
  const order = await createOrder(req.body);
  await inventoryService.reserve(order.items);      // If this fails...
  await paymentService.charge(order.total);          // ...none of this runs
  await emailService.sendConfirmation(order);        // ...the user sees an error
  await analyticsService.trackPurchase(order);
  res.json(order);
});
 
// ✅ Async event — order is placed, downstream services react independently
app.post("/api/orders", async (req, res) => {
  const order = await createOrder(req.body);
 
  await messageQueue.publish("order.placed", {
    orderId: order.id,
    items: order.items,
    total: order.total,
    customerId: order.customerId,
  });
 
  res.json(order);
});
 
// Each service consumes the event independently
// inventory-service listens to "order.placed"
// payment-service listens to "order.placed"
// email-service listens to "order.placed"
// analytics-service listens to "order.placed"

// At-most-once: fire and forget
// Message may be lost, but never duplicated
// Use for: metrics, logging, non-critical notifications
 
// At-least-once: guaranteed delivery, possible duplicates
// Consumer MUST be idempotent
// Use for: most business events (payments, emails, orders)
 
// Exactly-once: no loss, no duplicates (very expensive)
// Requires distributed transactions or deduplication
// Use for: financial transactions (or use at-least-once + idempotency)

import amqp from "amqplib";
 
// Producer
async function publishOrderEvent(order: Order) {
  const connection = await amqp.connect(process.env.RABBITMQ_URL!);
  const channel = await connection.createChannel();
 
  await channel.assertExchange("orders", "topic", { durable: true });
 
  channel.publish(
    "orders",
    "order.placed",
    Buffer.from(JSON.stringify(order)),
    { persistent: true }, // Survive broker restarts
  );
}
 
// Consumer
async function startOrderConsumer() {
  const connection = await amqp.connect(process.env.RABBITMQ_URL!);
  const channel = await connection.createChannel();
 
  await channel.assertExchange("orders", "topic", { durable: true });
  const queue = await channel.assertQueue("inventory-service", {
    durable: true,
  });
  await channel.bindQueue(queue.queue, "orders", "order.placed");
 
  channel.prefetch(10); // Process 10 messages at a time
 
  channel.consume(queue.queue, async (msg) => {
    if (!msg) return;
 
    try {
      const order = JSON.parse(msg.content.toString());
      await reserveInventory(order);
      channel.ack(msg); // Acknowledge successful processing
    } catch (error) {
      channel.nack(msg, false, true); // Requeue on failure
    }
  });
}

import { Kafka } from "kafkajs";
 
const kafka = new Kafka({
  clientId: "order-service",
  brokers: [process.env.KAFKA_BROKER!],
});
 
// Producer
const producer = kafka.producer();
await producer.connect();
 
await producer.send({
  topic: "orders",
  messages: [
    {
      key: order.id, // Partition by order ID for ordering guarantees
      value: JSON.stringify({ event: "order.placed", data: order }),
    },
  ],
});
 
// Consumer group — messages distributed among group members
const consumer = kafka.consumer({ groupId: "inventory-service" });
await consumer.connect();
await consumer.subscribe({ topic: "orders", fromBeginning: false });
 
await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const event = JSON.parse(message.value!.toString());
 
    if (event.event === "order.placed") {
      await reserveInventory(event.data);
    }
  },
});