MQTT asíncrono en Python con aiomqtt y asyncio: tareas, workers y reconexiones

aiomqtt con asyncio te ofrece MQTT en Python sin callbacks: haces await de una publicación, recorres los mensajes entrantes con async for y ejecutas tantas tareas concurrentes como necesites en un solo hilo. En este tutorial construirás un cliente asíncrono conectado a broker.connectmq.com que publica lecturas de sensores, consume mensajes mediante una cola de workers y se reconecta limpiamente cuando cae la red. También verás el detalle del bucle de eventos en Windows que hace tropezar a casi todo el mundo.

¿Por qué aiomqtt en lugar de paho-mqtt a secas?

paho-mqtt es una biblioteca basada en callbacks. Su bucle de red se ejecuta en su propio hilo y llama a tus funciones cuando ocurre algo. Para scripts pequeños está bien, pero resulta incómodo cuando tu aplicación ya está construida sobre asyncio, por ejemplo un servicio FastAPI, un gateway que también habla HTTP y Modbus de forma asíncrona o un bot. Mezclar hilos y un bucle de eventos implica run_coroutine_threadsafe, locks y errores sutiles al apagar.

aiomqtt envuelve paho-mqtt y expone en su lugar una API idiomática de asyncio:

Por debajo sigue siendo paho, así que el comportamiento del protocolo es idéntico. Si vienes de los callbacks, el tutorial de paho mqtt en Python es una referencia útil para los conceptos.

Instalar aiomqtt

python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install "aiomqtt>=2.0"

Los ejemplos usan asyncio.TaskGroup y except*, que requieren Python 3.11 o superior. Elige un prefijo de topic aleatorio para el servidor MQTT público gratuito de broker.connectmq.com y así tus pruebas no se mezclarán con las de otros usuarios; aquí usamos connectmq/demo/k7f3q2. Recuerda que ese servidor es solo para pruebas y prototipos, no para datos de producción.

La trampa del bucle de eventos en Windows

aiomqtt depende de los métodos add_reader y add_writer del bucle de eventos. En Windows el bucle por defecto es el Proactor, que no los implementa, así que obtienes un NotImplementedError en cuanto te conectas. Ejecuta tu programa sobre un bucle de eventos selector, como hacen los ejemplos de abajo. Linux y macOS ya usan un bucle selector por defecto.

Un ejemplo mínimo de aiomqtt con asyncio

Empieza con el suscriptor y el publicador más pequeños que funcionan, en un solo archivo:

import asyncio
import sys
import uuid

import aiomqtt

BROKER = "broker.connectmq.com"
PREFIX = "connectmq/demo/k7f3q2"


async def main():
    async with aiomqtt.Client(BROKER, 1883, identifier=f"aio-{uuid.uuid4().hex[:8]}") as client:
        await client.subscribe(f"{PREFIX}/hello", qos=1)
        await client.publish(f"{PREFIX}/hello", payload="hello from asyncio", qos=1)
        async for message in client.messages:
            print(message.topic, message.payload.decode())
            break  # parar tras el primer mensaje


if sys.platform == "win32":
    asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop)
else:
    asyncio.run(main())

El bloque async with se conecta al entrar y se desconecta al salir, incluso si se lanza una excepción. client.messages es un iterador asíncrono que entrega cada mensaje de cada suscripción en orden de llegada.

Un cliente realista: tareas concurrentes y una cola de workers

Las aplicaciones reales hacen varias cosas a la vez: publican lecturas periódicamente, reaccionan a comandos y procesan datos entrantes que pueden implicar E/S lenta, como una escritura en base de datos o una llamada HTTP. El truco es no bloquear nunca el bucle que lee los mensajes. Guarda esto como aio_client.py:

import asyncio
import json
import random
import sys
import time
import uuid

import aiomqtt

BROKER = "broker.connectmq.com"
PORT = 1883
PREFIX = "connectmq/demo/k7f3q2"
CLIENT_ID = f"aio-{uuid.uuid4().hex[:8]}"
RECONNECT_INTERVAL = 5
WORKERS = 3


async def publish_readings(client: aiomqtt.Client) -> None:
    seq = 0
    while True:
        reading = {"seq": seq, "temperature": round(random.uniform(20, 25), 2), "ts": time.time()}
        await client.publish(f"{PREFIX}/sensors/room1", json.dumps(reading), qos=1)
        seq += 1
        await asyncio.sleep(2)


