Paho MQTT Reconnect: Robustes Backoff, erneutes Abonnieren und Offline-Pufferung in Python

Das Reconnect-Verhalten von Paho MQTT entscheidet darüber, ob ein Client ein wackliges Netzwerk übersteht oder um drei Uhr nachts stillschweigend aufhört zu arbeiten. Dieser Leitfaden erklärt, was paho-mqtt 2.x automatisch erledigt, welche drei Lücken es offen lässt, und liefert Ihnen einen getesteten Python-Client mit exponentiellem Backoff, erneutem Abonnieren beim Verbinden, einem Offline-Puffer und persistenten MQTT-5-Sessions.

Symptome fehlerhafter Reconnect-Logik

Reconnect-Fehler bringen ein Programm selten zum Absturz. Sie zeigen sich als leisere Ausfälle, die leicht falsch diagnostiziert werden:

• Der Client verbindet sich nach einem WLAN-Ausfall neu, protokolliert "connected", empfängt aber nie wieder eine Nachricht.
• Ein Skript, das bei unterbrochenem Netzwerk gestartet wurde, beendet sich sofort mit einem Socket-Fehler, statt zu warten.
• Messwerte aus einem kurzen Ausfall verschwinden und hinterlassen Lücken in Diagrammen, OEE-Auswertungen oder im Energiemonitoring.
• Hunderte Geräte verbinden sich nach einem Broker-Neustart exakt im selben Moment neu und überlasten das Netzwerk.
• Zwei Prozesse trennen sich alle paar Sekunden gegenseitig.

Jedes dieser Symptome hat eine konkrete Ursache und eine konkrete Lösung.

Was Paho automatisch erledigt

Wenn Sie die Netzwerkschleife mit loop_start() oder loop_forever() ausführen, verbindet sich paho-mqtt nach dem Abbruch einer bestehenden Verbindung selbstständig neu. Die Wartezeit zwischen den Versuchen beginnt bei min_delay und verdoppelt sich nach jedem Fehlschlag, bis sie max_delay erreicht, konfiguriert über reconnect_delay_set(). Die Standardwerte sind 1 und 120 Sekunden. Eine erfolgreiche Verbindung setzt die Wartezeit zurück.

Paho behält außerdem noch nicht vom Broker bestätigte QoS-1- und QoS-2-Nachrichten und überträgt sie nach dem Reconnect erneut, sofern die Session fortgesetzt wird.

Das deckt viel ab, aber nicht alles. Die verbleibenden Lücken liegen in Ihrer Verantwortung.

Lücke 1: Die erste Verbindung schlägt fehl

client.connect() führt die erste Verbindung synchron aus und löst eine Exception aus, typischerweise ConnectionRefusedError, socket.gaierror oder TimeoutError, wenn der Broker nicht erreichbar ist. Der automatische Reconnect startet nie, weil es nie eine Verbindung gab. Deshalb sterben Skripte, die beim Booten gestartet werden, bevor das Netzwerk bereit ist, häufig.

Die Lösung ist connect_async() gefolgt von loop_start(). Der Verbindungsversuch wandert in die Netzwerkschleife, die nach denselben Backoff-Regeln erneut versucht.

Lücke 2: Abonnements verschwinden

Bei einer Clean Session verwirft der Broker Ihre Abonnements, wenn die Verbindung endet. Paho merkt sie sich nicht und abonniert nicht für Sie erneut. Wenn Sie subscribe() einmalig nach connect() aufrufen, verbindet sich der Client fröhlich neu und empfängt dann nichts.

Die Lösung ist einfach und nicht verhandelbar: Abonnieren Sie in on_connect. Dieser Callback läuft nach jeder erfolgreichen Verbindung, einschließlich Reconnects, sodass Abonnements immer wiederhergestellt werden.

Lücke 3: Offline veröffentlichte Nachrichten

