Mettre en tampon les messages MQTT hors ligne en Python avec SQLite (store and forward)

Pour mettre en tampon les messages MQTT hors ligne en Python, écrivez d'abord chaque mesure dans une file SQLite locale et laissez un émetteur distinct vider cette file vers le broker uniquement lorsque la connexion est active. Dans ce tutoriel, vous allez construire ce publisher store-and-forward avec paho-mqtt 2.x : il résiste aux coupures réseau et aux redémarrages, rejoue les messages dans leur ordre d'origine, ne les supprime qu'après acquittement par le broker et limite l'occupation du disque. Il fonctionne sur un Raspberry Pi ou n'importe quelle machine Linux.

Pourquoi la file intégrée de paho-mqtt ne suffit pas

paho-mqtt met déjà en file les messages sortants pendant la reconnexion : pourquoi construire autre chose ? Parce que cette file vit en mémoire. Si le processus redémarre, si le Pi reboote ou si le courant est coupé pendant une panne réseau, tout ce qu'elle contient est perdu. Pour une passerelle en cellulaire, un site au Wi-Fi peu fiable ou un véhicule qui traverse des tunnels, ce sont précisément les moments qui comptent.

Il existe aussi un problème plus subtil dans la façon dont le broker traite les publishers. Les sessions persistantes et la QoS 1 ou 2 protègent les messages destinés aux abonnés temporairement hors ligne. Elles ne font rien pour un publisher qui ne peut pas joindre le broker du tout. La durabilité côté publication relève du client.

La conception qui résout ces deux problèmes est un classique, l'outbox :

1. L'application ne publie jamais directement. Elle insère une ligne dans une table SQLite : topic, payload, QoS, indicateur retain et horodatage.
2. Un thread émetteur lit les lignes les plus anciennes, les publie en QoS 1 et attend le PUBACK du broker.
3. Seules les lignes acquittées sont supprimées. Tout le reste attend la tentative suivante.

Comme SQLite est un fichier, la file survit aux redémarrages. Comme l'émetteur traite les lignes dans l'ordre d'insertion sur une seule connexion, l'ordre est préservé.

Concevoir la table outbox

outbox
  id        INTEGER PRIMARY KEY AUTOINCREMENT   -- defines replay order
  topic     TEXT NOT NULL
  payload   BLOB NOT NULL
  qos       INTEGER NOT NULL DEFAULT 1
  retain    INTEGER NOT NULL DEFAULT 0
  created   REAL NOT NULL                      -- unix time of the reading

Deux réglages SQLite font une grande différence sur un Pi. journal_mode=WAL permet au producteur d'insérer pendant que l'émetteur lit, et réduit le nombre d'écritures par transaction. synchronous=NORMAL offre un équilibre raisonnable entre sécurité et usure de la carte SD en mode WAL. Placez la base sur un SSD ou une partition inscriptible si votre système de fichiers racine est en lecture seule.

Le publisher store-and-forward

Le module ci-dessous est autonome. Importez OfflineBufferedPublisher, appelez publish() depuis n'importe quel endroit de votre application et il s'occupe du reste.

import json
import os
import sqlite3
import threading
import time
import uuid

import paho.mqtt.client as mqtt


