MQTT asynchrone en Python avec aiomqtt et asyncio : tâches, workers et reconnexions

aiomqtt avec asyncio vous donne MQTT en Python sans callbacks : vous faites un await sur un publish, vous parcourez les messages entrants avec async for et vous exécutez autant de tâches concurrentes que nécessaire sur un seul thread. Dans ce tutoriel, vous allez construire un client asynchrone connecté à broker.connectmq.com qui publie des mesures de capteurs, consomme les messages via une file de workers et se reconnecte proprement quand le réseau tombe. Vous verrez aussi le détail de la boucle d'événements sous Windows qui piège presque tout le monde.

Pourquoi aiomqtt plutôt que paho-mqtt seul ?

paho-mqtt est une bibliothèque à callbacks. Sa boucle réseau tourne dans son propre thread et appelle vos fonctions lorsqu'il se passe quelque chose. C'est très bien pour de petits scripts, mais cela devient malcommode quand votre application repose déjà sur asyncio, par exemple un service FastAPI, une passerelle qui parle aussi HTTP et Modbus de façon asynchrone, ou un bot. Mélanger threads et boucle d'événements implique run_coroutine_threadsafe, des verrous et des bugs d'arrêt subtils.

aiomqtt encapsule paho-mqtt et expose à la place une API asyncio idiomatique :

Sous le capot, c'est toujours paho, donc le comportement protocolaire est identique. Si vous venez du monde des callbacks, le tutoriel paho mqtt python est une référence utile pour les concepts.

Installer aiomqtt

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

Les exemples utilisent asyncio.TaskGroup et except*, qui nécessitent Python 3.11 ou plus récent. Choisissez un préfixe de topic aléatoire pour le serveur MQTT public gratuit, réservé aux tests et aux prototypes, afin que vos essais restent séparés de ceux des autres utilisateurs ; ici, nous utilisons connectmq/demo/k7f3q2.

Le piège de la boucle d'événements sous Windows

aiomqtt s'appuie sur les méthodes add_reader et add_writer de la boucle d'événements. Sous Windows, la boucle par défaut est la boucle Proactor, qui ne les implémente pas : vous obtenez donc une NotImplementedError dès la connexion. Exécutez plutôt votre programme sur une boucle selector, comme le font les exemples ci-dessous. Linux et macOS utilisent déjà une boucle selector par défaut.

Un exemple minimal aiomqtt asyncio

Commencez par le plus petit subscriber et publisher fonctionnels, dans un seul fichier :

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  # s'arrêter après le premier message


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

Le bloc async with se connecte à l'entrée et se déconnecte à la sortie, même si une exception est levée. client.messages est un itérateur asynchrone qui renvoie chaque message de chaque abonnement dans l'ordre d'arrivée.

Un client réaliste : tâches concurrentes et file de workers

Les applications réelles font plusieurs choses à la fois : publier des mesures à intervalle régulier, réagir à des commandes et traiter des données entrantes qui peuvent impliquer des E/S lentes, comme une écriture en base de données ou un appel HTTP. L'astuce consiste à ne jamais bloquer la boucle qui lit les messages. Enregistrez ceci sous 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)  # transmettre vite, continuer à lire


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)  # simule un appel lent à une base de données 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:  # un message invalide ne doit pas tuer le 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 privé :
        # username="YOUR_USERNAME", password="YOUR_PASSWORD",
        # tls_context=ssl.create_default_context(), et port 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")

Lancez python aio_client.py. Toutes les deux secondes, une mesure est publiée, reçue en retour via l'abonnement et traitée par l'un des trois workers. Envoyez une commande depuis un autre terminal :

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

Comment ça fonctionne

Concurrence structurée avec TaskGroup

Le publisher et le consumer s'exécutent comme des tâches sœurs dans un TaskGroup. Si l'une d'elles lève une exception, par exemple parce que la connexion est tombée et que publish() échoue, le groupe annule l'autre tâche et relève l'erreur enveloppée dans un groupe d'exceptions. except* aiomqtt.MqttError l'intercepte, quelle que soit la tâche qui a échoué. Sans task group, l'échec d'une tâche peut passer inaperçu pendant que l'autre continue de tourner sur une connexion morte.

Garder la boucle de messages rapide

Le consumer ne fait rien d'autre que déplacer les messages dans une asyncio.Queue. Ce sont les workers qui s'occupent de la partie lente. C'est important, car aiomqtt livre les messages dans l'ordre via un itérateur unique : si vous attendiez une écriture en base d'une demi-seconde dans la boucle async for, tous les autres messages patienteraient derrière. Avec trois workers, trois messages sont traités simultanément.

Le maxsize de la file fournit une contre-pression (backpressure). Si les workers ne suivent pas, queue.put() attend, ce qui ralentit la lecture depuis le broker au lieu de faire grossir la mémoire sans limite. Si l'ordre par topic est important, répartissez les messages entre les workers selon un hachage du topic plutôt que via une file partagée.

