Reconnexion paho MQTT : backoff robuste, réabonnement et mise en tampon hors ligne en Python

Le comportement de reconnexion de paho MQTT fait toute la différence entre un client qui survit à un réseau instable et un client qui cesse silencieusement de fonctionner à 3 heures du matin. Ce guide explique ce que paho-mqtt 2.x fait automatiquement, les trois failles qu'il laisse ouvertes, et vous fournit un client Python testé avec backoff exponentiel, réabonnement à la connexion, tampon hors ligne et sessions MQTT 5 persistantes.

Symptômes d'une logique de reconnexion défaillante

Les bugs de reconnexion font rarement planter un programme. Ils se manifestent par des pannes plus discrètes, faciles à mal diagnostiquer :

• Le client se reconnecte après une coupure Wi-Fi, journalise « connected », mais ne reçoit plus jamais aucun message.
• Un script lancé alors que le réseau était coupé se termine immédiatement avec une erreur de socket au lieu d'attendre.
• Les mesures prises pendant une courte panne disparaissent, laissant des trous dans les graphiques.
• Des centaines d'appareils se reconnectent exactement au même moment après un redémarrage du broker et saturent le réseau.
• Deux processus se déconnectent mutuellement toutes les quelques secondes.

Chacun de ces symptômes a une cause précise et une correction précise.

Ce que paho fait automatiquement

Lorsque vous exécutez la boucle réseau avec loop_start() ou loop_forever(), paho-mqtt se reconnecte de lui-même après la coupure d'une connexion établie. Le délai entre deux tentatives commence à min_delay et double après chaque échec jusqu'à atteindre max_delay, valeurs configurées avec reconnect_delay_set(). Les valeurs par défaut sont 1 et 120 secondes. Une connexion réussie réinitialise le délai.

paho conserve aussi les messages QoS 1 et QoS 2 en vol que le broker n'a pas encore acquittés et les retransmet après la reconnexion, à condition que la session soit reprise.

Cela couvre beaucoup de cas, mais pas tous. Les failles restantes relèvent de votre responsabilité.

Faille 1 : la première connexion échoue

client.connect() effectue la première connexion de manière synchrone et lève une exception, généralement ConnectionRefusedError, socket.gaierror ou TimeoutError, si le broker est injoignable. La reconnexion automatique ne démarre jamais, puisqu'il n'y a jamais eu de connexion. C'est pourquoi les scripts lancés au démarrage, avant que le réseau ne soit prêt, meurent souvent.

La correction consiste à utiliser connect_async() suivi de loop_start(). La tentative de connexion passe alors dans la boucle réseau, qui réessaie selon les mêmes règles de backoff.

Faille 2 : les abonnements disparaissent

Avec une session propre (clean session), le broker supprime vos abonnements à la fin de la connexion. paho ne les enregistre pas et ne se réabonne pas à votre place. Si vous appelez subscribe() une seule fois après connect(), le client se reconnectera sans broncher puis ne recevra plus rien.

La correction est simple et non négociable : abonnez-vous dans on_connect. Ce callback s'exécute après chaque connexion réussie, reconnexions comprises, et les abonnements sont donc toujours restaurés.

Faille 3 : les messages publiés hors ligne

Les mesures produites pendant que la liaison est coupée doivent bien aller quelque part. Les messages QoS 0 sont de type « fire-and-forget » : les publier pendant une déconnexion revient à les perdre. S'appuyer sur la file interne de paho pour les niveaux de QoS supérieurs n'est pas non plus une bonne idée, car elle est illimitée et invisible pour votre application. Un tampon hors ligne explicite et borné, vidé après la reconnexion, vous donne le contrôle sur la mémoire et sur ce qu'il faut abandonner lorsqu'une panne dure trop longtemps.

Un client de reconnexion paho MQTT robuste

La classe ci-dessous comble les trois failles et ajoute du jitter au backoff pour qu'une flotte d'appareils ne se reconnecte pas en même temps. Elle utilise MQTT 5 avec une expiration de session pour que le broker conserve les abonnements et les messages QoS 1 en attente pendant les courtes pannes.

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 de (topic, qos)
        self.session_expiry_s = session_expiry_s
        self.offline = deque(maxlen=buffer_size)  # le plus ancien est supprimé quand c'est plein
        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()
        # Backoff exponentiel ; un plancher aléatoire étale les reconnexions d'une flotte
        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)  # toujours se réabonner
        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)

    # --- publication -----------------------------------------------------
    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()

Pourquoi chaque élément est là

Le numéro de séquence de chaque payload facilite les tests : après une panne, les trous ou doublons dans seq montrent exactement ce qui s'est passé. Comme la QoS 1 garantit une livraison « au moins une fois », les consommateurs doivent tolérer un doublon occasionnel, par exemple en ignorant les numéros de séquence déjà vus. Notre article sur les niveaux de QoS MQTT explique pourquoi des doublons sont possibles.

Comment tester la logique de reconnexion

