RabbitMQ desde cero: cómo funcionan las colas de mensajes

Una aplicación recibe una solicitud para generar un informe. Si lo crea dentro de la petición HTTP, el usuario tendrá que esperar y un fallo del servicio de informes puede hacer fallar toda la operación. Otra opción es registrar el trabajo pendiente, responder que fue aceptado y procesarlo después.

RabbitMQ ayuda a conectar esas dos etapas. No convierte el trabajo en instantáneo ni elimina los fallos: permite que quien solicita una tarea y quien la ejecuta se coordinen mediante mensajes. A cambio, tendremos que decidir qué ocurre si un mensaje se pierde, se repite o no puede procesarse.

Vamos a construir un productor y un consumidor en Node.js, con RabbitMQ en Docker Compose. El trabajo será simulado: no enviaremos correos ni ejecutaremos cobros reales. Así podremos observar confirmaciones, reintentos y mensajes fallidos sin introducir efectos externos.

Qué hace un message broker

Un message broker es un intermediario que recibe mensajes y los encamina hacia sus destinos. RabbitMQ puede conservarlos en colas hasta que haya consumidores disponibles. Un mensaje contiene bytes y propiedades: puede representar una orden, como «genera este informe», o un evento, como «el pedido fue creado». El significado lo define la aplicación. RabbitMQ: publishers.

En una comunicación síncrona, el solicitante espera una respuesta de la operación remota para continuar ese flujo. En una comunicación asíncrona mediante mensajes, entrega el trabajo y su resultado llega por otro mecanismo: una consulta de estado, una notificación o un evento posterior.

No es lo mismo que escribir async y await: una llamada HTTP puede usar código no bloqueante y, aun así, mantener al usuario esperando al servicio remoto. Tampoco es necesario renunciar a toda espera para usar una cola: el productor puede esperar la confirmación del broker sin esperar a que termine el trabajo de negocio.

En nuestro caso podríamos responder HTTP 202 Accepted y devolver un identificador de trabajo. La API tendría que guardar y exponer su estado; RabbitMQ no implementa ese contrato HTTP por nosotros. Si la operación es rápida y necesita una respuesta inmediata, una llamada directa puede ser más sencilla.

Una cola permite absorber un pico temporal, pero no crea capacidad infinita. Si llegan diez trabajos por segundo y solo puedes completar cinco, la espera seguirá creciendo hasta que reduzcas la entrada o aumentes la capacidad.

Productores, exchanges, queues y bindings

El laboratorio usa AMQP 0-9-1, uno de los protocolos admitidos por RabbitMQ. Sus piezas principales son:

El productor publica en un exchange; el consumidor se suscribe a una cola. La routing key es una cadena que acompaña a la publicación. El exchange la interpreta según su tipo y los bindings existentes. No es necesariamente el nombre de una cola. RabbitMQ: exchanges.

---
config:
  flowchart:
    wrappingWidth: 220
