Primeros pasos con RabbitMQ usando Python

RabbitMQ es un broker de mensajería de código abierto que implementa el protocolo AMQP (Advanced Message Queuing Protocol). Permite la comunicación entre aplicaciones y servicios mediante el intercambio de mensajes. RabbitMQ es ampliamente utilizado en arquitecturas de microservicios y sistemas distribuidos debido a su capacidad para manejar grandes volúmenes de mensajes y su flexibilidad en la configuración.

Voy a enseñarte a iniciarte mediante unos pasos mínimos.

1. Levantar RabbitMQ con Docker

Creamos un archivo compose.yml en la raíz del proyecto y añadimos el siguiente contenido:

services:
  rabbitmq:
    image: rabbitmq
    ports:
      - 5672:5672
      - 15672:15672
    volumes:
      - "./rabbit_enabled_plugins:/etc/rabbitmq/enabled_plugins"

El puerto 5672 es el puerto por defecto de RabbitMQ para la comunicación de mensajes, mientras que el puerto 15672 es la interfaz de administración web. El volumen rabbit_enabled_plugins se utiliza para habilitar plugins adicionales en RabbitMQ.

2. Habilitar plugins

Creamos un archivo rabbit_enabled_plugins en la raíz del proyecto y añadimos el siguiente contenido:

[
rabbitmq_amqp1_0,
rabbitmq_management,
rabbitmq_web_dispatch,
rabbitmq_management_agent,
rabbitmq_stomp
].

Estamos habilitando los siguientes plugins:

• rabbitmq_amqp1_0: Protocolo AMQP 1.0.
• rabbitmq_management: Interfaz de administración web.
• rabbitmq_web_dispatch: Protocolo HTTP para la interfaz de administración.
• rabbitmq_management_agent: Agente de administración.
• rabbitmq_stomp: Protocolo STOMP (Streaming Text Oriented Messaging Protocol). Es opcional y se puede omitir si no lo necesitas.

3. Levantar RabbitMQ y crear queue

Ahora que tenemos todo configurado, podemos levantar RabbitMQ con el siguiente comando:

docker compose up

Esto iniciará RabbitMQ y podrás acceder a la interfaz de administración web en http://localhost:15672. Las credenciales por defecto son guest para el usuario y guest para la contraseña.

Dentro necesitarás crear una cola. Pulsa en la pestaña “Queues and Streams” y luego en “Add a new queue”. Introduce el nombre de la cola (por ejemplo, some-queue) y pulsa en “Add queue”.

En un futuro podrás crearla usando el cliente de Python, pero para este ejemplo lo haremos desde la interfaz web por comodidad.

4. Configurando Python

Daré por sentado que estas usando un gestor de paquetes moderno como es uv. En caso contrario puedes utilizar pip o pipenv para instalar las dependencias necesarias.

uv init
uv add rabbitmq-amqp-python-client
uv run rabbitmq-send.py

5. Enviando mensajes

Creamos un archivo rabbitmq-send.py en la raíz del proyecto y añadimos el siguiente contenido:

from rabbitmq_amqp_python_client import (
    Environment,
    Message,
    OutcomeState,
    AddressHelper,
)

queue_name = "some-queue"

# Crear una conexión a RabbitMQ
environment = Environment(uri="amqp://guest:guest@localhost:5672/")
connection = environment.connection()
connection.dial()

# Crear una cola
addr = AddressHelper.queue_address(queue_name)

# Crear un publicador o suscriptor
publisher = connection.publisher(addr)

# Creamos un mensaje
message = Message(body="Hello World!")

# Publicamos el mensaje 10 veces
for _ in range(10):
    status = publisher.publish(message)

    match status.remote_state:
        case OutcomeState.ACCEPTED:
            print("Message accepted")
        case OutcomeState.REJECTED:
            print("Message rejected")
        case OutcomeState.RELEASED:
            print("Message released")

# Cerrar la conexión
publisher.close()
connection.close()
environment.close()

Para lanzar el script, ejecuta el siguiente comando:

uv run rabbitmq-send.py

Se han enviado 10 mensajes a la cola some-queue. Puedes verificarlo en la interfaz de administración web de RabbitMQ.

Ahora vamos a recibirlos.

6. Recibiendo mensajes

Creamos un archivo rabbitmq-receive.py en la raíz del proyecto y añadimos el siguiente contenido:

from rabbitmq_amqp_python_client import (
    Environment,
    AddressHelper,
    Event,
    AMQPMessagingHandler,
)

# Crear una conexión a RabbitMQ
environment = Environment(uri="amqp://guest:guest@localhost:5672/")
connection = environment.connection()
connection.dial()

# Creamos una clase para manejar los mensajes
class MyMessageHandler(AMQPMessagingHandler):
    def __init__(self):
        super().__init__()
        self._count = 0

    def on_message(self, event: Event):
        print("received message: " + str(event.message.body))

        # accepting
        self.delivery_context.accept(event)

# Crear un publicador o suscriptor
queue_address = AddressHelper.queue_address("some-queue")

# Crear un consumidor con la clase MyMessageHandler
consumer = connection.consumer(queue_address, message_handler=MyMessageHandler())

# Iniciar el consumidor
consumer.run()

Ahora lo ejecutamos:

uv run rabbitmq-receive.py

Podrás visualizar los mensajes 10 mensajes que te estamos esperando pacientemente.

¡Buen trabajo!

Conclusiones

Aún queda mucho por aprender sobre RabbitMQ, como distribuir la carga entre varios consumidores, crear colas temporales, streams, exchanges, protocolos, plugins, etc. Es una herramienta muy potente y versátil que puede ayudarte a construir sistemas distribuidos, escalables, con arquitectura de microservicios, basados en eventos y mucho más.

¡Ten cuidado cuanto te metes en el agujero de conejo!