Modos de operación de RabbitMQ

Cola de mensajes (MQ)

1.1 ¿Qué es una cola de mensajes?

Una cola de mensajes es un mecanismo de comunicación entre aplicaciones. Los servicios intercambian datos y se comunican mediante la lectura y escritura de mensajes en una cola. Se utiliza ampliamente para el procesamiento asíncrono de tareas que requieren mucho tiempo y para el desacoplamiento entre aplicaciones.

1.2 Herramienta común: RabbitMQ (implementación del protocolo AMQP)

RabbitMQ

2.1 Instalación de RabbitMQ

# Instalación mediante contenedor Docker (recomendado)
docker pull rabbitmq
docker pull rabbitmq:management # Versión con interfaz gráfica

# Windows 10 - Descarga e instalación oficial (requiere entorno Erlang)
https://www.erlang.org/downloads
https://www.rabbitmq.com/download.html

# Linux Ubuntu
sudo apt-get install rabbitmq-server

2.2 Inicio de RabbitMQ (Docker recomendado)

# Ejecutar el contenedor
docker run -it --name mq -p 5672:5672 -p 15672:15672 rabbitmq:management

# Credenciales por defecto: usuario guest, contraseña guest
# Puerto 5672: para el envío de mensajes al servidor RabbitMQ
# Puerto 15672: para la interfaz web de administración

2.3 Conexión de Python con RabbitMQ

# Instalación de la librería Pika para Python
pip install pika

# Documentación de desarrollo: https://www.rabbitmq.com/tutorials/tutorial-one-python.html

2.4 Arquitectura de RabbitMQ (Proudctor-Broker-Consumidor)

Componentes principales

Broker: Proceso del servidor de cola de mensajes, incluye Exchange y Queue.
Exchange: Conmutador de mensajes que enruta mensajes a una cola específica según reglas predefinidas.
Queue: Cola que almacena mensajes hasta que son enviados al consumidor designado.
Producer: Cliente productor que envía mensajes al broker.
Consumer: Cliente consumidor que recibe mensajes del broker.

Flujo de envío de mensajes

El productor establece una conexión TCP con el broker.
El productor abre un canal de comunicación.
El productor envía mensajes al broker a través del canal, los cuales son enrutados por el Exchange.
El Exchange enruta el mensaje a la cola correspondiente.

Flujo de recepción de mensajes

El consumidor estbalece una conexión TCP con el broker.
El consumidor se suscribe a una cola específica.
Cuando hay mensajes en la cola, el broker los envía al consumidor de forma predeterminada.
El consumidor recibe y procesa el mensaje.

2.4.1 Modelo de mensajes básico

Aplicación: Operaciones que consumen mucho tiempo, como servicios de SMS durante el registro de usuarios.

Script de envío (sender.py)

import pika

# Establecer conexión con el broker
conn = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = conn.channel()

# Declarar la cola para mensajes
channel.queue_declare(queue='mensajes_basicos')

# Enviar mensaje a la cola
channel.basic_publish(exchange='', routing_key='mensajes_basicos', body='¡Hola Mundo!')
print("[*] Mensaje enviado: ¡Hola Mundo!")

# Cerrar conexión
conn.close()

Script de recepción (receiver.py)

import pika
import sys

def procesar_mensaje(ch, method, properties, body):
    print("[*] Mensaje recibido: %r" % body)

# Establecer conexión
conn = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
canal = conn.channel()

# Declarar la cola
canal.queue_declare(queue='mensajes_basicos')

# Configurar consumidor
canal.basic_consume(queue='mensajes_basicos', on_message_callback=procesar_mensaje, auto_ack=True)

print('[*] Esperando mensajes. Presione CTRL+C para salir')
canal.start_consuming()

Resumen: En el modelo básico, el productor envía mensajes a una cola específica y el consumidor los recibe de esa cola.

2.4.2 Modelo de trabajo (Work Queue)

Aplicación: Pedidos en plataformas de comercio electrónico, donde múltiples usuarios pueden sloicitar el mismo pedido. La tarea se completa cuando cualquier usuario finaliza su solicitud.

Script de tarea (task_generator.py)

import pika

# Conexión con RabbitMQ
conexion = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
canal = conexion.channel()

# Declarar cola persistente
canal.queue_declare(queue='tareas_pendientes', durable=True)

