RabbitMQ y AMQP 0-9-1: exchanges, bindings, colas y enrutamiento

RabbitMQ es el broker que implementa AMQP 0-9-1 con mayor fidelidad, y casi todo lo que confunde a quien empieza viene de no conocer el modelo de ese protocolo. Un productor nunca envía a una cola. Envía a un exchange, el exchange consulta sus bindings y las copias del mensaje acaban en cero, una o muchas colas, de donde las toman los consumidores. Una vez que tienes clara esa imagen, la flexibilidad de RabbitMQ —el mismo mensaje yendo a una cola de trabajo, a un log de auditoría y a un feed de un dashboard sin que el productor lo sepa— se convierte en algo con lo que diseñas en lugar de algo contra lo que peleas. Esta guía está orientada a RabbitMQ 3.13 y 4.x con el cliente Java oficial.

El modelo AMQP 0-9-1

Cinco conceptos bastan para describir el protocolo.

Una conexión es una conexión TCP de larga duración, normalmente con TLS, autenticada contra un virtual host (vhost): un espacio de nombres de exchanges, colas y permisos que permite a un mismo broker servir a varias aplicaciones de forma aislada. Dentro de una conexión, las aplicaciones abren channels: sesiones multiplexadas y baratas sobre las que se hace toda la publicación y el consumo. La regla es una conexión por proceso y un channel por thread, porque los channels no son thread-safe y las conexiones son caras.

Un exchange recibe mensajes de los productores y los enruta. No almacena nada. Una cola (queue) almacena mensajes y los entrega a los consumidores. Un binding une un exchange con una cola y puede llevar una binding key (o, en los exchanges headers, un conjunto de cabeceras). Cuando un productor envía un mensaje, indica un exchange y aporta una routing key; el tipo de exchange decide cómo se compara la routing key con las binding keys.

Todo lo anterior es programable: las aplicaciones declaran exchanges, colas y bindings a través del protocolo, y las declaraciones son idempotentes siempre que los argumentos coincidan. Por eso un productor y un consumidor pueden declarar ambos la cola que comparten y arrancar en cualquier orden.

publisher --(exchange, routing key)--> exchange --bindings--> queue(s) --> consumer(s)

Los cuatro tipos de exchange

Direct

Un exchange direct enruta un mensaje a cada cola cuya binding key sea exactamente igual a la routing key. Es la herramienta para la entrega punto a punto y para repartir mensajes en un conjunto pequeño y fijo de colas, como una cola por nivel de severidad de log.

El exchange por defecto es un exchange direct predeclarado cuyo nombre es la cadena vacía. Cada cola queda enlazada automáticamente a él con su propio nombre como binding key, de modo que basicPublish("", "orders", ...) entrega en la cola llamada orders. Es cómodo para tutoriales y colas de respuesta, y una trampa para todo lo demás: los productores acaban fijando en el código los nombres de las colas, que es justo el acoplamiento que los exchanges existen para eliminar.

Fanout

Un exchange fanout ignora la routing key y copia el mensaje a todas las colas enlazadas. Es la primitiva de difusión: enlaza una cola por suscriptor y cada uno recibe su propia copia, a su propio ritmo y con sus propios acknowledgements.

Topic

Un exchange topic compara la routing key con patrones de binding. Las routing keys son palabras separadas por puntos, de hasta 255 bytes; en una binding key, * coincide exactamente con una palabra y # con cero o más palabras.

Un exchange topic con una binding key # se comporta como un fanout, y con una clave sin comodines se comporta como un exchange direct; por eso muchos equipos usan un único exchange topic por bounded context y lo expresan todo mediante convenciones de routing key como <entity>.<region>.<event>.

Headers

Un exchange headers ignora la routing key y compara las cabeceras del mensaje con los argumentos del binding; el argumento x-match con valor all exige que coincidan todas las cabeceras indicadas, y con any basta con una. Enruta por varios atributos a la vez, a costa de ser más lento y menos legible que las claves topic. La mayoría de los diseños que recurren a él saldrían ganando si codificaran esos atributos en una routing key.

