Um MQTT-Nachrichten in Python offline zu puffern, schreiben Sie jeden Messwert zuerst in eine lokale SQLite-Warteschlange und lassen einen separaten Sender diese Warteschlange nur dann zum Broker leeren, wenn die Verbindung steht. In diesem Tutorial bauen Sie einen solchen Store-and-Forward-Publisher mit paho-mqtt 2.x: Er übersteht Netzausfälle und Neustarts, sendet Nachrichten in der ursprünglichen Reihenfolge nach, löscht sie erst, wenn der Broker sie bestätigt hat, und hält den Speicherbedarf begrenzt. Er läuft auf einem Raspberry Pi oder jedem anderen Linux-Rechner.
Warum die eingebaute Warteschlange von paho-mqtt nicht ausreicht
paho-mqtt puffert ausgehende Nachrichten bereits, während es sich neu verbindet. Warum also etwas bauen? Weil diese Warteschlange im Arbeitsspeicher liegt. Startet der Prozess neu, bootet der Pi neu oder fällt während eines Ausfalls der Strom weg, ist alles darin verloren. Für ein Gateway am Mobilfunk, einen Standort mit unzuverlässigem WLAN oder ein Fahrzeug, das durch Tunnel fährt, sind genau das die entscheidenden Momente.
Hinzu kommt ein subtiles Problem damit, wie der Broker Publisher behandelt. Persistente Sessions und QoS 1 oder 2 schützen Nachrichten an Subscriber, die vorübergehend offline sind. Für einen Publisher, der den Broker überhaupt nicht erreicht, tun sie nichts. Dauerhaftigkeit auf der Publisher-Seite ist Aufgabe des Clients.
Das Design, das beide Probleme löst, ist eine klassische Outbox:
- Die Anwendung veröffentlicht nie direkt. Sie fügt eine Zeile in eine SQLite-Tabelle ein: Topic, Payload, QoS, Retain-Flag und Zeitstempel.
- Ein Sender-Thread liest die ältesten Zeilen, veröffentlicht sie mit QoS 1 und wartet auf das PUBACK des Brokers.
- Nur bestätigte Zeilen werden gelöscht. Alles andere bleibt für den nächsten Versuch erhalten.
Da SQLite eine Datei ist, übersteht die Warteschlange Neustarts. Da der Sender die Zeilen in Einfügereihenfolge über eine einzige Verbindung verarbeitet, bleibt die Reihenfolge erhalten.
Die Outbox-Tabelle entwerfen
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
Zwei SQLite-Einstellungen machen auf einem Pi einen großen Unterschied. journal_mode=WAL erlaubt dem Producer das Einfügen, während der Sender liest, und reduziert die Zahl der Schreibvorgänge pro Transaktion. synchronous=NORMAL ist im WAL-Modus ein vernünftiger Kompromiss zwischen Sicherheit und SD-Karten-Verschleiß. Legen Sie die Datenbank auf eine SSD oder eine beschreibbare Partition, wenn Sie ein schreibgeschütztes Root-Dateisystem betreiben.
Der Store-and-Forward-Publisher
Das folgende Modul ist in sich abgeschlossen. Importieren Sie OfflineBufferedPublisher, rufen Sie publish() an beliebiger Stelle in Ihrer Anwendung auf, und das Modul erledigt den Rest.
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()
# ---- Producer-Seite ------------------------------------------------
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,))
# ---- Sender-Seite --------------------------------------------------
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 # Verbindung abgebrochen; diese Zeile und alle folgenden behalten
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()
Einen Ausfall testen
Starten Sie einen Subscriber, der die Sequenznummern ausgibt, und führen Sie dann den Publisher aus:
mosquitto_sub -h broker.connectmq.com -p 1883 -i buffer-viewer-q5w8e1 -t "connectmq/demo/q5w8e1/#" -v
python buffered_publisher.py
Simulieren Sie nun einen Ausfall. Auf einem Pi können Sie das Netzwerk mit sudo ip link set wlan0 down trennen (tun Sie das an einer lokalen Konsole, nicht per SSH über dieselbe Schnittstelle), eine Minute warten und es mit sudo ip link set wlan0 up wieder aktivieren. Der Zähler „pending“ wächst während des Ausfalls. Sobald paho sich neu verbindet, arbeitet der Sender den Rückstau ab, und der Subscriber erhält jede fehlende Sequenznummer in der richtigen Reihenfolge, gefolgt von den Live-Messwerten. Beenden Sie das Skript testweise während des Ausfalls und starten Sie es erneut: Der Rückstau ist noch da.
Reihenfolge, Duplikate und Zeitstempel
Reihenfolge. Der Sender veröffentlicht jeweils eine Nachricht und wartet auf ihre Bestätigung, bevor er sie löscht und weitermacht. Über eine einzelne MQTT-Verbindung bleibt so die ursprüngliche Reihenfolge erhalten. Den Durchsatz können Sie erhöhen, indem Sie mehrere unbestätigte Nachrichten gleichzeitig zulassen, doch behalten Sie einen einzigen Sender-Thread und eine einzige Verbindung bei, wenn die Reihenfolge wichtig ist.
Duplikate. QoS 1 bedeutet „mindestens einmal“. Bricht die Verbindung ab, nachdem der Broker eine Nachricht empfangen hat, aber bevor das PUBACK ankam, wird die Zeile nach der Wiederverbindung erneut gesendet. Nehmen Sie eine Sequenznummer oder eine eindeutige ID in die Payload auf, damit Abnehmer Duplikate verwerfen können. MQTT-QoS-Stufen erklärt die Kompromisse und warum selbst QoS 2 Sie nicht vor einem Neustart des Clients schützt.
Zeitstempel. Eine nachgesendete Nachricht kommt Minuten oder Stunden zu spät an. Schreiben Sie immer den Messzeitpunkt in die Payload und lassen Sie die Abnehmer diesen statt der Ankunftszeit verwenden. Mit MQTT 5 können Sie für Messwerte, die nach einer Weile bedeutungslos werden, etwa eine Live-Position, zusätzlich ein Message Expiry Interval setzen.
Feintuning für reale Einsätze
| Einstellung | Standard hier | Wann ändern |
|---|---|---|
max_rows |
200.000 | Anhand des Speicherplatzes und der erwarteten Ausfalldauer dimensionieren |
batch |
20 | Größere Batches reduzieren SQLite-Lesezugriffe bei großem Rückstau |
max_rate |
8 Nachr./s | Hält das Nachsenden unter dem Limit des öffentlichen Brokers von 10 Nachrichten/s pro Client; in einem privaten Plan erhöhen |
| QoS | 1 | 0 nur für Daten verwenden, deren Verlust Sie verkraften können |
| DB-Speicherort | Arbeitsverzeichnis | Bei schreibgeschützten Systemen auf SSD oder eine beschreibbare Partition verlegen |
Ist die Warteschlange voll, verwirft diese Implementierung die ältesten Zeilen, was zu Telemetrie passt, bei der die jüngsten Daten am wichtigsten sind. Für Ereignisprotokolle bevorzugen Sie möglicherweise, keine neuen Zeilen mehr anzunehmen und stattdessen einen Alarm auszulösen. In beiden Fällen sollten Sie den Wert von pending() als Gesundheitsmetrik veröffentlichen, damit ein wachsender Rückstau auf Ihrem Dashboard sichtbar wird.
Betreiben Sie den Publisher als systemd-Dienst, damit er mit dem Gerät neu startet, wie in Python-MQTT-Skript beim Booten ausführen gezeigt. Weitere Möglichkeiten, MQTT-Daten lokal zu speichern, finden Sie in MQTT-Daten in SQLite mit Python, das die Subscriber-Seite behandelt.
Alternative: lokaler Broker mit Bridge
Ein weiteres verbreitetes Store-and-Forward-Design setzt einen Mosquitto-Broker direkt auf den Pi. Lokale Anwendungen veröffentlichen an localhost, und Mosquitto leitet die Topics per Bridge an den Cloud-Broker weiter und puffert Nachrichten, solange die Uplink-Verbindung ausgefallen ist. Das passt gut, wenn mehrere Prozesse auf demselben Gerät Daten veröffentlichen oder wenn Sie den Anwendungscode nicht ändern können. Die Anleitung zur Mosquitto-Bridge-Konfiguration beschreibt die Einstellungen.
Die Kompromisse sollte man kennen. Sie betreiben dann auf jedem Gateway einen Broker mit eigener Konfiguration, eigenen Updates und eigenen Zugangsdaten. Die Warteschlangenlimits und das Persistenzverhalten von Mosquitto müssen sorgfältig abgestimmt werden, um Neustarts zu überstehen, und die Fehlersuche bei Datenlücken bedeutet, zwei Broker statt einem zu untersuchen. Die Python-Outbox hält alles in einem einzigen Prozess, den Sie vollständig kontrollieren und mit einem Unit-Test prüfen können, was viele Teams für ein Gateway mit einer einzigen Aufgabe einfacher finden. Wählen Sie die Bridge, wenn sich viele lokale Producer einen Uplink teilen, und die Outbox, wenn eine Anwendung die Daten besitzt.
Broker-Einstellungen für die Produktion
Der kostenlose öffentliche MQTT-Broker genügt für diesen Test, doch alles, was dort veröffentlicht wird, ist öffentlich. Für echte Daten erstellen Sie einen privaten Broker bei ConnectMQ und starten den Publisher dann mit MQTT_PORT=8883, MQTT_USERNAME=YOUR_USERNAME, MQTT_PASSWORD=YOUR_PASSWORD und MQTT_CLIENT_ID=YOUR_CLIENT_ID aus dem Portal. Ein verwalteter MQTT-Broker sorgt für die Verfügbarkeit auf der Serverseite, während diese Outbox die Lücken auf der Client-Seite schließt. Die paho-mqtt-Dokumentation beschreibt MQTTMessageInfo und die Reconnect-Optionen im Detail.
Dieses Muster ist ein zentraler Baustein für jedes Gateway in der Reihe Raspberry-Pi-MQTT-Tutorial, vom CAN-Bus-Leser bis zum LoRa-Empfänger.
Mit ConnectMQ keinen Messwert mehr verlieren
Kombinieren Sie eine dauerhafte lokale Outbox mit einem zuverlässigen Broker, und Ausfälle werden für Ihre Dashboards unsichtbar. Erstellen Sie unter app.connectmq.com kostenlos und ohne Kreditkarte einen Serverless-Broker, richten Sie diesen Publisher per TLS darauf aus und beobachten Sie nach dem nächsten Netzausfall, wie der Rückstau in den Live-Topic-Explorer von ConnectMQ abfließt.
Häufig gestellte Fragen
Puffert eine persistente MQTT-Session Nachrichten für einen Publisher?
Nein. Persistente Sessions halten Nachrichten für Subscriber vor, die offline sind. Ein Publisher, der den Broker nicht erreicht, muss lokal puffern, und genau das leistet diese Outbox.
Kommen nachgesendete Nachrichten in der richtigen Reihenfolge an?
Ja, solange ein einziger Sender die Zeilen in ID-Reihenfolge über eine Verbindung veröffentlicht und auf jede Bestätigung wartet, wie in diesem Code.
Kann ich das mit asyncio verwenden?
Ja. Dasselbe Outbox-Design funktioniert mit aiomqtt: Führen Sie die Drain-Schleife als asynchronen Task aus und verwenden Sie aiosqlite oder führen Sie SQLite-Aufrufe in einem Thread-Executor aus.
Wie vermeide ich, die SD-Karte zu verschleißen?
Verwenden Sie den WAL-Modus, halten Sie Payloads kompakt und legen Sie die Datenbank nach Möglichkeit auf eine SSD. Alle paar Sekunden einen Messwert zu schreiben, ist für guten Speicher eine moderate Last.