# Generar múltiples tareas
for num_tarea in range(1, 21):
    mensaje = f'Tarea numero {num_tarea}'
    canal.basic_publish(
        exchange='',
        routing_key='tareas_pendientes',
        body=mensaje,
        properties=pika.BasicProperties(delivery_mode=2)
    )
    print(f'[+] Tarea enviada: {mensaje}')

conexion.close()

Script de trabajador (worker.py)

import pika
import time

# Conexión con el servidor
conexion = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
canal = conexion.channel()

# Declarar cola persistente
canal.queue_declare(queue='tareas_pendientes', durable=True)

def ejecutar_tarea(ch, method, properties, body):
    print(f'[*] Procesando tarea: {body.decode()}')
    time.sleep(2)  # Simular procesamiento
    print('[*] Tarea completada')
    ch.basic_ack(delivery_tag=method.delivery_tag)

# Configurar prefetch para procesamiento secuencial
canal.basic_qos(prefetch_count=1)

# Suscribirse a la cola
canal.basic_consume(queue='tareas_pendientes', on_message_callback=ejecutar_tarea)

print('[*] Trabajador listo. Presione CTRL+C para salir')
canal.start_consuming()

Resumen: Los trabajadores procesan tareas de la cola. Con prefetch_count=1 se garantiza procesamiento secuencial, distribuyendo la carga de manera equitativa.

2.4.3 Modo Fanout (difusión)

Los Exchanges en RabbitMQ no almacenan mensajes, solo los enrutan. Los tipos principales son: Fanout (difusión), Direct (enrutamiento directo), Topic (patrones) y Headers (cabeceras).

Aplicación: Actualización simultánea de caché y base de datos para información de productos.

Productor (publisher.py)

import pika

conexion = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
canal = conexion.channel()

# Declarar exchange tipo fanout
canal.exchange_declare(exchange='difusion_productos', exchange_type='fanout')

mensaje = 'Nuevo producto disponible: Smartphone XYZ'
canal.basic_publish(exchange='difusion_productos', routing_key='', body=mensaje)
print(f'[+] Mensaje enviado: {mensaje}')

conexion.close()

Consumidor 1 (cache_consumer.py)

import pika

conexion = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
canal = conexion.channel()

canal.exchange_declare(exchange='difusion_productos', exchange_type='fanout')

# Cola temporal exclusiva
resultado = canal.queue_declare(queue='', exclusive=True)
nombre_cola = resultado.method.queue

# Vincular cola al exchange
canal.queue_bind(exchange='difusion_productos', queue=nombre_cola)

def actualizar_cache(ch, method, properties, body):
    print(f'[Cache] Actualizando con: {body.decode()}')

canal.basic_consume(queue=nombre_cola, on_message_callback=actualizar_cache, auto_ack=True)

print('[Cache] Esperando mensajes...')
canal.start_consuming()

Consumidor 2 (database_consumer.py)

import pika

conexion = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
canal = conexion.channel()

canal.exchange_declare(exchange='difusion_productos', exchange_type='fanout')

resultado = canal.queue_declare(queue='', exclusive=True)
nombre_cola = resultado.method.queue

canal.queue_bind(exchange='difusion_productos', queue=nombre_cola)

def actualizar_bd(ch, method, properties, body):
    print(f'[BD] Guardando: {body.decode()}')

canal.basic_consume(queue=nombre_cola, on_message_callback=actualizar_bd, auto_ack=True)

print('[BD] Esperando mensajes...')
canal.start_consuming()

Resumen: El exchange Fanout envía mensajes a todas las colas vinculadas, permitiendo difusión a múltiples consumidores.

2.4.4 Modo Direct (enrutamiento directo)

Aplicación: Categorización de productos en comercio electrónico según criterios específicos.

Productor (router.py)

import pika

conexion = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
canal = conexion.channel()

canal.exchange_declare(exchange='categorias_productos', exchange_type='direct')

categorias = ['electronica', 'hogar', 'deportes']

for cat in categorias:
    mensaje = f'Nuevo producto en categoría: {cat}'
    canal.basic_publish(exchange='categorias_productos', routing_key=cat, body=mensaje)
    print(f'[+] Enrutado a {cat}: {mensaje}')

conexion.close()

Consumidor para electrónica (electronics_consumer.py)

import pika

conexion = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
canal = conexion.channel()

