MQTT Assíncrono em Python com aiomqtt e asyncio: Tasks, Workers e Reconexões

O aiomqtt com asyncio oferece MQTT em Python sem callbacks: você faz await em um publish, percorre as mensagens recebidas com async for e executa quantas tasks concorrentes precisar em uma única thread. Neste tutorial você vai construir um cliente assíncrono conectado ao broker.connectmq.com, o servidor MQTT público da ConnectMQ, que publica leituras de sensores, consome mensagens por meio de uma fila de workers e reconecta de forma limpa quando a rede cai. Você também vai conhecer o detalhe do event loop no Windows que pega quase todo mundo.

Por que aiomqtt em vez do paho-mqtt puro?

O paho-mqtt é uma biblioteca baseada em callbacks. O network loop dele roda em uma thread própria e chama as suas funções quando algo acontece. Isso funciona bem em scripts pequenos, mas fica desajeitado quando a sua aplicação já é construída sobre asyncio, por exemplo um serviço FastAPI, um gateway de chão de fábrica, numa planta em Joinville por exemplo, que também fala HTTP e Modbus de forma assíncrona, ou um bot. Misturar threads e um event loop significa run_coroutine_threadsafe, locks e bugs sutis no encerramento.

O aiomqtt encapsula o paho-mqtt e expõe, em vez disso, uma API idiomática de asyncio:

Por baixo dos panos ainda é o paho, então o comportamento do protocolo é idêntico. Se você vem dos callbacks, o tutorial de Paho MQTT em Python é uma referência útil para os conceitos.

Instale o aiomqtt

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

Os exemplos usam asyncio.TaskGroup e except*, que exigem Python 3.11 ou mais recente. Escolha um prefixo de topic aleatório para o servidor MQTT público (gratuito), para que os seus testes fiquem separados dos de outros usuários (ele serve para testes e protótipos, não para dados de produção); aqui usamos connectmq/demo/k7f3q2.

A pegadinha do event loop no Windows

O aiomqtt depende dos métodos add_reader e add_writer do event loop. No Windows, o event loop padrão é o Proactor, que não os implementa, e você recebe um NotImplementedError no instante em que conecta. Execute o seu programa em um selector event loop, como fazem os exemplos abaixo. Linux e macOS já usam um selector loop por padrão.

Um exemplo mínimo de aiomqtt com asyncio

Comece pelo menor subscriber e publisher funcionais em um único arquivo:

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  # para depois da primeira mensagem


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

O bloco async with conecta na entrada e desconecta na saída, mesmo que uma exceção seja lançada. client.messages é um iterador assíncrono que entrega cada mensagem de todas as assinaturas, na ordem de chegada.

Um cliente realista: tasks concorrentes e uma fila de workers

Aplicações reais fazem várias coisas ao mesmo tempo: publicam leituras em intervalos, reagem a comandos e processam dados recebidos que podem envolver I/O lento, como uma gravação em banco de dados ou uma chamada HTTP. O segredo é nunca bloquear o loop que lê as mensagens. Salve isto 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)  # repassa rápido e continua lendo


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 uma chamada lenta a banco de dados ou 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:  # uma mensagem ruim não pode derrubar o 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 ConnectMQ privado:
        # username="YOUR_USERNAME", password="YOUR_PASSWORD",
        # tls_context=ssl.create_default_context(), e a porta 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")

Execute python aio_client.py. A cada dois segundos uma leitura é publicada, recebida de volta pela assinatura e processada por um dos três workers. Envie um comando a partir de outro terminal:

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

Como funciona

Concorrência estruturada com TaskGroup

O publisher e o consumidor rodam como tasks irmãs dentro de um TaskGroup. Se qualquer uma delas lançar uma exceção, por exemplo porque a conexão caiu e o publish() falhou, o grupo cancela a outra task e relança o erro embrulhado em um exception group. except* aiomqtt.MqttError o captura, não importa qual task falhou. Sem um task group, uma falha em uma task pode passar despercebida enquanto a outra continua rodando sobre uma conexão morta.

Mantendo o loop de mensagens rápido

O consumidor não faz nada além de mover as mensagens para uma asyncio.Queue. Os workers fazem a parte lenta. Isso importa porque o aiomqtt entrega as mensagens em ordem por um único iterador; se você esperasse uma gravação de meio segundo no banco dentro do loop async for, todas as outras mensagens ficariam esperando atrás dela. Com três workers, três mensagens são processadas ao mesmo tempo.

O maxsize da fila cria backpressure. Se os workers não derem conta, queue.put() espera, o que desacelera a leitura do broker em vez de fazer a memória crescer sem limite. Se a ordem por topic importa, distribua as mensagens aos workers por hash do topic em vez de usar uma fila compartilhada.