Declarar la topología en Java

El cliente oficial es com.rabbitmq:amqp-client (5.x). Este productor declara un exchange topic durable y una quorum queue durable, los enlaza y publica un mensaje persistente:

import com.rabbitmq.client.*;
import java.nio.charset.StandardCharsets;
import java.util.Map;

public class OrderPublisher {
  public static void main(String[] args) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setUri("amqps://orders:[email protected]:5671/shop");  // vhost "shop"

    try (Connection connection = factory.newConnection("order-service");
         Channel channel = connection.createChannel()) {

      channel.exchangeDeclare("orders", BuiltinExchangeType.TOPIC, /* durable */ true);
      channel.queueDeclare("fulfilment", /* durable */ true, /* exclusive */ false,
          /* autoDelete */ false, Map.of("x-queue-type", "quorum"));
      channel.queueBind("fulfilment", "orders", "order.*.placed");

      channel.confirmSelect();                      // enable publisher confirms on this channel

      AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
          .contentType("application/json")
          .deliveryMode(2)                          // 2 = persistent: written to disk by durable queues
          .messageId("o-9911")
          .build();

      channel.basicPublish("orders", "order.eu.placed", /* mandatory */ true, props,
          "{\"orderId\":\"o-9911\",\"total\":149.90}".getBytes(StandardCharsets.UTF_8));

      channel.waitForConfirmsOrDie(5_000);          // throws if the broker did not confirm in 5 s
    }
  }
}

Tres flags de ese código deciden la durabilidad, y los tres son necesarios para que un mensaje sobreviva a un reinicio del broker: el exchange debe ser durable (o desaparecerá junto con sus bindings), la cola debe ser durable y el mensaje debe publicarse con delivery mode 2. Las quorum queues son siempre durables y siempre persisten los mensajes; el flag de delivery mode sigue importando en las colas clásicas.

Tipos de cola

RabbitMQ tiene tres tipos de cola, que se eligen con el argumento x-queue-type al declararla y que no se pueden cambiar después.

Una quorum queue guarda sus mensajes en una mayoría de los nodos del clúster y elige un líder mediante Raft; sobrevive a la pérdida de una minoría de nodos sin perder mensajes confirmados, que es la propiedad que la convirtió en la opción recomendada por defecto. Además, registra cuántas veces se ha reentregado un mensaje en la cabecera x-delivery-count y puede mover los mensajes a un dead-letter exchange al superar un x-delivery-limit configurable, algo que las colas clásicas no pueden hacer. El precio es más disco y memoria por mensaje y una latencia mayor que la de una cola clásica en un solo nodo.

Un stream es un log de solo anexado: los consumidores leen desde un offset, varios consumidores pueden leer los mismos mensajes y la retención se fija por tamaño o antigüedad, no por consumo. Los streams hablan AMQP 0-9-1 a través del cliente normal, pero el protocolo y el cliente de streams dedicados son mucho más rápidos. Si te encuentras queriendo un replay al estilo Kafka dentro de RabbitMQ, los streams son la respuesta.

RabbitMQ 4.0 también convirtió AMQP 1.0 en un protocolo central junto a 0-9-1, así que una aplicación Java puede usar el cliente Qpid JMS o el cliente AMQP 1.0 propio de RabbitMQ contra las mismas colas; el modelo de exchanges y bindings descrito aquí no cambia.

Consumo, acknowledgements y prefetch

Un consumidor se suscribe a una cola con basicConsume y el broker le empuja las entregas. Lo que ocurre después depende del modo de acknowledgement elegido.

Con autoAck = true, el broker olvida el mensaje en cuanto lo escribe en el socket. Si el consumidor cae, se pierde todo lo que estaba en vuelo. Con autoAck = false, el mensaje permanece en la cola, marcado como unacked, hasta que el consumidor envía basicAck; si antes se cierra el channel, el broker lo vuelve a encolar para otro consumidor. Eso es la entrega at-least-once.

channel.basicQos(20);   // prefetch: at most 20 unacknowledged deliveries per consumer on this channel