Des reconnexions propres

Un client aiomqtt est réutilisable mais pas réentrant : vous pouvez réentrer dans async with client après sa sortie, mais pas deux fois en même temps. La boucle extérieure while True fait exactement cela. Après un échec, elle attend quelques secondes, se reconnecte, republie le statut online et se réabonne, puisque les abonnements appartiennent à la connexion. Le Last Will publie « offline » si le processus meurt sans déconnexion propre.

En production, remplacez l'intervalle fixe par un backoff exponentiel plafonné à une ou deux minutes, afin qu'une flotte de clients ne martèle pas le broker en même temps après une panne. Le guide de reconnexion paho-mqtt traite plus en détail du backoff et de la stratégie de session, et les mêmes idées s'appliquent ici.

Utiliser les propriétés MQTT 5 avec aiomqtt

Comme le client ci-dessus se connecte avec ProtocolVersion.V5, vous pouvez attacher des propriétés MQTT 5 à des messages individuels. aiomqtt réutilise la classe Properties de paho, le code est donc le même que dans un client à callbacks :

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

props = Properties(PacketTypes.PUBLISH)
props.MessageExpiryInterval = 60          # abandonner si non livré sous 60 s
props.ContentType = "application/json"
await client.publish(f"{PREFIX}/sensors/room1", payload, qos=1, properties=props)

L'expiration des messages est utile pour les commandes : un « allumer » qui arrive dix minutes trop tard est souvent pire qu'un ordre jamais reçu. Côté réception, message.properties expose les mêmes attributs. Le guide des fonctionnalités MQTT 5 explique quelles propriétés valent la peine d'être utilisées.

Problèmes courants

1. NotImplementedError sous Windows : vous êtes sur la boucle Proactor. Utilisez la boucle selector comme indiqué.
2. MqttError: Could not connect : le port 1883 est bloqué ou le nom d'hôte est erroné. Essayez TLS sur le port 8883 avec un tls_context.
3. La connexion tombe toutes les quelques secondes : un second client utilise le même identifiant. Générez-en un unique par processus.
4. Les messages arrivent en retard ou par rafales : quelque chose de lent s'exécute dans async for. Déplacez-le dans les workers.
5. Limitation de débit sur le serveur MQTT public gratuit : il autorise 10 messages par seconde par client et 10 abonnements par client. Gardez un intervalle de publication raisonnable.
6. Des appels bloquants figent tout : un requests.get() ou un time.sleep() synchrone dans une coroutine bloque toute la boucle. Utilisez des bibliothèques asynchrones ou asyncio.to_thread().

Passer en production

Sur un broker ConnectMQ privé, vous passez le username, le password générés et le client ID comme identifier, ainsi que tls_context=ssl.create_default_context() sur le port 8883. Le guide Python MQTT TLS explique la vérification des certificats, et le tutoriel Python MQTT JSON montre comment versionner des payloads comme les mesures ci-dessus. Pour stocker ce que traitent les workers, adaptez le logger Python MQTT vers SQLite, et pour une comparaison avec des clients dans d'autres langages, consultez le panorama des bibliothèques clientes MQTT. Le projet aiomqtt sur GitHub documente chaque option du client.

Exécutez vos services Python asynchrones sur ConnectMQ

ConnectMQ est un broker MQTT managé qui prend en charge MQTT 3.1.1 et 5.0 sur TCP, TLS et WebSocket, avec l'expiration de session, les retained messages et le Last Will sur lesquels s'appuie ce client. Développez sur broker.connectmq.com, puis déployez sur un broker MQTT cloud privé sans modifier votre code au-delà des paramètres de connexion. Inscrivez-vous gratuitement sur le plan Serverless, sans carte bancaire.

Questions fréquentes

aiomqtt est-il plus rapide que paho-mqtt ?

Il utilise paho-mqtt en interne, donc le débit brut du protocole est similaire. L'avantage tient à la structure : le code asynchrone permet à un seul thread de gérer de nombreuses opérations concurrentes sans callbacks ni verrous.

Pourquoi aiomqtt échoue-t-il sous Windows ?

La boucle d'événements Proactor par défaut de Windows ne dispose pas des méthodes add_reader et add_writer dont aiomqtt a besoin. Exécutez votre programme sur asyncio.SelectorEventLoop et tout fonctionne.

aiomqtt se reconnecte-t-il automatiquement ?

Non. Une connexion perdue lève aiomqtt.MqttError, et vous vous reconnectez en réentrant dans async with client au sein d'une boucle, ce qui vous permet aussi de contrôler le backoff et le réabonnement.

Puis-je utiliser aiomqtt avec FastAPI ?

Oui. Démarrez le client dans le gestionnaire lifespan de l'application et exécutez le consumer comme tâche d'arrière-plan sur la même boucle d'événements.