advanced Por Mathias Paulenko 4 jul 2026

Referencia Detallada de Arquitectura RabbitMQ

Disenar y operar RabbitMQ para mensajeria confiable. Cubre exchanges, queues, bindings, patrones de routing, dead letter queues, clustering y mejores practicas de produccion para workloads de alto throughput.

rabbitmq messaging guia amqp exchanges queues routing dead-letter

Temas: messaging architecture infrastructure

Nota para desarrolladores hispanohablantes: Esta guía incluye ejemplos y convenciones de nomenclatura adaptadas a equipos que trabajan en español. Cuando existen diferencias significativas en terminología técnica entre el inglés y el español, se indican explícitamente para facilitar la comunicación en equipos multiculturales.

Introducción

RabbitMQ es un message broker ampliamente usado que implementa AMQP (Advanced Message Queuing Protocol). Destaca en rutear mensajes entre producers y consumers con tipos de exchange flexibles, garantias de entrega confiable, y features ricos de queue. Lo siguiente es una guia practica para arquitectura RabbitMQ, tipos de exchange, patrones de routing, y mejores practicas de produccion.

Arquitectura RabbitMQ

Componentes Clave

Producer → Exchange → (Binding + Routing Key) → Queue → Consumer
              ↑
         Tipos de Exchange:
         - Direct:  routing key == binding key
         - Topic:   routing key matchea patron
         - Fanout:  broadcast a todas las queues bound
         - Headers: matchea headers del mensaje

Exchange: Recibe mensajes de producers y los rutea a queues.
Queue: Un buffer que almacena mensajes hasta que los consumers los procesan.
Binding: Un link entre un exchange y una queue con una regla de routing.
Routing Key: Un string que el exchange usa para decidir que queue recibe el mensaje.
Connection: Una conexion TCP entre un client y RabbitMQ.
Channel: Una conexion virtual dentro de una conexion. Multiplexa multiples canales sobre una conexion TCP.

Tipos de Exchange

Direct Exchange

Rutea mensajes a queues donde el routing key matchea exactamente el binding key.

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()

# Declarar un direct exchange
channel.exchange_declare(exchange="orders_direct", exchange_type="direct")

# Declarar queues
channel.queue_declare(queue="orders_created")
channel.queue_declare(queue="orders_cancelled")

# Bindear queues al exchange con routing keys
channel.queue_bind(exchange="orders_direct", queue="orders_created", routing_key="created")
channel.queue_bind(exchange="orders_direct", queue="orders_cancelled", routing_key="cancelled")

# Publicar mensajes
channel.basic_publish(
    exchange="orders_direct",
    routing_key="created",
    body='{"order_id": 123, "total": 49.99}'
)

channel.basic_publish(
    exchange="orders_direct",
    routing_key="cancelled",
    body='{"order_id": 124, "reason": "customer_request"}'
)

Topic Exchange

Rutea mensajes basado en patrones de routing key. Wildcards: * matchea una palabra, # matchea cero o mas palabras.

# Declarar un topic exchange
channel.exchange_declare(exchange="logs_topic", exchange_type="topic")

# Bindear queues con patrones
channel.queue_bind(exchange="logs_topic", queue="all_errors", routing_key="*.error")
channel.queue_bind(exchange="logs_topic", queue="app_errors", routing_key="app.*")
channel.queue_bind(exchange="logs_topic", queue="all_logs", routing_key="#")

# Publicar mensajes
channel.basic_publish(exchange="logs_topic", routing_key="app.error", body="App error occurred")
# → Va a: all_errors, app_errors, all_logs

channel.basic_publish(exchange="logs_topic", routing_key="db.warning", body="DB warning")
# → Va a: all_logs

channel.basic_publish(exchange="logs_topic", routing_key="api.error.critical", body="API critical")
# → Va a: all_errors, all_logs

Fanout Exchange

Broadcastea mensajes a todas las queues bound, ignorando el routing key.

# Declarar un fanout exchange
channel.exchange_declare(exchange="notifications_fanout", exchange_type="fanout")