channel.basicConsume("fulfilment", /* autoAck */ false, "fulfilment-worker-1",
    new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String tag, Envelope env, AMQP.BasicProperties props, byte[] body) throws IOException {
        long deliveryTag = env.getDeliveryTag();
        try {
          fulfil(new String(body, StandardCharsets.UTF_8));
          channel.basicAck(deliveryTag, /* multiple */ false);
        } catch (TransientFailure e) {
          channel.basicNack(deliveryTag, false, /* requeue */ true);    // try again, possibly elsewhere
        } catch (RuntimeException e) {
          channel.basicNack(deliveryTag, false, /* requeue */ false);   // dead-letter it, if a DLX is set
        }
      }
    });

basicQos fija el prefetch count: cuántas entregas puede empujar el broker a un consumidor antes de recibir acknowledgements. Sin él, el broker envía tan rápido como permite el socket, un consumidor acumula miles de mensajes en memoria mientras los demás están ociosos y, si cae, todos vuelven a la cola. Un prefetch de entre 10 y 100 sirve para la mayoría de las cargas; 1 da una equidad perfecta a costa de un viaje de ida y vuelta por mensaje.

Dos límites protegen al broker de los consumidores que se portan mal. Reentregar con requeue = true un mensaje que falla siempre crea un bucle infinito que quema CPU; comprueba env.isRedeliver() o, en las quorum queues, la cabecera x-delivery-count, y deja de reencolar tras unos cuantos intentos. Y un consumidor que retiene una entrega sin confirmarla más allá del consumer timeout (30 minutos por defecto) verá cómo el broker le cierra el channel, una sorpresa habitual en consumidores de tipo batch.

Publisher confirms

Los acknowledgements protegen el lado del consumidor; los publisher confirms protegen al productor. Tras confirmSelect, el broker envía basic.ack por cada mensaje una vez enrutado y, para mensajes persistentes en colas durables, escrito en disco, o basic.nack si no pudo gestionarlo. waitForConfirmsOrDie es la forma síncrona sencilla; para ganar rendimiento, publica por lotes o registra un ConfirmListener y lleva la cuenta de los números de secuencia pendientes, volviendo a publicar todo lo que reciba nack. Junto con mandatory = true y un ReturnListener, que devuelve los mensajes que no coincidieron con ningún binding, los confirms te dan un productor que sabe qué ha aceptado el broker.

TTL, límites y dead-lettering

Las colas aceptan argumentos opcionales que controlan qué ocurre con los mensajes que no se consumen.

• x-message-ttl — milisegundos que un mensaje puede esperar antes de ser descartado o enviado a dead letter. La propiedad expiration de cada mensaje hace lo mismo para un solo mensaje.
• x-expires — elimina la propia cola tras ese número de milisegundos sin consumidores ni actividad; útil para colas de respuesta por cliente.
• x-max-length / x-max-length-bytes con x-overflow — limita el tamaño de la cola; drop-head descarta los mensajes más antiguos y reject-publish hace que las nuevas publicaciones reciban nack con confirms activados, para que el productor note la contrapresión.
• x-dead-letter-exchange y x-dead-letter-routing-key — adónde van los mensajes rechazados, caducados o desbordados.

Un dead-letter exchange (DLX) es un exchange normal; enlazarle una cola te da un lugar donde se acumulan los mensajes fallidos con una cabecera x-death que describe por qué y cuántas veces. La combinación de TTL y DLX es también la forma en que RabbitMQ implementa reintentos diferidos sin plugins: haz nack del mensaje fallido hacia una cola de «espera» cuyo TTL sea de treinta segundos y cuyo dead-letter exchange enrute de vuelta a la cola de trabajo.

Map<String, Object> args = Map.of(
    "x-queue-type", "quorum",
    "x-dead-letter-exchange", "orders.dlx",
    "x-delivery-limit", 5,               // quorum queues: dead-letter after 5 redeliveries
    "x-message-ttl", 86_400_000);        // one day
channel.queueDeclare("fulfilment", true, false, false, args);