class OfflineBufferedPublisher:
    def __init__(self, host, port=1883, client_id=None, username=None, password=None,
                 db_path="outbox.db", max_rows=200_000, batch=20, max_rate=8.0):
        self.db_path = db_path
        self.max_rows = max_rows
        self.batch = batch
        self.min_interval = 1.0 / max_rate if max_rate else 0
        self._lock = threading.Lock()
        self._wake = threading.Event()
        self._stop = threading.Event()

        self.db = sqlite3.connect(db_path, check_same_thread=False)
        self.db.execute("PRAGMA journal_mode=WAL")
        self.db.execute("PRAGMA synchronous=NORMAL")
        self.db.execute("""CREATE TABLE IF NOT EXISTS outbox (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            topic TEXT NOT NULL, payload BLOB NOT NULL,
            qos INTEGER NOT NULL DEFAULT 1, retain INTEGER NOT NULL DEFAULT 0,
            created REAL NOT NULL)""")
        self.db.commit()

        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                                  client_id=client_id or f"buffered-{uuid.uuid4().hex[:8]}")
        if username:
            self.client.username_pw_set(username, password)
        if port == 8883:
            self.client.tls_set()
        self.client.reconnect_delay_set(1, 60)
        self.client.on_connect = self._on_connect
        self.client.on_disconnect = self._on_disconnect
        self.client.connect_async(host, port, keepalive=30)
        self.client.loop_start()

        self.sender = threading.Thread(target=self._drain, daemon=True)
        self.sender.start()

    # ---- côté producteur -----------------------------------------------
    def publish(self, topic, payload, qos=1, retain=False):
        if not isinstance(payload, (bytes, bytearray)):
            payload = json.dumps(payload).encode() if not isinstance(payload, str) else payload.encode()
        with self._lock:
            self.db.execute("INSERT INTO outbox (topic, payload, qos, retain, created) VALUES (?,?,?,?,?)",
                            (topic, payload, qos, int(retain), time.time()))
            self._enforce_limit()
            self.db.commit()
        self._wake.set()

    def pending(self):
        with self._lock:
            return self.db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]

    def _enforce_limit(self):
        count = self.db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]
        if count > self.max_rows:
            self.db.execute("DELETE FROM outbox WHERE id IN (SELECT id FROM outbox ORDER BY id LIMIT ?)",
                            (count - self.max_rows,))

    # ---- côté émetteur -------------------------------------------------
    def _on_connect(self, client, userdata, flags, reason_code, properties):
        print("connected:", reason_code)
        self._wake.set()

    def _on_disconnect(self, client, userdata, flags, reason_code, properties):
        print("disconnected:", reason_code)

    def _drain(self):
        while not self._stop.is_set():
            if not self.client.is_connected():
                self._wake.wait(2)
                self._wake.clear()
                continue
            with self._lock:
                rows = self.db.execute(
                    "SELECT id, topic, payload, qos, retain FROM outbox ORDER BY id LIMIT ?",
                    (self.batch,)).fetchall()
            if not rows:
                self._wake.wait(5)
                self._wake.clear()
                continue
            for row_id, topic, payload, qos, retain in rows:
                info = self.client.publish(topic, payload, qos=qos, retain=bool(retain))
                try:
                    info.wait_for_publish(timeout=10)
                except (ValueError, RuntimeError):
                    break
                if not info.is_published():
                    break  # connexion perdue ; on garde cette ligne et toutes les suivantes
                with self._lock:
                    self.db.execute("DELETE FROM outbox WHERE id = ?", (row_id,))
                    self.db.commit()
                if self.min_interval:
                    time.sleep(self.min_interval)

    def close(self):
        self._stop.set()
        self._wake.set()
        self.sender.join(timeout=5)
        self.client.loop_stop()
        self.client.disconnect()
        self.db.close()


if __name__ == "__main__":
    pub = OfflineBufferedPublisher(
        host=os.getenv("MQTT_HOST", "broker.connectmq.com"),
        port=int(os.getenv("MQTT_PORT", "1883")),
        client_id=os.getenv("MQTT_CLIENT_ID"),
        username=os.getenv("MQTT_USERNAME"),
        password=os.getenv("MQTT_PASSWORD"),
    )
    prefix = os.getenv("TOPIC_PREFIX", "connectmq/demo/q5w8e1/sensor1")
    n = 0
    try:
        while True:
            n += 1
            pub.publish(f"{prefix}/reading", {"seq": n, "ts": round(time.time(), 3), "value": 20 + n % 5})
            print("queued", n, "pending", pub.pending())
            time.sleep(2)
    except KeyboardInterrupt:
        pub.close()

Tester une coupure

Lancez un abonné qui affiche les numéros de séquence, puis exécutez le publisher :

mosquitto_sub -h broker.connectmq.com -p 1883 -i buffer-viewer-q5w8e1 -t "connectmq/demo/q5w8e1/#" -v
python buffered_publisher.py

Simulez maintenant une coupure. Sur un Pi, vous pouvez couper le réseau avec sudo ip link set wlan0 down (depuis une console locale, pas via SSH sur la même interface), attendre une minute, puis le rétablir avec sudo ip link set wlan0 up. Le compteur « pending » augmente pendant la coupure. Lorsque paho se reconnecte, l'émetteur vide l'arriéré et l'abonné reçoit dans l'ordre chaque numéro de séquence manquant, suivi des mesures en direct. Essayez d'arrêter le script pendant la coupure puis de le relancer : l'arriéré est toujours là.

Ordre, doublons et horodatages

Ordre. L'émetteur publie un message à la fois et attend son acquittement avant de le supprimer et de passer au suivant. Sur une seule connexion MQTT, cela préserve l'ordre d'origine. Vous pouvez augmenter le débit en autorisant plusieurs messages en vol, mais gardez un seul thread émetteur et une seule connexion si l'ordre compte.

Doublons. La QoS 1 garantit une livraison au moins une fois. Si la connexion tombe après que le broker a reçu un message mais avant l'arrivée du PUBACK, la ligne est renvoyée après la reconnexion. Incluez un numéro de séquence ou un identifiant unique dans le payload pour que les consommateurs puissent écarter les doublons. Les niveaux de QoS MQTT expliquent les compromis, et pourquoi la QoS 2 ne vous protège toujours pas d'un redémarrage du client.

