RabbitMQ: un agente de mensajería y streaming confiable

RabbitMQ es un message broker (intermediario de mensajes) que facilita el intercambio de datos entre productores (quienes envían mensajes) y consumidores (quienes los reciben). Los mensajes se almacenan en colas hasta que son procesados, garantizando una comunicación asíncrona, confiable y escalable. funciona como un middleware de mensajería. Implementa el estándar Advanced Message Queuing Protocol (AMQP).

El servidor RabbitMQ está escrito en Erlang y utiliza el framework Open Telecom Platform (OTP) para construir sus capacidades de ejecución distribuida y conmutación ante errores.

Además es multiplataforma, compatible con lenguajes como Java, Python, Go, etc. Soporta varios protocolos además de AMQP, como MQTT y STOMP. Puede manejar grandes volúmenes de mensajes y configurarse para alta disponibilidad. Incluye una consola web para monitorear y gestionar colas.

RabbitMQ se utiliza principalmente para:

• Desacoplar sistemas: Permite que diferentes aplicaciones o microservicios se comuniquen sin estar directamente conectados, mejorando la modularidad y la escalabilidad. 
• Procesamiento asíncrono: En lugar de procesar tareas en tiempo real, los mensajes se encolan y se procesan cuando los consumidores están listos, ideal para tareas como procesamiento en segundo plano. 
• Distribución de cargas: Permite distribuir tareas entre múltiples trabajadores (workers) para balancear la carga de trabajo. 
• Alta disponibilidad: Soporta configuraciones de clústeres y colas replicadas para garantizar que los mensajes no se pierdan, incluso si un servidor falla. 
• Soporte para múltiples patrones: Facilita patrones de mensajería como publicación/suscripción (pub/sub), enrutamiento basado en reglas, y colas de trabajo (work queues).

Se puede usar para:

• Microservicios: Comunicación entre servicios en arquitecturas distribuidas. 
• Procesamiento de tareas pesadas: Por ejemplo, enviar correos electrónicos, procesar imágenes o ejecutar análisis de datos en segundo plano. 
• Sistemas en tiempo real: Como notificaciones o actualizaciones en aplicaciones web. 
• Integración de sistemas: Conectar aplicaciones heterogéneas que usan diferentes tecnologías.

El proyecto RabbitMQ consta de diferentes partes:

1. El servidor de intercambio RabbitMQ, que recibe los mensajes y los almacena en colas de intercambio. Los clientes podrán después recuperarlos. 
2. Pasarelas para los protocolos HTTP, XMPP y STOMP. 
3. Bibliotecas de clientes para Java, framework .NET y Python. 
4. El plugin Shovel (en inglés: pala) que se encarga de replicar mensajes entre corredores de mensajes. Ejemplos

Creando un emisor y receptor de mensajes con RabbitMQ & Python

Debemos instalar Erlang, ya que RabbitMQ está escrito en este lenguaje.

https://www.erlang.org/downloads

Debemos instalar RabbitMQ:

https://www.rabbitmq.com/docs/download

En Windows, agregar la carpeta sbin de RabbitMQ (por ejemplo, C:\Program Files\RabbitMQ Server\rabbitmq_server-3.13.x\sbin) a la variable de entorno PATH para ejecutar comandos como rabbitmqctl desde cualquier terminal.

El Linux (Debian/Ubuntu):

wget https://packages.erlang-solutions.com/erlang-solutions_2.0_all.deb
sudo dpkg -i erlang-solutions_2.0_all.deb
sudo apt update
sudo apt install esl-erlang

En Windows, iniciamos y detenemos el servicio de esta forma:

$ net start RabbitMQ
$ net stop RabbitMQ

Si no usamos el servicio:

$ cd "C:\Program Files\RabbitMQ Server\rabbitmq_server-3.13.x\sbin"
$ rabbitmq-server.bat

En Linux, iniciamos y detenemos el servicio de esta forma:

$ sudo systemctl start rabbitmq-server
$ sudo systemctl stop rabbitmq-server
$ sudo systemctl status rabbitmq-server

Además de verificar el estado.

Si no usamos el servicio:

$ sudo rabbitmq-server

Instalamos Pika para implementar un emisor (productor) y receptor (consumidor) de mensajes:

$ pip install pika

Código del Emisor

send.py

import pika

# Establecer conexión con RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.create_channel()

# Declarar una cola llamada 'hello'
channel.queue_declare(queue='hello')

# Enviar un mensaje a la cola
message = "¡Hola, RabbitMQ!"
channel.basic_publish(exchange='',
                     routing_key='hello',  # Nombre de la cola
                     body=message)

print(f" [x] Enviado: '{message}'")

# Cerrar la conexión
connection.close()

Código del Receptor:

receive.py

import pika

# Callback que se ejecuta cuando se recibe un mensaje
def callback(ch, method, properties, body):
    print(f" [x] Recibido: {body.decode()}")

# Establecer conexión con RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.create_channel()

# Declarar la misma cola 'hello'
channel.queue_declare(queue='hello')

# Configurar el consumidor para escuchar mensajes
channel.basic_consume(queue='hello',
                      auto_ack=True,  # Confirma automáticamente la recepción del mensaje
                      on_message_callback=callback)

print(' [*] Esperando mensajes. Presiona CTRL+C para salir.')
# Iniciar la escucha de mensajes
channel.start_consuming()

Antes de ejecutar estos programas debemos asegurarnos que el servidor de RabbitMQ este funcionando.

Windows:

$ rabbitmqctl.bat status

Linux:

$  sudo rabbitmqctl status

Ejecutar receptor:

$ python receive.py

Ejecutar emisor:

$ python send.py

RabbitMQ es una herramienta poderosa para gestionar la comunicación asíncrona en sistemas distribuidos, asegurando que los mensajes se entreguen de manera confiable y eficiente.

Continuaremos sobre este tema en próximas entregas.

Enlaces:

https://www.rabbitmq.com/
https://es.wikipedia.org/wiki/RabbitMQ