Los argumentos fijados en la declaración quedan congelados; cambiarlos implica borrar la cola. Usa policies para todo lo que pueda cambiar: rabbitmqctl set_policy dlx "^fulfilment$" '{"dead-letter-exchange":"orders.dlx"}' --apply-to queues aplica el mismo ajuste sin volver a declarar la cola y se puede actualizar en caliente.

Administración del broker

El plugin de gestión (management) ofrece una interfaz web y una API HTTP en el puerto 15672 (AMQP en sí va por el 5672, TLS por el 5671 y las métricas de Prometheus por el 15692). Muestra exchanges, bindings, profundidad de las colas, tasas de mensajes, conexiones y channels, y permite publicar y obtener mensajes a mano para depurar. La cuenta por defecto guest/guest solo puede iniciar sesión desde localhost; crea usuarios reales y concédeles permisos por vhost antes de que se conecte nada más:

rabbitmqctl add_vhost shop
rabbitmqctl add_user orders 'change-me'
rabbitmqctl set_permissions -p shop orders '^orders.*' '^orders$' '^(fulfilment|orders.*)$'
#                                          configure     write       read     (regular expressions on names)
rabbitmqctl list_queues -p shop name type messages messages_unacknowledged consumers
rabbitmq-diagnostics status

Los tres patrones de permisos son expresiones regulares sobre los nombres de los recursos para configure (declarar y eliminar), write (publicar en un exchange) y read (consumir de una cola, enlazar una cola). La herramienta de línea de comandos rabbitmqadmin envuelve la API HTTP para automatizar declaraciones con scripts, y rabbitmq-diagnostics informa de la salud de los nodos, las alarmas y el estado del clúster. La primera alarma con la que te toparás en la práctica es la alarma de recursos de memoria o disco: cuando un nodo supera su umbral máximo (high watermark), bloquea todas las conexiones que publican hasta que los consumidores se ponen al día, lo que para el productor parece un cuelgue.

Errores comunes

• Publicar con el nombre de una cola en un exchange personalizado. La routing key debe coincidir con un binding; si no hay binding, el mensaje se descarta y no se registra nada. Usa mandatory junto con un return listener mientras desarrollas.
• Una conexión por mensaje, o un channel compartido entre threads. Lo primero agota los descriptores de archivo y al propio broker; lo segundo corrompe el flujo del protocolo. Una conexión por proceso, un channel por thread.
• Cola durable, mensaje transitorio. Un delivery mode 1 en una cola clásica no sobrevive a un reinicio aunque la cola sí lo haga. Usa el modo 2, o quorum queues, que ignoran esa distinción.
• Sin prefetch. Un consumidor se lo lleva todo, los demás están ociosos y una caída provoca miles de reentregas. Llama siempre a basicQos.
• Bucles infinitos de reencolado. Un mensaje envenenado al que se hace nack con requeue = true vuelve de inmediato. Limita las reentregas con x-delivery-limit o comprobando x-death, y envíalo a dead letter.
• Fijar ajustes cambiantes en los argumentos de la cola. Los TTL, los límites y los nombres de DLX cambian; los argumentos de cola no. Usa policies.

Puntos clave

• Los productores envían a exchanges con una routing key; los bindings deciden qué colas reciben copias; los consumidores leen de las colas. Nada más en RabbitMQ tiene sentido sin este modelo.
• Los exchanges direct comparan claves exactas, fanout copia a todos, topic aplica patrones * y #, headers compara atributos; un exchange topic por contexto con una convención de routing keys cubre la mayoría de los diseños.
• Las quorum queues son la opción por defecto para trabajo durable; las colas clásicas cubren los usos transitorios y exclusivos; los streams añaden logs reproducibles.
• La entrega at-least-once requiere a la vez exchange durable, cola durable, mensaje persistente, publisher confirms y acknowledgements manuales.
• El prefetch de basicQos reparte la carga con equidad entre los consumidores y abarata las caídas; los consumer timeouts y los delivery limits impiden que los mensajes atascados o envenenados tumben el broker.
• Expresa los TTL, los límites de longitud y el enrutamiento a dead letter como policies para poder cambiarlos sin volver a declarar las colas.