# Bindear queues (routing key es ignorado)
channel.queue_bind(exchange="notifications_fanout", queue="email_queue")
channel.queue_bind(exchange="notifications_fanout", queue="sms_queue")
channel.queue_bind(exchange="notifications_fanout", queue="push_queue")

# Publicar: todas las queues reciben el mensaje
channel.basic_publish(
    exchange="notifications_fanout",
    routing_key="",  # Ignorado para fanout
    body='{"user_id": 123, "message": "Order shipped"}'
)

Headers Exchange

Rutea basado en headers del mensaje en lugar de routing keys.

# Declarar un headers exchange
channel.exchange_declare(exchange="headers_exchange", exchange_type="headers")

# Bindear queues con matching de headers
channel.queue_bind(
    exchange="headers_exchange",
    queue="priority_orders",
    routing_key="",
    arguments={"x-match": "all", "priority": "high", "type": "order"}
)

channel.queue_bind(
    exchange="headers_exchange",
    queue="all_orders",
    routing_key="",
    arguments={"x-match": "any", "type": "order"}
)

# Publicar con headers
channel.basic_publish(
    exchange="headers_exchange",
    routing_key="",
    body='{"order_id": 123}',
    properties=pika.BasicProperties(
        headers={"priority": "high", "type": "order"}
    )
)

Features de Queue

Durable Queues

Las durable queues sobreviven reinicios de broker. Los mensajes marcados como persistent se escriben a disco.

# Declarar una durable queue
channel.queue_declare(queue="orders", durable=True)

# Publicar mensajes persistentes
channel.basic_publish(
    exchange="",
    routing_key="orders",
    body="order data",
    properties=pika.BasicProperties(delivery_mode=2)  # Persistent
)

Exclusive y Auto-Delete Queues

# Exclusive: solo accesible por la conexion que la declara, eliminada al desconectar
channel.queue_declare(queue="temp_queue", exclusive=True)

# Auto-delete: eliminada cuando el ultimo consumer se desconecta
channel.queue_declare(queue="task_queue", auto_delete=True)

Dead Letter Exchange (DLX)

Los mensajes que expiran, son rechazados, o exceden limites de longitud de queue se envian a un dead letter exchange.

# Declarar un dead letter exchange
channel.exchange_declare(exchange="orders_dlx", exchange_type="direct")

# Declarar una dead letter queue
channel.queue_declare(queue="orders_dead_letter")
channel.queue_bind(exchange="orders_dlx", queue="orders_dead_letter", routing_key="orders")

# Declarar la queue principal con configuracion DLX
args = {
    "x-dead-letter-exchange": "orders_dlx",
    "x-dead-letter-routing-key": "orders",
    "x-message-ttl": 60000,  # Mensajes expiran despues de 60 segundos
    "x-max-retries": 3       # Contador de retry custom
}
channel.queue_declare(queue="orders", arguments=args)

# Consumer con manejo de dead letter
def process_message(ch, method, properties, body):
    try:
        process_order(json.loads(body))
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        # Rechazar y requeuear si quedan retries, sino dead letter
        retries = properties.headers.get("x-retry-count", 0) if properties.headers else 0
        if retries < 3:
            ch.basic_publish(
                exchange="",
                routing_key="orders",
                body=body,
                properties=pika.BasicProperties(
                    headers={"x-retry-count": retries + 1},
                    delivery_mode=2
                )
            )
        ch.basic_ack(delivery_tag=method.delivery_tag)  # Ack original para remover de queue

Priority Queues

# Declarar una priority queue
args = {"x-max-priority": 10}
channel.queue_declare(queue="priority_orders", arguments=args)

# Publicar con prioridad
channel.basic_publish(
    exchange="",
    routing_key="priority_orders",
    body="urgent order",
    properties=pika.BasicProperties(priority=9)  # Numero mas alto = mayor prioridad
)

channel.basic_publish(
    exchange="",
    routing_key="priority_orders",
    body="normal order",
    properties=pika.BasicProperties(priority=1)
)