Ne faites pas confiance à un code de reconnexion que vous n'avez pas cassé volontairement. Lancez le script et un subscriber côte à côte :

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

Essayez ensuite chaque mode de panne tour à tour :

1. Coupure réseau. Désactivez le Wi-Fi ou débranchez le câble pendant 30 secondes. Le journal affiche « connection lost », puis des tentatives échouées avec des délais croissants, puis « connected » et « flushing N buffered messages ». Le subscriber voit les numéros de séquence en tampon arriver en rafale.
2. Port bloqué sous Linux. Simulez un broker injoignable alors que le reste du réseau fonctionne :

sudo iptables -A OUTPUT -p tcp --dport 1883 -j DROP
# attendre 60 secondes, puis supprimer la règle
sudo iptables -D OUTPUT -p tcp --dport 1883 -j DROP

Des paquets ignorés constituent un test plus sévère qu'une connexion refusée, car le client ne s'en aperçoit qu'à l'expiration du keep-alive, soit environ une fois et demie l'intervalle de keep-alive. 3. Démarrage hors ligne. Coupez le réseau avant de lancer le script. Avec connect_async(), il attend et se connecte au retour du réseau au lieu de se terminer. 4. Prise de contrôle de session. Lancez un second client avec le même client ID, par exemple mosquitto_sub -i resilient-<run-id> -t x. Le broker déconnecte le client d'origine, qui se reconnecte et déconnecte à son tour l'intrus. C'est le symptôme des « deux processus qui se battent » ; le provoquer volontairement permet de le reconnaître facilement par la suite. 5. Commandes pendant une panne. Pendant que le script est hors ligne, publiez une commande QoS 1 sur son topic cmd. Comme la session a été conservée, la commande est livrée au retour du client.

Les reason codes à gérer

MQTT 5 vous indique pourquoi une connexion s'est terminée. Journalisez le reason code et réagissez à ceux qu'une nouvelle tentative ne corrigera pas.

Réessayer indéfiniment après un refus 0x86 ne fait qu'inonder les journaux du broker, et de nombreux brokers, dont ConnectMQ, bloquent une adresse IP après des échecs de connexion répétés. Dans _on_connect, envisagez d'appeler client.disconnect() et d'alerter un humain lorsque le code indique un problème permanent.

Paramètres de production

Pour de vrais déploiements, connectez-vous en TLS sur le port 8883 avec des identifiants délivrés par votre broker. Avec un broker MQTT managé ConnectMQ, construisez le client ainsi :

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)],
)

Gardez quelques limites en tête. Sur le serveur MQTT public gratuit, les sessions expirent au bout de 30 minutes au maximum : l'expiration de 1 800 secondes utilisée ci-dessus est donc la valeur utile maximale. Le plan gratuit Serverless prend en charge la QoS jusqu'à 1, ce qu'utilise ce client. Les plans payants ajoutent la QoS 2 si vous avez besoin d'une livraison « exactement une fois ». Le broker publie aussi l'état des connexions sur $SYS/broker/connection/{clientId}/state, ce qui est pratique pour confirmer qu'un appareil est revenu après une panne.

Pour un contexte plus large, notre tutoriel paho-mqtt Python couvre les bases, le guide de l'API de callbacks paho-mqtt 2.0 explique les signatures de callbacks de la version 2 utilisées ici, et le comparatif des bibliothèques clientes MQTT vous aide si vous hésitez entre paho et des bibliothèques asynchrones. La documentation officielle paho-mqtt liste toutes les options liées à la reconnexion.

Démarrez avec ConnectMQ

Testez dès aujourd'hui votre logique de reconnexion sur le serveur MQTT public gratuit, puis passez à des identifiants privés lors du déploiement. ConnectMQ exploite un broker MQTT 5 entièrement managé avec TLS, expiration de session et reason codes explicites, pour que vos clients reçoivent un retour utile en cas de problème. Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire.

Questions fréquentes

paho-mqtt se reconnecte-t-il automatiquement ?

Oui, une fois qu'une connexion a été établie et que la boucle réseau tourne avec loop_start() ou loop_forever(). Il ne réessaie pas un premier appel connect() qui a échoué, et il ne se réabonne pas à votre place.

Pourquoi mon client paho ne reçoit-il plus de messages après la reconnexion ?

Les abonnements ont été perdus avec l'ancienne session. Placez vos appels subscribe() dans on_connect pour qu'ils s'exécutent après chaque connexion réussie.

Quel délai de reconnexion utiliser ?

Commencez entre 1 et 5 secondes et plafonnez entre 60 et 120 secondes. Rendre aléatoire le délai initial de chaque appareil évite que toute une flotte se reconnecte au même instant.

Dois-je utiliser clean_start=False ?

Utilisez-le avec un intervalle d'expiration de session lorsque vous voulez que le broker conserve les abonnements et mette en file les messages QoS 1 pendant les courtes pannes. Gardez un client ID stable, car la session y est liée.