Messwerte, die entstehen, während die Verbindung unterbrochen ist, brauchen einen Ort. QoS-0-Nachrichten sind Fire-and-Forget, Veröffentlichen ohne Verbindung verwirft sie also. Sich bei höheren QoS-Stufen auf die interne Warteschlange von paho zu verlassen ist ebenfalls kein guter Plan, denn sie ist unbegrenzt und für Ihre Anwendung unsichtbar. Ein expliziter, begrenzter Offline-Puffer, den Sie nach dem Reconnect leeren, gibt Ihnen die Kontrolle über den Speicher und darüber, was verworfen wird, wenn ein Ausfall zu lange dauert.

Ein robuster Paho-MQTT-Reconnect-Client

Die folgende Klasse schließt alle drei Lücken und ergänzt das Backoff um Jitter, damit sich eine Geräteflotte nicht im Gleichschritt neu verbindet. Sie verwendet MQTT 5 mit Session Expiry, sodass der Broker Abonnements und eingereihte QoS-1-Nachrichten bei kurzen Ausfällen behält.

pip install "paho-mqtt>=2.0"

import json
import logging
import random
import threading
import time
import uuid
from collections import deque

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

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("mqtt")


class ResilientClient:
    def __init__(self, host, port, client_id, subscriptions,
                 username=None, password=None, use_tls=False,
                 session_expiry_s=1800, buffer_size=1000):
        self.host = host
        self.port = port
        self.subscriptions = subscriptions  # Liste von (topic, qos)
        self.session_expiry_s = session_expiry_s
        self.offline = deque(maxlen=buffer_size)  # Älteste werden bei vollem Puffer verworfen
        self.lock = threading.Lock()
        self.on_payload = None

        c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                        client_id=client_id, protocol=mqtt.MQTTv5)
        if username:
            c.username_pw_set(username, password)
        if use_tls:
            c.tls_set()
        # Exponentielles Backoff; eine zufällige Untergrenze verteilt sich neu verbindende Flotten
        c.reconnect_delay_set(min_delay=random.randint(1, 5), max_delay=60)
        c.on_connect = self._on_connect
        c.on_disconnect = self._on_disconnect
        c.on_connect_fail = self._on_connect_fail
        c.on_message = self._on_message
        self.client = c

    def start(self):
        props = Properties(PacketTypes.CONNECT)
        props.SessionExpiryInterval = self.session_expiry_s
        self.client.connect_async(self.host, self.port, keepalive=30,
                                  clean_start=False, properties=props)
        self.client.loop_start()

    def stop(self):
        self.client.disconnect()
        self.client.loop_stop()

    # --- Callbacks -------------------------------------------------------
    def _on_connect(self, client, userdata, flags, reason_code, properties):
        if reason_code.is_failure:
            log.warning("connect refused: %s", reason_code)
            return
        log.info("connected, session_present=%s", flags.session_present)
        if self.subscriptions:
            client.subscribe(self.subscriptions)  # immer erneut abonnieren
        self._flush()

    def _on_disconnect(self, client, userdata, flags, reason_code, properties):
        if reason_code == 0:
            log.info("clean disconnect")
        else:
            log.warning("connection lost: %s, reconnecting", reason_code)

    def _on_connect_fail(self, client, userdata):
        log.warning("connection attempt failed, retrying with backoff")

    def _on_message(self, client, userdata, msg):
        if self.on_payload:
            self.on_payload(msg.topic, msg.payload)

    # --- Veröffentlichen -------------------------------------------------
    def publish(self, topic, payload, qos=1, retain=False):
        if self.client.is_connected():
            info = self.client.publish(topic, payload, qos=qos, retain=retain)
            if info.rc == mqtt.MQTT_ERR_SUCCESS:
                return True
        with self.lock:
            self.offline.append((topic, payload, qos, retain))
        return False

    def _flush(self):
        with self.lock:
            pending = list(self.offline)
            self.offline.clear()
        if pending:
            log.info("flushing %d buffered messages", len(pending))
        for topic, payload, qos, retain in pending:
            self.client.publish(topic, payload, qos=qos, retain=retain)