Horodatages. Un message rejoué arrive avec des minutes ou des heures de retard. Mettez toujours l'heure de mesure dans le payload et faites en sorte que les consommateurs l'utilisent plutôt que l'heure d'arrivée. Avec MQTT 5, vous pouvez aussi définir un intervalle d'expiration sur les mesures qui perdent leur sens au bout d'un moment, comme une position en direct.

Réglages pour les déploiements réels

Lorsque la file est pleine, cette implémentation supprime les lignes les plus anciennes, ce qui convient à la télémétrie où les données récentes comptent le plus. Pour des journaux d'événements, vous préférerez peut-être refuser les nouvelles lignes et lever une alarme. Dans les deux cas, publiez le compteur pending() comme indicateur de santé, afin qu'un arriéré croissant soit visible sur votre tableau de bord.

Exécutez le publisher comme service systemd pour qu'il redémarre avec l'appareil, comme indiqué dans lancer un script Python MQTT au démarrage. Pour d'autres façons de stocker localement les données MQTT, consultez MQTT vers SQLite en Python, qui couvre le côté abonné.

Alternative : un broker local avec un bridge

Une autre conception store-and-forward courante place un broker Mosquitto sur le Pi lui-même. Les applications locales publient sur localhost, et Mosquitto relaie les topics vers le broker cloud via un bridge, en mettant les messages en file pendant que la liaison montante est coupée. C'est un bon choix lorsque plusieurs processus du même appareil publient des données, ou lorsque vous ne pouvez pas modifier le code de l'application. Le guide de configuration d'un bridge Mosquitto détaille les réglages.

Les compromis méritent d'être connus. Vous exploitez désormais un broker sur chaque passerelle, avec sa propre configuration, ses mises à jour et ses identifiants. Les limites de file et le comportement de persistance de Mosquitto demandent un réglage soigneux pour survivre aux redémarrages, et diagnostiquer des trous dans les données implique d'examiner deux brokers au lieu d'un. L'outbox Python garde tout dans un seul processus que vous maîtrisez entièrement et que vous pouvez tester avec un test unitaire, ce que beaucoup d'équipes trouvent plus simple pour une passerelle à usage unique. Choisissez le bridge lorsque de nombreux producteurs locaux partagent une liaison montante, et l'outbox lorsqu'une seule application possède les données.

Réglages du broker de production

Le serveur MQTT public gratuit convient pour ce test, mais tout ce qui y est publié est public. Pour des données réelles, créez un broker privé sur ConnectMQ, puis lancez le publisher avec MQTT_PORT=8883, MQTT_USERNAME=YOUR_USERNAME, MQTT_PASSWORD=YOUR_PASSWORD et MQTT_CLIENT_ID=YOUR_CLIENT_ID récupérés dans le portail. Un broker MQTT managé assure la disponibilité côté serveur, tandis que cette outbox comble les trous côté client. La documentation paho-mqtt décrit en détail MQTTMessageInfo et les options de reconnexion.

Ce pattern est une brique de base de chaque passerelle de la série tutoriel MQTT sur Raspberry Pi, des lecteurs de bus CAN aux récepteurs LoRa.

Ne perdez plus aucune mesure avec ConnectMQ

Combinez une outbox locale durable avec un broker fiable et les coupures deviennent invisibles pour vos tableaux de bord. Créez gratuitement un broker Serverless sur app.connectmq.com, sans carte bancaire, pointez ce publisher vers lui en TLS et regardez l'arriéré se vider dans l'explorateur de topics en direct de ConnectMQ après votre prochaine coupure réseau.

Questions fréquentes

Une session persistante MQTT met-elle en tampon les messages d'un publisher ?

Non. Les sessions persistantes conservent les messages destinés aux abonnés hors ligne. Un publisher qui ne peut pas joindre le broker doit mettre les messages en tampon localement, ce que fait cette outbox.

Les messages rejoués arriveront-ils dans le bon ordre ?

Oui, tant qu'un seul émetteur publie les lignes dans l'ordre des ID sur une seule connexion et attend chaque acquittement, comme dans ce code.

Puis-je l'utiliser avec asyncio ?

Oui. La même conception d'outbox fonctionne avec aiomqtt : exécutez la boucle de vidage comme tâche asynchrone et utilisez aiosqlite ou exécutez les appels SQLite dans un exécuteur de threads.

Comment éviter d'user la carte SD ?

Utilisez le mode WAL, gardez des payloads compacts et placez la base sur un SSD lorsque c'est possible. Écrire une mesure toutes les quelques secondes représente une charge modeste pour un stockage de qualité.