Reconexões elegantes

Um cliente aiomqtt é reutilizável, mas não reentrante: você pode entrar em async with client de novo depois que ele sai, mas não duas vezes ao mesmo tempo. O while True externo faz exatamente isso. Após uma falha, ele espera alguns segundos, reconecta, republica o status online e assina de novo, já que as assinaturas pertencem à conexão. O Last Will publica "offline" se o processo morrer sem uma desconexão limpa.

Em produção, troque o intervalo fixo por backoff exponencial limitado a um ou dois minutos, para que uma frota de clientes não martele o broker em sincronia depois de uma queda. O guia de reconexão do paho-mqtt discute backoff e estratégia de sessão em mais profundidade, e as mesmas ideias valem aqui.

Usando properties do MQTT 5 com aiomqtt

Como o cliente acima conecta com ProtocolVersion.V5, você pode anexar properties do MQTT 5 a mensagens individuais. O aiomqtt reaproveita a classe Properties do paho, então o código é o mesmo de um cliente baseado em callbacks:

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

props = Properties(PacketTypes.PUBLISH)
props.MessageExpiryInterval = 60          # descarta se não for entregue em 60 s
props.ContentType = "application/json"
await client.publish(f"{PREFIX}/sensors/room1", payload, qos=1, properties=props)

A expiração de mensagens é útil para comandos: um "ligar" que chega dez minutos atrasado costuma ser pior do que um que nunca chega. No lado de quem recebe, message.properties expõe os mesmos atributos. O guia de recursos do MQTT 5 explica quais properties valem a pena.

Problemas comuns

1. NotImplementedError no Windows: você está no loop Proactor. Use o selector loop como mostrado.
2. MqttError: Could not connect: a porta 1883 está bloqueada ou o hostname está errado. Tente TLS na 8883 com um tls_context.
3. A conexão cai a cada poucos segundos: um segundo cliente está usando o mesmo identifier. Gere um único por processo.
4. Mensagens chegam atrasadas ou em rajadas: algo lento está rodando dentro do async for. Mova para os workers.
5. Limite de taxa no servidor MQTT público: ele permite 10 mensagens por segundo por cliente e 10 assinaturas por cliente. Mantenha um intervalo de publicação razoável.
6. Chamadas bloqueantes congelam tudo: um requests.get() ou time.sleep() síncrono dentro de uma coroutine bloqueia o loop inteiro. Use bibliotecas assíncronas ou asyncio.to_thread().

Indo para produção

Em um broker MQTT gerenciado privado da ConnectMQ, você passa o username, o password gerados e o client ID como identifier, além de tls_context=ssl.create_default_context() na porta 8883. O guia de TLS com MQTT em Python explica a verificação de certificados, e o tutorial de JSON com MQTT em Python mostra como versionar payloads como as leituras acima. Para armazenar o que os workers processam, adapte o logger de MQTT para SQLite em Python e, para comparar com clientes de outras linguagens, veja a visão geral de bibliotecas de cliente MQTT. O projeto aiomqtt no GitHub documenta todas as opções do cliente.

Rode os seus serviços Python assíncronos no ConnectMQ

O ConnectMQ é um broker MQTT gerenciado que suporta MQTT 3.1.1 e 5.0 sobre TCP, TLS e WebSocket, com session expiry, retained messages e Last Will, dos quais este cliente depende. Desenvolva com o broker.connectmq.com e depois implante em um broker MQTT na nuvem privado sem mudar o código além das configurações de conexão. Cadastre-se gratuitamente no plano Serverless, sem cartão de crédito.

Perguntas frequentes

O aiomqtt é mais rápido que o paho-mqtt?

Ele usa o paho-mqtt internamente, então o throughput bruto do protocolo é semelhante. A vantagem é a estrutura: código assíncrono permite que uma única thread cuide de muitas operações concorrentes sem callbacks nem locks.

Por que o aiomqtt falha no Windows?

O event loop Proactor, padrão no Windows, não tem os métodos add_reader e add_writer de que o aiomqtt precisa. Execute o programa em asyncio.SelectorEventLoop e ele funciona.

O aiomqtt reconecta automaticamente?

Não. Uma conexão perdida lança aiomqtt.MqttError, e você reconecta entrando de novo em async with client dentro de um loop, o que também permite controlar o backoff e a nova assinatura.

Posso usar aiomqtt com FastAPI?

Sim. Inicie o cliente no handler de lifespan da aplicação e rode o consumidor como uma task em segundo plano no mesmo event loop.