Patrones de Consumer

Work Queue (Competing Consumers)

Multiples consumers comparten una queue. Cada mensaje es procesado por exactamente un consumer.

# Consumer 1, 2, 3 todos consumen de la misma queue
def consume_tasks():
    channel.basic_qos(prefetch_count=1)  # Fair dispatch: un mensaje a la vez
    
    channel.basic_consume(
        queue="tasks",
        on_message_callback=process_task
    )
    channel.start_consuming()

def process_task(ch, method, properties, body):
    try:
        do_work(json.loads(body))
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception:
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

Publish/Subscribe

# Publisher
def publish_notification(message):
    channel.basic_publish(
        exchange="notifications",
        routing_key="",
        body=json.dumps(message)
    )

# Subscriber 1: Servicio de Email
def email_consumer():
    channel.queue_declare(queue="email_notifications", exclusive=True)
    channel.queue_bind(exchange="notifications", queue="email_notifications")
    channel.basic_consume(queue="email_notifications", on_message_callback=send_email)
    channel.start_consuming()

# Subscriber 2: Servicio de SMS
def sms_consumer():
    channel.queue_declare(queue="sms_notifications", exclusive=True)
    channel.queue_bind(exchange="notifications", queue="sms_notifications")
    channel.basic_consume(queue="sms_notifications", on_message_callback=send_sms)
    channel.start_consuming()

RPC (Request/Reply)

import uuid

# Client
class RPCClient:
    def __init__(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
        self.channel = self.connection.channel()
        result = self.channel.queue_declare(queue="", exclusive=True)
        self.callback_queue = result.method.queue
        self.channel.basic_consume(
            queue=self.callback_queue,
            on_message_callback=self.on_response,
            auto_ack=True
        )
    
    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body
    
    def call(self, message):
        self.response = None
        self.corr_id = str(uuid.uuid4())
        self.channel.basic_publish(
            exchange="",
            routing_key="rpc_queue",
            properties=pika.BasicProperties(
                reply_to=self.callback_queue,
                correlation_id=self.corr_id
            ),
            body=json.dumps(message)
        )
        while self.response is None:
            self.connection.process_data_events()
        return json.loads(self.response)

# Server
def on_request(ch, method, props, body):
    request = json.loads(body)
    response = process_request(request)
    ch.basic_publish(
        exchange="",
        routing_key=props.reply_to,
        properties=pika.BasicProperties(correlation_id=props.correlation_id),
        body=json.dumps(response)
    )
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue="rpc_queue", on_message_callback=on_request)
channel.start_consuming()

Clustering y Alta Disponibilidad

Setup de Cluster

# En rabbit2: unir al cluster de rabbit1
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@rabbit1
rabbitmqctl start_app

# Verificar status del cluster
rabbitmqctl cluster_status

# Output:
# Cluster status of node rabbit@rabbit2 ...
# Nodes: [rabbit@rabbit1, rabbit@rabbit2, rabbit@rabbit3]

Quorum Queues

Quorum queues proveen queues replicadas, durables con consenso Raft. Reemplazan a las classic mirrored queues.

# Declarar una quorum queue
channel.queue_declare(
    queue="orders",
    durable=True,
    arguments={"x-queue-type": "quorum"}
)

Mirrored Queues (Classic)

# Policy: mirrorar orders queue a todos los nodos
rabbitmqctl set_policy ha-orders "orders" \
  '{"ha-mode":"all","ha-sync-mode":"automatic"}'

# Policy: mirrorar a exactamente 2 nodos
rabbitmqctl set_policy ha-orders "orders" \
  '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

Tuning de Performance

Publisher Confirms

# Habilitar publisher confirms
channel.confirm_delivery()

try:
    channel.basic_publish(
        exchange="orders",
        routing_key="created",
        body="order data",
        properties=pika.BasicProperties(delivery_mode=2),
        mandatory=True  # Retornar si no hay queue bound
    )
    print("Message confirmed")
