Asynchrones MQTT in Python mit aiomqtt und asyncio: Tasks, Worker und Reconnects

Mit aiomqtt und asyncio nutzen Sie MQTT in Python ohne Callbacks: Sie warten mit await auf ein Publish, iterieren mit async for über eingehende Nachrichten und führen beliebig viele parallele Tasks in einem einzigen Thread aus. In diesem Tutorial bauen Sie einen asynchronen Client, der mit broker.connectmq.com verbunden ist, Sensorwerte veröffentlicht, Nachrichten über eine Worker-Queue verarbeitet und sich sauber neu verbindet, wenn das Netzwerk ausfällt. Außerdem sehen Sie das Detail der Windows-Event-Loop, über das fast jeder stolpert.

Warum aiomqtt statt reinem paho-mqtt?

paho-mqtt ist eine Callback-Bibliothek. Ihre Netzwerkschleife läuft in einem eigenen Thread und ruft Ihre Funktionen auf, wenn etwas passiert. Für kleine Skripte ist das in Ordnung, wird aber umständlich, wenn Ihre Anwendung bereits auf asyncio basiert, etwa ein FastAPI-Service, ein Edge-Gateway in der Fertigungshalle, das zusätzlich asynchron HTTP, Modbus oder OPC UA spricht, oder ein Bot. Threads und eine Event-Loop zu mischen bedeutet run_coroutine_threadsafe, Locks und subtile Fehler beim Herunterfahren.

aiomqtt kapselt paho-mqtt und bietet stattdessen eine idiomatische asyncio-API:

Unter der Haube steckt weiterhin paho, das Protokollverhalten ist also identisch. Wenn Sie von Callbacks kommen, ist das paho-mqtt-Python-Tutorial eine nützliche Referenz für die Konzepte.

aiomqtt installieren

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

Die Beispiele verwenden asyncio.TaskGroup und except*, die Python 3.11 oder neuer voraussetzen. Wählen Sie für den öffentlichen MQTT-Broker, der nur für Tests und Prototypen gedacht ist, ein zufälliges Topic-Präfix, damit Ihre Tests von denen anderer Nutzer getrennt bleiben; hier verwenden wir connectmq/demo/k7f3q2.

Die Stolperfalle mit der Windows-Event-Loop

aiomqtt nutzt die Methoden add_reader und add_writer der Event-Loop. Unter Windows ist die Standard-Event-Loop die Proactor-Loop, die diese nicht implementiert, sodass Sie beim Verbinden sofort einen NotImplementedError erhalten. Führen Sie Ihr Programm stattdessen auf einer Selector-Event-Loop aus, wie es die folgenden Beispiele tun. Linux und macOS verwenden standardmäßig bereits eine Selector-Loop.

Ein minimales aiomqtt-asyncio-Beispiel

Beginnen Sie mit dem kleinsten funktionierenden Subscriber und Publisher in einer Datei:

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  # nach der ersten Nachricht beenden


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

Der async with-Block verbindet beim Eintritt und trennt beim Verlassen, selbst wenn eine Exception ausgelöst wird. client.messages ist ein asynchroner Iterator, der jede Nachricht aus jedem Abonnement in Ankunftsreihenfolge liefert.

Ein realistischer Client: parallele Tasks und eine Worker-Queue

Echte Anwendungen erledigen mehrere Dinge gleichzeitig: Messwerte nach Zeitplan veröffentlichen, auf Befehle reagieren und eingehende Daten verarbeiten, was langsame I/O wie einen Datenbank-Schreibvorgang oder einen HTTP-Aufruf beinhalten kann. Der Trick besteht darin, die Schleife, die Nachrichten liest, niemals zu blockieren. Speichern Sie dies als 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)  # schnell übergeben, weiterlesen


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)  # steht für einen langsamen Datenbank- oder HTTP-Aufruf
    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:  # eine fehlerhafte Nachricht darf den Worker nicht beenden
            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),
        # Privater ConnectMQ-Broker:
        # username="YOUR_USERNAME", password="YOUR_PASSWORD",
        # tls_context=ssl.create_default_context(), und 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")