async def consume(client: aiomqtt.Client, queue: asyncio.Queue) -> None:
    await client.subscribe(f"{PREFIX}/sensors/+", qos=1)
    await client.subscribe(f"{PREFIX}/commands/#", qos=1)
    async for message in client.messages:
        await queue.put(message)  # entregar rápido y seguir leyendo


async def handle(message: aiomqtt.Message) -> None:
    if message.topic.matches(f"{PREFIX}/commands/#"):
        print(f"Command on {message.topic}: {message.payload.decode()}")
        return
    data = json.loads(message.payload)
    await asyncio.sleep(0.5)  # simula una llamada lenta a base de datos o HTTP
    print(f"Processed reading {data['seq']}: {data['temperature']} C")


async def worker(name: int, queue: asyncio.Queue) -> None:
    while True:
        message = await queue.get()
        try:
            await handle(message)
        except Exception as exc:  # un mensaje erróneo no debe matar al worker
            print(f"Worker {name} failed on {message.topic}: {exc}")
        finally:
            queue.task_done()


async def main() -> None:
    queue: asyncio.Queue = asyncio.Queue(maxsize=1000)
    workers = [asyncio.create_task(worker(i, queue)) for i in range(WORKERS)]

    client = aiomqtt.Client(
        BROKER,
        PORT,
        identifier=CLIENT_ID,
        protocol=aiomqtt.ProtocolVersion.V5,
        keepalive=30,
        will=aiomqtt.Will(f"{PREFIX}/status/{CLIENT_ID}", payload="offline", qos=1, retain=True),
        # Broker privado de ConnectMQ:
        # username="YOUR_USERNAME", password="YOUR_PASSWORD",
        # tls_context=ssl.create_default_context(), y el puerto 8883
    )

    try:
        while True:
            try:
                async with client:
                    print("Connected")
                    await client.publish(f"{PREFIX}/status/{CLIENT_ID}", "online", qos=1, retain=True)
                    async with asyncio.TaskGroup() as tg:
                        tg.create_task(publish_readings(client))
                        tg.create_task(consume(client, queue))
            except* aiomqtt.MqttError as group:
                print(f"Connection lost: {group.exceptions[0]}. Reconnecting in {RECONNECT_INTERVAL}s")
            await asyncio.sleep(RECONNECT_INTERVAL)
    finally:
        for w in workers:
            w.cancel()


if __name__ == "__main__":
    try:
        if sys.platform == "win32":
            asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop)
        else:
            asyncio.run(main())
    except KeyboardInterrupt:
        print("Stopped")

Ejecuta python aio_client.py. Cada dos segundos se publica una lectura, que vuelve a recibirse a través de la suscripción y la procesa uno de los tres workers. Envía un comando desde otra terminal:

mosquitto_pub -h broker.connectmq.com -p 1883 -i aio-cmd-test -t 'connectmq/demo/k7f3q2/commands/led' -m 'on'

Cómo funciona

Concurrencia estructurada con TaskGroup

El publicador y el consumidor se ejecutan como tareas hermanas dentro de un TaskGroup. Si cualquiera de ellas lanza una excepción, por ejemplo porque se cayó la conexión y publish() falla, el grupo cancela la otra tarea y vuelve a lanzar el error envuelto en un grupo de excepciones. except* aiomqtt.MqttError lo captura sea cual sea la tarea que falló. Sin un task group, el fallo de una tarea puede pasar desapercibido mientras la otra sigue funcionando sobre una conexión muerta.

Mantener rápido el bucle de mensajes

El consumidor no hace nada más que mover mensajes a una asyncio.Queue. Los workers se encargan de la parte lenta. Esto importa porque aiomqtt entrega los mensajes en orden a través de un único iterador; si hicieras await de una escritura de medio segundo en base de datos dentro del bucle async for, todos los demás mensajes esperarían detrás. Con tres workers se procesan tres mensajes a la vez.

El maxsize de la cola aporta contrapresión. Si los workers no dan abasto, queue.put() espera, lo que ralentiza la lectura desde el broker en lugar de hacer crecer la memoria sin límite. Si importa el orden por topic, reparte los mensajes entre los workers según un hash del topic en lugar de usar una cola compartida.

Reconexiones limpias