except pika.exceptions.UnroutableError:
    print("Message was not routed to any queue")

Optimizacion de Prefetch

# Prefetch count controla cuantos mensajes unacknowledged puede tener un consumer
channel.basic_qos(prefetch_count=10)  # Procesar hasta 10 mensajes concurrentemente

# Muy bajo: subutiliza el consumer
# Muy alto: distribucion injusta entre consumers
# Sweet spot: tipicamente 10-100 dependiendo del tiempo de procesamiento

Gestion de Connection y Channel

# Reusar conexiones, multiplexar con channels
connection = pika.BlockingConnection(pika.ConnectionParameters(
    host="rabbitmq",
    port=5672,
    virtual_host="/",
    credentials=pika.PlainCredentials("user", "password"),
    heartbeat=60,          # Keep alive
    blocked_connection_timeout=300  # Timeout si esta bloqueado
))

# Crear channels segun sea necesario (lightweight)
channel1 = connection.channel()  # Para publicar
channel2 = connection.channel()  # Para consumir

Monitoreo

Metricas Clave

Metrica	Descripcion	Threshold de Alerta
Queue depth	Mensajes ready en queue	> 10,000 sostenido
Consumer count	Consumers activos por queue	< 1 para queues criticas
Publish rate	Mensajes publicados por segundo	Baseline + 200%
Deliver rate	Mensajes entregados por segundo	< publish rate sostenido
Unacked messages	Mensajes esperando acknowledgment	> 5,000
Connection count	Conexiones abiertas	> 1000
Memory usage	Uso de RAM del broker	> 80% del watermark

Management API

import requests

# Obtener stats de queue via management API
response = requests.get(
    "http://rabbitmq:15672/api/queues",
    auth=("admin", "password")
)

for queue in response.json():
    print(f"Queue: {queue['name']}")
    print(f"  Messages: {queue['messages']}")
    print(f"  Consumers: {queue['consumers']}")
    print(f"  Unacked: {queue['messages_unacknowledged']}")

Checklist de Producción

Preguntas Frecuentes

¿Cuándo debería usar RabbitMQ vs Kafka?

Usa RabbitMQ para patrones de routing complejos (topic exchanges, fanout), RPC request/reply, y cuando necesitas acknowledgment por mensaje. Usa Kafka para streaming de alto throughput, event sourcing, y log aggregation donde el orden dentro de particiones importa mas que el routing complejo.

¿Cuál es la diferencia entre quorum queues y mirrored queues?

Quorum queues usan consenso Raft para replicacion, proporcionando garantias de consistencia mas fuertes. Mirrored queues (classic) usan un modelo master-slave. RabbitMQ recomienda quorum queues para nuevos deployments. Mirrored queues estan deprecadas en favor de quorum queues y streams.

¿Cómo manejo poison messages?

Usa un dead letter exchange. Configura la queue principal con x-dead-letter-exchange. Cuando un mensaje es rechazado (basic_nack sin requeue), expira, o excede el max delivery count, va al DLX. Monitorea la dead letter queue e investiga la causa.

¿Qué es prefetch count y cómo debería setearlo?

Prefetch count limita el numero de mensajes unacknowledged que un consumer puede tener. Setearlo muy bajo subutiliza el consumer; muy alto causa distribucion injusta. Empieza con 10 para la mayoria de workloads. Aumenta para consumers rapidos, disminuye para consumers lentos o cuando el orden importa.

¿Puede RabbitMQ garantizar entrega exactly-once?

No. RabbitMQ proporciona entrega at-least-once. Los mensajes pueden duplicarse si un consumer crashea despues de procesar pero antes de acknowledge. Haz los consumers idempotent trackeando IDs de mensajes procesados o usando logica de deduplication.

¿Cuántas conexiones y channels debería usar?

Usa una conexion long-lived por proceso y multiplexa con channels. Los channels son baratos (virtuales). Evita abrir una nueva conexion por request. Limita los channels a unas decenas por conexion. Monitorea el conteo de conexiones — demasiadas conexiones desperdician recursos.