Führen Sie python aio_client.py aus. Alle zwei Sekunden wird ein Messwert veröffentlicht, über das Abonnement wieder empfangen und von einem der drei Worker verarbeitet. Senden Sie einen Befehl aus einem anderen Terminal:

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

So funktioniert es

Strukturierte Nebenläufigkeit mit TaskGroup

Publisher und Consumer laufen als Geschwister-Tasks in einer TaskGroup. Löst einer von beiden eine Exception aus, etwa weil die Verbindung abgebrochen ist und publish() fehlschlägt, bricht die Gruppe den anderen Task ab und löst den Fehler erneut aus, verpackt in einer Exception-Gruppe. except* aiomqtt.MqttError fängt ihn ab, egal welcher Task fehlgeschlagen ist. Ohne Task-Gruppe kann ein Fehler in einem Task unbemerkt bleiben, während der andere auf einer toten Verbindung weiterläuft.

Die Nachrichtenschleife schnell halten

Der Consumer tut nichts anderes, als Nachrichten in eine asyncio.Queue zu verschieben. Die Worker übernehmen den langsamen Teil. Das ist wichtig, weil aiomqtt Nachrichten der Reihe nach über einen einzigen Iterator liefert; würden Sie innerhalb der async for-Schleife auf einen halbsekündigen Datenbank-Schreibvorgang warten, müsste jede andere Nachricht dahinter warten. Mit drei Workern werden drei Nachrichten parallel verarbeitet.

Das maxsize der Queue sorgt für Backpressure. Wenn die Worker nicht hinterherkommen, wartet queue.put(), was das Lesen vom Broker verlangsamt, statt den Speicher unbegrenzt wachsen zu lassen. Ist die Reihenfolge pro Topic wichtig, verteilen Sie Nachrichten per Topic-Hash an die Worker statt über eine gemeinsame Queue.

Saubere Reconnects

Ein aiomqtt-Client ist wiederverwendbar, aber nicht reentrant: Sie können async with client nach dem Verlassen erneut betreten, aber nicht zweimal gleichzeitig. Genau das macht die äußere while True-Schleife. Nach einem Fehler wartet sie einige Sekunden, verbindet sich neu, veröffentlicht den Online-Status erneut und abonniert erneut, da Abonnements zur Verbindung gehören. Der Last Will veröffentlicht "offline", wenn der Prozess ohne sauberes Trennen beendet wird.

Ersetzen Sie in der Produktion das feste Intervall durch exponentielles Backoff mit einer Obergrenze von ein bis zwei Minuten, damit eine Flotte von Clients den Broker nach einem Ausfall nicht im Gleichschritt bombardiert. Der Leitfaden zu paho-mqtt-Reconnects behandelt Backoff und Session-Strategie ausführlicher, und dieselben Ideen gelten auch hier.

MQTT-5-Properties mit aiomqtt nutzen

Da sich der obige Client mit ProtocolVersion.V5 verbindet, können Sie einzelnen Nachrichten MQTT-5-Properties mitgeben. aiomqtt verwendet die Properties-Klasse von paho, der Code ist also derselbe wie in einem Callback-basierten Client:

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

props = Properties(PacketTypes.PUBLISH)
props.MessageExpiryInterval = 60          # verwerfen, wenn nicht innerhalb von 60 s zugestellt
props.ContentType = "application/json"
await client.publish(f"{PREFIX}/sensors/room1", payload, qos=1, properties=props)

Message Expiry ist für Befehle nützlich: Ein "Einschalten", das zehn Minuten zu spät ankommt, ist oft schlimmer als eines, das nie ankommt. Auf der Empfängerseite stellt message.properties dieselben Attribute bereit. Der Leitfaden zu den MQTT-5-Funktionen erklärt, welche Properties sich lohnen.

Häufige Probleme