if __name__ == "__main__":
    run_id = uuid.uuid4().hex[:8]
    prefix = f"connectmq/demo/{run_id}"
    rc = ResilientClient(
        host="broker.connectmq.com", port=1883,
        client_id=f"resilient-{run_id}",
        subscriptions=[(f"{prefix}/cmd/#", 1)],
    )
    rc.on_payload = lambda t, p: log.info("command on %s: %s", t, p)
    rc.start()
    print(f"publishing to {prefix}/telemetry, send commands to {prefix}/cmd/...")
    try:
        n = 0
        while True:
            n += 1
            payload = json.dumps({"seq": n, "ts": int(time.time() * 1000)})
            sent = rc.publish(f"{prefix}/telemetry", payload, qos=1)
            log.info("seq %d %s (buffered: %d)", n, "sent" if sent else "buffered", len(rc.offline))
            time.sleep(2)
    except KeyboardInterrupt:
        rc.stop()

Wozu jedes Teil da ist

Die Sequenznummer in jeder Payload erleichtert das Testen: Nach einem Ausfall zeigen Lücken oder Duplikate in seq genau, was passiert ist. Da QoS 1 eine At-least-once-Zustellung ist, sollten Consumer gelegentliche Duplikate tolerieren, etwa indem sie bereits gesehene Sequenznummern ignorieren. Unsere Erklärung der MQTT-QoS-Stufen zeigt, warum Duplikate möglich sind.

Reconnect-Logik testen

Vertrauen Sie keinem Reconnect-Code, den Sie nicht absichtlich kaputt gemacht haben. Lassen Sie das Skript und einen Subscriber nebeneinander laufen:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-<run-id> -t "connectmq/demo/<run-id>/#" -v

Probieren Sie dann jeden Fehlerfall nacheinander aus:

1. Netzwerkausfall. Deaktivieren Sie das WLAN oder ziehen Sie das Kabel für 30 Sekunden. Das Log zeigt "connection lost", dann fehlschlagende Versuche mit wachsenden Wartezeiten, dann "connected" und "flushing N buffered messages". Der Subscriber sieht die gepufferten Sequenznummern schubweise eintreffen.
2. Blockierter Port unter Linux. Simulieren Sie einen unerreichbaren Broker, während der Rest des Netzwerks funktioniert:

sudo iptables -A OUTPUT -p tcp --dport 1883 -j DROP
# 60 Sekunden warten, dann die Regel entfernen
sudo iptables -D OUTPUT -p tcp --dport 1883 -j DROP

Verworfene Pakete sind ein härterer Test als eine abgelehnte Verbindung, denn der Client bemerkt es nur über den Keep-Alive-Timeout, etwa das Anderthalbfache des Keep-Alive-Intervalls. 3. Offline starten. Trennen Sie das Netzwerk vor dem Start des Skripts. Mit connect_async() wartet es und verbindet sich, wenn das Netzwerk zurückkommt, statt sich zu beenden. 4. Session-Übernahme. Starten Sie einen zweiten Client mit derselben Client-ID, zum Beispiel mosquitto_sub -i resilient-<run-id> -t x. Der Broker trennt den ursprünglichen Client, der sich neu verbindet und seinerseits den Eindringling trennt. Das ist das Symptom "zwei Prozesse kämpfen"; es einmal absichtlich zu sehen, macht es später leicht erkennbar. 5. Befehle während eines Ausfalls. Während das Skript offline ist, veröffentlichen Sie einen QoS-1-Befehl auf seinem cmd-Topic. Da die Session erhalten blieb, wird der Befehl zugestellt, sobald der Client zurück ist.

Reason Codes, die Sie behandeln sollten

MQTT 5 teilt Ihnen mit, warum eine Verbindung endete. Protokollieren Sie den Reason Code und reagieren Sie auf diejenigen, die ein erneuter Versuch nicht behebt.

Eine 0x86-Ablehnung endlos zu wiederholen überflutet nur die Logs des Brokers, und viele Broker, auch ConnectMQ, sperren eine IP-Adresse nach wiederholt fehlgeschlagenen Anmeldungen. Erwägen Sie, in _on_connect client.disconnect() aufzurufen und einen Menschen zu alarmieren, wenn der Code auf ein dauerhaftes Problem hinweist.