Un cliente aiomqtt es reutilizable pero no reentrante: puedes volver a entrar en async with client cuando ha salido, pero no dos veces al mismo tiempo. El bucle exterior while True hace exactamente eso. Tras un fallo espera unos segundos, se reconecta, vuelve a publicar el estado online y se vuelve a suscribir, ya que las suscripciones pertenecen a la conexión. El Last Will publica "offline" si el proceso muere sin una desconexión limpia.

En producción, sustituye el intervalo fijo por un backoff exponencial limitado a uno o dos minutos, para que una flota de clientes no machaque al broker al unísono tras una caída. La guía de reconexión de paho-mqtt trata el backoff y la estrategia de sesiones con más detalle, y las mismas ideas se aplican aquí.

Usar propiedades de MQTT 5 con aiomqtt

Como el cliente anterior se conecta con ProtocolVersion.V5, puedes adjuntar propiedades de MQTT 5 a mensajes individuales. aiomqtt reutiliza la clase Properties de paho, así que el código es el mismo que en un cliente basado en callbacks:

from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

props = Properties(PacketTypes.PUBLISH)
props.MessageExpiryInterval = 60          # descartar si no se entrega en 60 s
props.ContentType = "application/json"
await client.publish(f"{PREFIX}/sensors/room1", payload, qos=1, properties=props)

La expiración de mensajes es útil para comandos: un "encender" que llega diez minutos tarde suele ser peor que uno que no llega nunca. En el lado receptor, message.properties expone los mismos atributos. La guía de funciones de MQTT 5 explica qué propiedades merece la pena usar.

Problemas habituales

1. NotImplementedError en Windows: estás en el bucle Proactor. Usa el bucle selector como se muestra.
2. MqttError: Could not connect: el puerto 1883 está bloqueado o el nombre de host es incorrecto. Prueba TLS en el 8883 con un tls_context.
3. La conexión se cae cada pocos segundos: otro cliente usa el mismo identificador. Genera uno único por proceso.
4. Los mensajes llegan tarde o a ráfagas: algo lento se ejecuta dentro de async for. Muévelo a los workers.
5. Limitación de tasa en el servidor público: permite 10 mensajes por segundo por cliente y 10 suscripciones por cliente. Mantén un intervalo de publicación razonable.
6. Las llamadas bloqueantes lo congelan todo: un requests.get() o time.sleep() síncrono dentro de una corrutina bloquea todo el bucle. Usa bibliotecas asíncronas o asyncio.to_thread().

Paso a producción

En un broker privado de ConnectMQ pasas el username y la password generados, el client ID como identifier y tls_context=ssl.create_default_context() en el puerto 8883. La guía de MQTT con TLS en Python explica la verificación de certificados, y el tutorial de JSON con MQTT en Python muestra cómo versionar payloads como las lecturas anteriores. Para almacenar lo que procesan los workers, adapta el logger de MQTT a SQLite en Python y, para comparar con clientes de otros lenguajes, consulta el resumen de bibliotecas cliente MQTT. El proyecto aiomqtt en GitHub documenta todas las opciones del cliente.

Ejecuta tus servicios Python asíncronos en ConnectMQ

ConnectMQ es un broker MQTT gestionado compatible con MQTT 3.1.1 y 5.0 sobre TCP, TLS y WebSocket, con session expiry, retained messages y Last Will, de los que depende este cliente. Desarrolla contra broker.connectmq.com y despliega después en un broker MQTT en la nube privado sin cambiar tu código más allá de los ajustes de conexión. Regístrate gratis en el plan Serverless, sin tarjeta de crédito.

Preguntas frecuentes

¿Es aiomqtt más rápido que paho-mqtt?

Usa paho-mqtt internamente, así que el rendimiento bruto del protocolo es similar. La ventaja es la estructura: el código asíncrono permite que un solo hilo gestione muchas operaciones concurrentes sin callbacks ni locks.

¿Por qué falla aiomqtt en Windows?

El bucle de eventos Proactor por defecto de Windows carece de los métodos add_reader y add_writer que necesita aiomqtt. Ejecuta tu programa sobre asyncio.SelectorEventLoop y funcionará.

¿Se reconecta aiomqtt automáticamente?

No. Una conexión caída lanza aiomqtt.MqttError, y te reconectas volviendo a entrar en async with client dentro de un bucle, lo que además te permite controlar el backoff y la resuscripción.

¿Puedo usar aiomqtt con FastAPI?

Sí. Inicia el cliente en el manejador lifespan de la aplicación y ejecuta el consumidor como tarea en segundo plano en el mismo bucle de eventos.