1. NotImplementedError unter Windows: Sie verwenden die Proactor-Loop. Nutzen Sie wie gezeigt die Selector-Loop.
2. MqttError: Could not connect: Port 1883 ist blockiert oder der Hostname ist falsch. Versuchen Sie TLS auf Port 8883 mit einem tls_context.
3. Verbindung bricht alle paar Sekunden ab: Ein zweiter Client verwendet denselben Identifier. Erzeugen Sie pro Prozess einen eindeutigen.
4. Nachrichten kommen verspätet oder schubweise an: Innerhalb von async for läuft etwas Langsames. Verlagern Sie es in Worker.
5. Ratenbegrenzung auf dem öffentlichen MQTT-Broker: Er erlaubt 10 Nachrichten pro Sekunde pro Client und 10 Abonnements pro Client. Halten Sie das Publish-Intervall vernünftig.
6. Blockierende Aufrufe frieren alles ein: Ein synchrones requests.get() oder time.sleep() in einer Coroutine blockiert die gesamte Loop. Verwenden Sie asynchrone Bibliotheken oder asyncio.to_thread().

Der Weg in die Produktion

In Industrie-4.0-Projekten ersetzen Sie das Testpräfix früh durch eine sprechende Hierarchie, etwa nach ISA-95 im Stil eines Unified Namespace (UNS) wie acme/werk-stuttgart/montage/linie-1/temperatur; wie das aussieht, zeigt der Beitrag zu Sparkplug B und Unified Namespace. Auf einem privaten ConnectMQ-Broker übergeben Sie den generierten username, das password und die Client-ID als identifier, dazu tls_context=ssl.create_default_context() auf Port 8883. Der Leitfaden zu Python MQTT TLS erklärt die Zertifikatsprüfung, und das Python-MQTT-JSON-Tutorial zeigt, wie Sie Payloads wie die obigen Messwerte versionieren. Um zu speichern, was die Worker verarbeiten, passen Sie den MQTT-zu-SQLite-Logger in Python an, und einen Vergleich mit Clients in anderen Sprachen bietet die Übersicht der MQTT-Client-Bibliotheken. Das aiomqtt-Projekt auf GitHub dokumentiert jede Client-Option.

Betreiben Sie Ihre asynchronen Python-Services auf ConnectMQ

ConnectMQ ist ein verwalteter MQTT-Broker, der MQTT 3.1.1 und 5.0 über TCP, TLS und WebSocket unterstützt, inklusive Session Expiry, Retained Messages und Last Will, auf die dieser Client angewiesen ist. Entwickeln Sie gegen broker.connectmq.com und wechseln Sie dann zu einem privaten Cloud-MQTT-Broker, ohne Ihren Code über die Verbindungseinstellungen hinaus zu ändern. Registrieren Sie sich kostenlos im Serverless-Plan, ohne Kreditkarte.

Häufig gestellte Fragen

Ist aiomqtt schneller als paho-mqtt?

aiomqtt nutzt intern paho-mqtt, der reine Protokolldurchsatz ist daher ähnlich. Der Vorteil liegt in der Struktur: Asynchroner Code ermöglicht es einem Thread, viele parallele Vorgänge ohne Callbacks oder Locks abzuwickeln.

Warum schlägt aiomqtt unter Windows fehl?

Der standardmäßigen Windows-Proactor-Event-Loop fehlen die Methoden add_reader und add_writer, die aiomqtt benötigt. Führen Sie Ihr Programm auf asyncio.SelectorEventLoop aus, dann funktioniert es.

Verbindet sich aiomqtt automatisch neu?

Nein. Eine abgebrochene Verbindung löst aiomqtt.MqttError aus, und Sie verbinden sich neu, indem Sie async with client in einer Schleife erneut betreten. So steuern Sie auch Backoff und erneutes Abonnieren.

Kann ich aiomqtt mit FastAPI verwenden?

Ja. Starten Sie den Client im Lifespan-Handler der Anwendung und führen Sie den Consumer als Hintergrund-Task auf derselben Event-Loop aus.