canal.exchange_declare(exchange='categorias_productos', exchange_type='direct')

resultado = canal.queue_declare(queue='', exclusive=True)
nombre_cola = resultado.method.queue

# Vincular solo a categoría electrónica
canal.queue_bind(exchange='categorias_productos', queue=nombre_cola, routing_key='electronica')

def procesar_electronica(ch, method, properties, body):
    print(f'[Electrónica] {body.decode()}')

canal.basic_consume(queue=nombre_cola, on_message_callback=procesar_electronica, auto_ack=True)

print('[Electrónica] Esperando productos...')
canal.start_consuming()

Resumen: El modo Direct enruta mensajes a colas específicas basadas en claves de enrutamiento exactas.

2.4.5 Modo Topic (patrones)

El modo Topic permite enrutamiento flexible mediante patrones con comodines: * (una palabra) y # (cero o más palabras).

Productor (event_publisher.py)

import pika

conexion = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
canal = conexion.channel()

canal.exchange_declare(exchange='eventos_sistema', exchange_type='topic')

eventos = [
    'sistema.error.critico',
    'sistema.info.general',
    'app.usuario.login',
    'app.pedido.creado',
    'pagos.transaccion.exitosa'
]

for evento in eventos:
    mensaje = f'Evento generado: {evento}'
    canal.basic_publish(exchange='eventos_sistema', routing_key=evento, body=mensaje)
    print(f'[+] Publicado: {evento}')

conexion.close()

Consumidor de errores (error_consumer.py)

import pika

conexion = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
canal = conexion.channel()

canal.exchange_declare(exchange='eventos_sistema', exchange_type='topic')

resultado = canal.queue_declare(queue='', exclusive=True)
nombre_cola = resultado.method.queue

# Suscribirse a todos los errores
canal.queue_bind(exchange='eventos_sistema', queue=nombre_cola, routing_key='sistema.error.*')

def registrar_error(ch, method, properties, body):
    print(f'[ERROR] {body.decode()}')

canal.basic_consume(queue=nombre_cola, on_message_callback=registrar_error, auto_ack=True)

print('[ERROR] Monitoreando errores del sistema...')
canal.start_consuming()

Resumen: El modo Topic permite suscripciones flexibles basadas en patrones jerárquicos de enrutamiento.

2.4.6 Modo Headers (cabeceras)

Este modo enruta mensajes basándose en las cabeceras del mensaje, no en la clave de enrutamiento.

Productor (message_sender.py)

import pika

conexion = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
canal = conexion.channel()

canal.exchange_declare(exchange='mensajes_prioridad', exchange_type='headers', durable=True)

# Enviar mensaje con cabeceras específicas
cabeceras = {'tipo': 'urgente', 'departamento': 'ventas'}
mensaje = 'Solicitud de reembolso pendiente'

canal.basic_publish(
    exchange='mensajes_prioridad',
    routing_key='',
    body=mensaje,
    properties=pika.BasicProperties(headers=cabeceras)
)
print(f'[+] Mensaje enviado con cabeceras: {cabeceras}')

conexion.close()

Consumidor para mensajes urgentes (urgent_consumer.py)

import pika

conexion = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
canal = conexion.channel()

canal.exchange_declare(exchange='mensajes_prioridad', exchange_type='headers', durable=True)

resultado = canal.queue_declare(queue='', exclusive=True)
nombre_cola = resultado.method.queue

# Vincular con condición de coincidencia parcial
args = {'tipo': 'urgente', 'x-match': 'any'}
canal.queue_bind(exchange='mensajes_prioridad', queue=nombre_cola, routing_key='', arguments=args)

def procesar_urgente(ch, method, properties, body):
    print(f'[URGENTE] {body.decode()}')
    print(f'  Cabeceras: {properties.headers}')

canal.basic_consume(queue=nombre_cola, on_message_callback=procesar_urgente, auto_ack=True)

print('[URGENTE] Esperando mensajes prioritarios...')
canal.start_consuming()

Resumen: El modo Headers permite enrutamiento basado en metadatos del mensaje, útil para clasificaciones complejas.

2.5 Recursos adicionales

Documentación oficial: https://www.rabbitmq.com/getstarted.html

Etiquetas: RabbitMQ AMQP cola de mensajes patrones de mensajería mensajería asíncrona

Publicado el 7-10 21:27

Friki Work