---
flowchart TB
  P["`**Productor**
  Solicita un informe`"]
  E["`**Exchange direct**
  informes`"]
  Q["`**Cola**
  informes.generar`"]
  C["`**Consumidor**
  Ejecuta el trabajo`"]

  P -->|informe.solicitado| E
  E -->|Binding coincidente| Q
  Q --> C
  C -.->|Ack al terminar| Q

Dos consumidores de la misma cola compiten por los trabajos; no reciben ambos una copia de cada mensaje. Para que dos funcionalidades independientes reciban el mismo evento, normalmente necesitarás dos colas vinculadas al exchange. RabbitMQ: consumers.

Direct, fanout y topic

En topic, * sustituye exactamente una palabra y # sustituye cero o más. Por ejemplo, pedido.*.creado coincide con pedido.web.creado, pero no con pedido.creado. Un exchange direct también puede enviar a varias colas si sus bindings coinciden: «direct» no significa «un único consumidor». Tutorial de routing, tutorial de topics.

Dos confirmaciones que resuelven problemas distintos

Un publisher confirm informa al productor de que el broker asumió responsabilidad por la publicación. Un consumer acknowledgment, o ack, informa al broker de que el consumidor terminó de procesar una entrega. El primero no demuestra que se haya generado el informe.

El consumidor debe confirmar después del trabajo que considera completado. Si su conexión se pierde antes del ack, una entrega pendiente puede volver a la cola y llegar a otro consumidor. Confirmar antes de ejecutar el trabajo abre una ventana de pérdida si el proceso cae. RabbitMQ: acknowledgments y confirms.

También hay un problema anterior al consumo: publicar en un exchange existente sin ningún binding coincidente. Por defecto, ese mensaje puede descartarse. Usaremos mandatory: true y atenderemos el evento return para detectar esa situación. Un confirm por sí solo no certifica que el mensaje haya encontrado una cola. RabbitMQ: mensajes sin ruta.

Durabilidad no es una sola opción

Estas decisiones se complementan:

• Una cola durable conserva su definición tras un reinicio; un exchange durable conserva su propia definición.
• Un mensaje persistente expresa que debe conservarse en almacenamiento duradero cuando corresponda a una cola durable.
• Los publisher confirms permiten esperar la confirmación del broker en lugar de suponer que escribir en el socket fue suficiente.
• Un volumen de Docker conserva los datos cuando se reemplaza el contenedor, siempre que no se elimine ese volumen.

Una cola durable no vuelve persistentes automáticamente todos sus mensajes. Y un volumen local no protege frente a la pérdida del disco. RabbitMQ: durabilidad de colas.

Usaremos colas clásicas en un solo nodo para estudiar el recorrido. No es una configuración de alta disponibilidad. Si necesitas replicación y tolerancia a fallos de nodos, evalúa quorum queues y una topología adecuada; una cola quorum de un solo nodo tampoco elimina ese punto de fallo. RabbitMQ: quorum queues.

Laboratorio: RabbitMQ en Docker y dos aplicaciones Node.js

Necesitas Docker con Compose, Node.js 22 o superior y pnpm. RabbitMQ se ejecutará en un contenedor; el productor y el consumidor serán procesos Node.js en tu equipo. El ejemplo utiliza RabbitMQ 4.4.0-management y amqplib 2.2.1, versiones comprobadas durante la preparación.

Crea una carpeta independiente del proyecto de tu aplicación:

mkdir rabbitmq-lab
cd rabbitmq-lab

pnpm init
pnpm add --save-exact amqplib@2.2.1

Conserva el pnpm-lock.yaml generado para repetir la instalación con pnpm install --frozen-lockfile.

1. Levantar el broker

Guarda este archivo como compose.yaml:

name: uncodigo-rabbitmq-lab

services:
  rabbitmq:
    image: rabbitmq:4.4.0-management
    hostname: rabbitmq-lab
    environment:
      RABBITMQ_DEFAULT_USER: demo
      RABBITMQ_DEFAULT_PASS: solo-para-este-laboratorio
      RABBITMQ_DEFAULT_VHOST: laboratorio
    ports:
      - "127.0.0.1:5673:5672"
      - "127.0.0.1:15673:15672"
    volumes:
      - rabbit-data:/var/lib/rabbitmq
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "-q", "check_running"]
      interval: 5s
      timeout: 5s
      retries: 12

volumes:
  rabbit-data:

Los puertos se publican solo en 127.0.0.1. Las credenciales son ficticias y exclusivas del laboratorio; no reutilices este usuario, contraseña o exposición del panel en producción. El hostname permanece estable y los datos viven en un volumen del proyecto Compose.

docker compose up -d --wait

La opción --wait usa el healthcheck para esperar a que el servicio esté saludable. En general, que un contenedor esté iniciado no significa que su aplicación esté lista. Docker: orden de inicio y healthchecks.

Abre el panel local de RabbitMQ con el usuario demo y la contraseña del Compose. El vhost laboratorio es el espacio lógico donde declararemos exchanges, colas y permisos.

Antes de publicar, configura mediante una policy el destino de los mensajes rechazados:

docker compose exec rabbitmq rabbitmqctl set_policy \
  -p laboratorio informes-dlx '^informes\.generar$' \
  '{"dead-letter-exchange":"informes.dlx","dead-letter-routing-key":"fallido"}' \
  --apply-to queues

La policy se aplicará a la cola cuando la creemos. Configurar el DLX mediante policy permite cambiarlo administrativamente, sin fijarlo como argumento inmutable en el código de declaración. RabbitMQ: configuración de dead-letter exchanges.

En cada terminal donde ejecutes Node.js, define la conexión. Los siguientes comandos usan una shell POSIX, como Bash o Zsh:

export AMQP_URL='amqp://demo:solo-para-este-laboratorio@127.0.0.1:5673/laboratorio?heartbeat=10'

2. Compartir la topología

Guarda broker.mjs. Ambos programas usarán las mismas declaraciones para evitar depender de cuál arranque primero:

import amqp from 'amqplib';

export const EXCHANGE = 'informes';
export const COLA = 'informes.generar';
export const RUTA = 'informe.solicitado';

export async function abrir() {
  const url = process.env.AMQP_URL;

  if (!url) {
    throw new Error('Falta AMQP_URL');
  }

  const conexion = await amqp.connect(url, { timeout: 5000 });
  conexion.on('error', () => {
    console.error('Error de conexión AMQP');
    process.exit(1);
  });

  const canal = await conexion.createConfirmChannel();
  canal.on('error', () => {
    console.error('Error de canal AMQP');
    process.exit(1);
  });

  await canal.assertExchange(EXCHANGE, 'direct', { durable: true });
  await canal.assertExchange('informes.dlx', 'direct', { durable: true });

  await canal.assertQueue('informes.fallidos', { durable: true });
  await canal.bindQueue('informes.fallidos', 'informes.dlx', 'fallido');

  await canal.assertQueue(COLA, { durable: true });
  await canal.bindQueue(COLA, EXCHANGE, RUTA);

  return { conexion, canal };
}

informes.dlx es otro exchange normal. Su binding enviará los mensajes con clave fallido a informes.fallidos, que sí es una cola. DLX y cola de fallidos no son sinónimos.

Volver a declarar una cola con atributos equivalentes es válido. Intentar cambiar sus propiedades incompatiblemente, por ejemplo de no durable a durable, genera un error de canal; no es una migración automática. RabbitMQ: propiedades de una cola.

3. Publicar y comprobar la confirmación

Guarda productor.mjs:

import { abrir, EXCHANGE, RUTA } from './broker.mjs';

const id = process.argv[2];
const modo = process.argv[3] ?? 'normal';
const ruta = process.argv[4] ?? RUTA;

if (!id || !/^[a-z0-9-]{1,64}$/.test(id)) {
  throw new Error('Indica un ID de 1 a 64 caracteres: a-z, 0-9 o guion');
}

const { conexion, canal } = await abrir();
let devuelto = false;

canal.on('return', () => {
  devuelto = true;
});

try {
  const contenido = Buffer.from(JSON.stringify({ version: 1, id, modo }));

  await new Promise((resolve, reject) => {
    canal.publish(EXCHANGE, ruta, contenido, {
      persistent: true,
      mandatory: true,
      contentType: 'application/json',
      messageId: id,
    }, (error) => {
      if (error) reject(error);
      else resolve();
    });
  });

  if (devuelto) {
    throw new Error('El mensaje no llegó a ninguna cola');
  }

  console.log(`Confirmado por RabbitMQ: ${id}`);
} catch {
  console.error('Publicación fallida o sin ruta; no asumir entrega');
  process.exitCode = 1;
} finally {
  await conexion.close();
}

Este programa publica un único mensaje por ejecución. La promesa espera el callback del confirm channel; no hace await canal.publish(...), porque ese método devuelve un booleano relacionado con el buffer de escritura, no una promesa de confirmación. Un productor de alto volumen también debe gestionar contrapresión y limitar publicaciones pendientes. API de amqplib.

Al publicar con mandatory, RabbitMQ envía el return antes del confirm de un mensaje sin ruta. Aquí la variable devuelto basta porque solo hay una publicación; con varias concurrentes necesitarías correlacionarlas. RabbitMQ: cuándo se confirma una publicación.

4. Consumir, validar y decidir qué confirmar

Guarda consumidor.mjs:

import { setTimeout as esperar } from 'node:timers/promises';
import { abrir, COLA } from './broker.mjs';

const { conexion, canal } = await abrir();

await canal.prefetch(1);

function leer(mensaje) {
  if (mensaje.content.length > 4096) throw new Error('Mensaje demasiado grande');

  const datos = JSON.parse(mensaje.content.toString('utf8'));

  if (
    datos?.version !== 1 ||
    typeof datos.id !== 'string' ||
    !/^[a-z0-9-]{1,64}$/.test(datos.id) ||
    mensaje.properties.messageId !== datos.id ||
    !['normal', 'transitorio', 'permanente'].includes(datos.modo)
  ) {
    throw new Error('Contrato inválido');
  }

  return datos;
}

async function procesar(mensaje) {
  let datos;

  try {
    datos = leer(mensaje);
  } catch {
    console.error('Contrato inválido: enviar a fallidos');
    canal.nack(mensaje, false, false);
    return;
  }

  for (let intento = 1; intento <= 3; intento++) {
    console.log(JSON.stringify({
      id: datos.id,
      intento,
      redelivered: mensaje.fields.redelivered,
    }));

    // Simula un servicio temporalmente indisponible, sin efectos externos.
    const falla = datos.modo === 'permanente' ||
      (datos.modo === 'transitorio' && intento < 3);

    if (!falla) {
      console.log(`Trabajo simulado terminado: ${datos.id}`);
      canal.ack(mensaje);
      return;
    }

    if (datos.modo === 'permanente' || intento === 3) break;

    await esperar(500 * intento);
  }

  console.error(`Sin completar; enviar a fallidos: ${datos.id}`);
  canal.nack(mensaje, false, false);
}

await canal.consume(COLA, (mensaje) => {
  if (!mensaje) {
    console.error('RabbitMQ canceló el consumidor');
    process.exit(1);
  }

  procesar(mensaje).catch(() => {
    console.error('Fallo inesperado; cerrar para permitir la reentrega');
    process.exit(1);
  });
}, { noAck: false });

process.once('SIGINT', () => {
  conexion.close().then(() => process.exit(0), () => process.exit(1));
});

console.log('Esperando mensajes. Ctrl+C para salir.');

prefetch(1) limita a una entrega pendiente de confirmación para este consumidor. Es útil para observar el comportamiento, pero no es un valor óptimo universal. noAck: false activa acknowledgments manuales. RabbitMQ: consumers y prefetch.

El modo transitorio simula dos fallos y éxito en el tercer intento. El modo permanente se rechaza sin repetir un trabajo que ya sabemos que no funcionará. Cualquier contrato inválido también se rechaza. Son decisiones del laboratorio, no una clasificación que RabbitMQ deduzca del contenido.

Los reintentos ocurren dentro de la misma entrega, que sigue sin confirmar. Al cerrar con Ctrl+C, una entrega pendiente puede volver a la cola; este cierre sencillo no espera a terminar trabajos. Tampoco configuramos recuperación automática de conexión: ante un fallo de AMQP, este laboratorio termina para que el problema sea visible.

Ejecutar y observar el recorrido

Publicar antes de iniciar el consumidor

Con el consumidor todavía detenido:

node productor.mjs informe-001

docker compose exec rabbitmq rabbitmqctl list_queues \
  -p laboratorio name messages_ready messages_unacknowledged

El productor debe mostrar Confirmado por RabbitMQ: informe-001. En un laboratorio recién creado, informes.generar tendrá un mensaje listo (messages_ready) y ninguno entregado pendiente de ack (messages_unacknowledged). No significa que el informe ya esté generado.

En el panel, abre Exchanges → informes → Bindings. El binding conecta el exchange direct con informes.generar mediante la clave informe.solicitado:

Después abre Queues and Streams → informes.generar. Ready cuenta los mensajes que esperan una entrega; Unacked, los entregados que todavía esperan confirmación del consumidor. En esta captura, tomada antes de iniciar el consumidor, vemos Ready: 1 y Unacked: 0:

Las capturas corresponden al laboratorio local. Puedes ampliarlas para leer los detalles del panel; las horas y tasas de tu ejecución serán distintas.

Para comprobar el reinicio con ese mensaje aún pendiente:

docker compose restart rabbitmq
docker compose up -d --wait

docker compose exec rabbitmq rabbitmqctl list_queues \
  -p laboratorio name messages_ready messages_unacknowledged

El mensaje debe seguir disponible. Esta comprobación muestra recuperación tras un reinicio conservando el volumen, no resistencia a la pérdida del disco o de todo el broker.

Ahora inicia el consumidor en una terminal:

node consumidor.mjs

Verás un intento para informe-001 y Trabajo simulado terminado: informe-001. Tras el ack, la cola quedará sin ese mensaje. La aplicación solo imprime el resultado: no genera un archivo ni guarda un estado de negocio.

Probar reintentos, contratos inválidos y mensajes sin ruta

Con el consumidor activo, ejecuta desde otra terminal:

node productor.mjs informe-002 transitorio
node productor.mjs informe-003 permanente
node productor.mjs informe-004 invalido

node productor.mjs informe-005 normal ruta.inexistente

El último comando debe terminar con código de salida 1. Es intencional: hay un exchange, pero no una cola vinculada a esa ruta. El modo invalido también es deliberado: el productor permite enviarlo para comprobar que el consumidor valida lo recibido.

Consulta las colas y abre informes.fallidos en el panel. Su apartado Get messages, con reencolado activado, permite inspeccionar el contenido y la cabecera x-death. Esa cabecera describe el paso por dead lettering; no cuenta nuestros reintentos internos en Node.js.

Selecciona Ack Mode: Nack message requeue true, deja Messages: 1 y pulsa Get Message(s). En el mensaje informe-003, el campo reason: rejected identifica el rechazo y queue: informes.generar señala la cola de origen. El payload conserva el modo permanente que enviamos:

Esta captura se tomó con solo ese mensaje en la cola de fallidos. El panel informó 0 messages remaining al obtenerlo, pero el modo seleccionado lo vuelve a encolar: ese texto no demuestra que la cola haya quedado vacía después de la inspección. Consultar mensajes modifica temporalmente su estado; usa este procedimiento en el laboratorio, no como un visor inocuo de una cola de producción.

En la ejecución local de estos archivos se verificó:

Son observaciones funcionales, no mediciones de capacidad ni garantías ante cualquier fallo. Si ya ejecutaste otros mensajes, los contadores pueden ser distintos.

Reintentos y dead-letter exchanges: decidir antes de repetir

canal.nack(mensaje, false, false) rechaza una entrega individual y no solicita reencolarla en la cola original. La policy la dirige al DLX. Sin un DLX válido, ese rechazo puede acabar en descarte. Una cola de fallidos necesita revisión y un procedimiento de recuperación; por sí sola no corrige ni reintenta nada.

En cambio, reencolar inmediatamente cada fallo puede crear un bucle rápido que consume CPU y no deja recuperarse a la dependencia. Separa errores temporales, mensajes imposibles de procesar y fallos desconocidos. Define límite de intentos, espera y un destino final.

Nuestra espera de 500 y 1.000 ms mantiene ocupada la única entrega permitida por el prefetch. Además, el contador se pierde si el proceso cae. Es suficiente para estudiar la decisión de reintentar; no es un planificador persistente de reintentos.

Para esperas gestionadas por el broker, las quorum queues de RabbitMQ 4.3 o superior admiten delayed retry. También existen diseños con colas de espera y TTL, pero hay que entender la expiración y los riesgos de los traslados. No necesitas copiar una arquitectura de varias colas antes de comprobar qué soporta tu versión. Quorum queues: delayed retry, TTL.

El dead lettering convencional no ofrece por defecto una transferencia infalible: puede perder mensajes si el destino no está disponible. Las quorum queues permiten configurar dead lettering at-least-once. El laboratorio usa colas clásicas y no prueba esa garantía adicional. RabbitMQ: seguridad del DLX.

At-most-once, at-least-once e idempotencia

At-most-once admite que un mensaje no llegue a procesarse y evita reentregas en el tramo considerado. Consumir con ack automático puede perder trabajo si el proceso falla después de recibirlo. At-least-once acepta posibles repeticiones para recuperar entregas no confirmadas; requiere configurar y operar de forma coherente el productor, el broker y el consumidor. No es una promesa que se active únicamente con durable: true. RabbitMQ: garantías de entrega.

Imagina que el consumidor guarda el informe y cae antes de enviar el ack. RabbitMQ vuelve a entregarlo. Si el segundo intento crea otro informe o cobra otra vez, el transporte está recuperándose, pero la operación de negocio es incorrecta.

Un consumidor idempotente conserva el mismo efecto de negocio cuando repite una operación identificada. messageId ayuda a transportar ese identificador, pero no activa deduplicación en nuestras colas. Compruébalo publicando otra vez:

node productor.mjs informe-001

El consumidor volverá a imprimir el trabajo. Incluso redelivered puede ser false, porque esta es una nueva publicación del mismo trabajo lógico. Ese indicador no sustituye una clave de idempotencia.

Para una operación almacenada en una base de datos, un diseño posible es:

1. Abrir una transacción.
2. Registrar una clave única por consumidor y operación.
3. Si ya existe un registro completado, evitar repetir el efecto.
4. Si es nuevo, aplicar el cambio de negocio y confirmar ambos registros en la misma transacción.
5. Enviar el ack después del commit.

Un Set en memoria no coordina varias réplicas y se pierde al reiniciar. Si el efecto está en un proveedor externo, una transacción local no lo cubre: necesitas un contrato idempotente con ese proveedor o un proceso de reconciliación. El problema se relaciona con la idempotencia en APIs REST.

También existe una ventana entre guardar una operación en tu base de datos y publicar su evento. El patrón transactional outbox registra el evento junto al cambio de negocio en una misma transacción; otro proceso lo publica y gestiona confirmaciones. Puede repetir publicaciones, por lo que no elimina la necesidad de consumidores idempotentes. Es una decisión arquitectónica adicional, no algo implementado en este laboratorio.

Qué falta antes de usarlo en producción

El ejemplo facilita observar el protocolo y los fallos. Para una aplicación real tendrías que concretar, al menos:

• Seguridad y operación: TLS, credenciales administradas, permisos separados para publicar, consumir y configurar topología; red privada para el broker y el panel.
• Disponibilidad: tipo de cola, replicación, recuperación y pruebas de fallos según cuánto trabajo puedes perder y cuánto tiempo puedes estar detenido.
• Capacidad: límites de cola, tamaño de mensajes, caducidad cuando tenga sentido, contrapresión y un prefetch ajustado a la carga.
• Aplicaciones: contratos versionados, idempotencia persistente, recuperación de conexiones y canales, límites de tiempo del trabajo y cierre ordenado.
• Diagnóstico: mensajes listos y pendientes de ack, tasas de publicación y consumo, mensajes fallidos, alarmas de memoria y disco y correlación por identificador. RabbitMQ: monitoring.

No registres el cuerpo completo de cada mensaje por comodidad. El identificador y el resultado suelen ser mejores puntos de partida para conectar el flujo con logs, métricas y trazas. También conviene separar las credenciales del código y administrarlas según el entorno, como explicamos en variables de entorno y gestores de secretos.

Evita cuatro atajos habituales: confirmar antes de terminar, reencolar errores permanentes indefinidamente, interpretar un publisher confirm como éxito de negocio y asumir que varias réplicas mantendrán un orden global de finalización. Las prioridades, reentregas y consumidores concurrentes afectan al orden observado. RabbitMQ: orden de los mensajes.

Limpiar el laboratorio y decidir si necesitas una cola

Detén el consumidor con Ctrl+C. Para retirar solo este laboratorio y borrar sus mensajes y volumen:

docker compose down --volumes
unset AMQP_URL

Ejecuta ese comando desde la carpeta del laboratorio: elimina los datos del proyecto Compose definido allí. Si quieres conservarlos para otra sesión, omite --volumes.

RabbitMQ encaja cuando puedes separar la aceptación de una tarea de su ejecución y necesitas gestionar trabajo pendiente entre procesos. Si la comunicación directa resuelve el problema, introducir un broker añade operación y estados que quizá no necesitas. La decisión útil no es «usar colas porque escalan», sino definir quién acepta el trabajo, quién lo termina y cómo se recupera cuando alguna de esas etapas falla.

Referencias

• RabbitMQ: tutorial JavaScript inicial y work queues.
• Exchanges, routing y topics.
• Publishers, consumers y acknowledgments y confirms.
• Queues, reliability, dead-letter exchanges y quorum queues.
• amqplib: API de canales.
• Docker Compose: inicio y disponibilidad de servicios.