Einstellungen für die Produktion

Verbinden Sie sich in echten Deployments über TLS auf Port 8883 mit den von Ihrem Broker ausgestellten Zugangsdaten. Für einen verwalteten MQTT-Broker von ConnectMQ erstellen Sie den Client so:

rc = ResilientClient(
    host="broker.connectmq.com", port=8883, use_tls=True,
    client_id="YOUR_CLIENT_ID", username="YOUR_USERNAME", password="YOUR_PASSWORD",
    subscriptions=[("connectmq/factory1/production/line1/+/cmd", 1)],
)

Das Abonnement folgt einer ISA-95-Hierarchie (Werk, Bereich, Linie, Maschine), wie sie in einem Unified Namespace (UNS) üblich ist. Gerade in solchen Architekturen ist sauberes erneutes Abonnieren wichtig, weil Steuerbefehle und Zustände aus der Fertigung nach einem Reconnect sofort wieder ankommen müssen.

Behalten Sie einige Grenzen im Blick. Auf dem öffentlichen Broker laufen Sessions nach höchstens 30 Minuten ab, die oben verwendeten 1.800 Sekunden Expiry sind dort also der maximal sinnvolle Wert. Der kostenlose Serverless-Plan unterstützt QoS bis 1, was dieser Client verwendet. Kostenpflichtige Pläne bieten zusätzlich QoS 2, falls Sie Exactly-once-Zustellung benötigen. Der Broker veröffentlicht außerdem den Verbindungsstatus unter $SYS/broker/connection/{clientId}/state, was praktisch ist, um zu bestätigen, dass ein Gerät nach einem Ausfall zurückgekehrt ist.

Für mehr Hintergrund behandelt unser paho-mqtt-Python-Tutorial die Grundlagen, der Leitfaden zur Callback-API von paho-mqtt 2.0 erklärt die hier verwendeten Callback-Signaturen der Version 2, und der Vergleich der MQTT-Client-Bibliotheken hilft, wenn Sie zwischen paho und asynchronen Bibliotheken wählen. Die offizielle paho-mqtt-Dokumentation listet jede Reconnect-bezogene Option auf.

Starten Sie mit ConnectMQ

Testen Sie Ihre Reconnect-Logik noch heute gegen den kostenlosen öffentlichen MQTT-Broker und wechseln Sie beim Deployment zu privaten Zugangsdaten. ConnectMQ ist ein verwalteter MQTT-Broker mit MQTT 5, TLS, Session Expiry und klaren Reason Codes, sodass Ihre Clients hilfreiches Feedback erhalten, wenn etwas schiefgeht. Erstellen Sie unter app.connectmq.com einen kostenlosen Serverless-Broker, ohne Kreditkarte.

Häufig gestellte Fragen

Verbindet sich paho-mqtt automatisch neu?

Ja, sobald eine Verbindung hergestellt wurde und die Netzwerkschleife mit loop_start() oder loop_forever() läuft. Einen fehlgeschlagenen ersten connect()-Aufruf wiederholt es nicht, und es abonniert nicht für Sie erneut.

Warum empfängt mein paho-Client nach dem Reconnect keine Nachrichten mehr?

Die Abonnements sind mit der alten Session verloren gegangen. Verschieben Sie Ihre subscribe()-Aufrufe in on_connect, damit sie nach jeder erfolgreichen Verbindung ausgeführt werden.

Welche Reconnect-Wartezeit sollte ich verwenden?

Beginnen Sie zwischen 1 und 5 Sekunden und begrenzen Sie auf 60 bis 120 Sekunden. Eine zufällige Startverzögerung pro Gerät verhindert, dass sich eine ganze Flotte im selben Moment neu verbindet.

Sollte ich clean_start=False verwenden?

Verwenden Sie es zusammen mit einem Session-Expiry-Intervall, wenn der Broker Abonnements behalten und QoS-1-Nachrichten bei kurzen Ausfällen einreihen soll. Halten Sie die Client-ID stabil, denn die Session ist an sie gebunden.