Store-and-Forward mit MQTT: Edge-Daten bei Ausfällen puffern und in richtiger Reihenfolge nachsenden

Store-and-Forward mit MQTT ist das Muster, das Industriedaten intakt hält, wenn das Netzwerk zwischen Edge und Cloud ausfällt. Statt Nachrichten zu verwerfen, solange der Broker nicht erreichbar ist, schreibt das Edge-Gerät sie in einen lokalen Speicher und sendet sie in der richtigen Reihenfolge nach, sobald die Verbindung zurück ist. Dieser Artikel erklärt, wie das Muster funktioniert, welche Designentscheidungen zählen, und liefert Ihnen eine lauffähige Python-Implementierung auf Basis von SQLite.

Warum MQTT allein nicht genügt

MQTT hat starke Zustellfunktionen: Bestätigungen bei QoS 1 und 2, persistente Sessions und Retained Messages. Sie alle greifen jedoch erst, wenn eine Nachricht das Gerät verlassen hat. Überlegen Sie, was während eines zweistündigen WAN-Ausfalls in einem Werk passiert:

• Das Gateway fragt weiter Maschinen ab und erzeugt Werte.
• Der MQTT-Client kann sich nicht verbinden, es gibt also keine Session, in die zugestellt werden könnte.
• Die meisten Client-Bibliotheken halten nicht gesendete Nachrichten mit Größenbegrenzung im Arbeitsspeicher und verlieren sie bei einem Neustart oder Stromausfall.

Persistente Sessions auf dem Broker helfen in der anderen Richtung: Sie halten Nachrichten für einen Subscriber bereit, der offline ist. Für einen Publisher, der den Broker nicht erreicht, tun sie nichts. Genau diese Lücke füllt Store-and-Forward, und deshalb gilt die Pufferung als Kernaufgabe jedes IIoT-Edge-Gateways.

So funktioniert Store-and-Forward mit MQTT

  Data source --> [ write to local queue (disk) ] --> forwarder --> MQTT broker
                         ^                               |
                         |      delete only after        |
                         +------- PUBACK received -------+

1. Jede Nachricht wird zuerst in eine dauerhafte lokale Queue geschrieben, mit einer Sequenznummer und dem Quellzeitstempel.
2. Ein Forwarder liest die ältesten Nachrichten und veröffentlicht sie mit QoS 1.
3. Erst wenn der Broker bestätigt (PUBACK), wird die Nachricht aus der Queue gelöscht.
4. Fällt die Verbindung aus, hält der Forwarder an, die Queue wächst, und die Datenquellen arbeiten unbeeinträchtigt weiter.
5. Nach dem Reconnect arbeitet der Forwarder den Rückstand mit kontrollierter Rate ab, älteste Nachricht zuerst, und fährt dann mit Live-Daten fort.

Alle Daten durch die Queue zu leiten, nicht nur die während eines Ausfalls erzeugten, hält die Reihenfolge einfach: Es gibt nur einen Pfad, und der ist immer First-in, First-out.

Designentscheidungen, die zählen

Den Puffer dimensionieren

Schätzen Sie ihn anhand von drei Zahlen: Nachrichten pro Sekunde im Normalbetrieb, durchschnittliche Nachrichtengröße einschließlich Overhead und der längste Ausfall, den Sie überstehen möchten. Beispiel: 20 Nachrichten pro Sekunde zu je 200 Byte über einen 24-stündigen Ausfall ergeben etwa 345 MB Payload plus Speicher-Overhead. Ist das zu viel, verringert eine Reduktion des Datenverkehrs an der Quelle mit Report by Exception den Bedarf oft drastisch.

Eine Verwerfungsregel wählen

Wenn der Puffer voll ist, müssen Sie wählen: die ältesten Daten verwerfen (aktuellen Zustand behalten) oder die neuesten (eine lückenlose Historie ab Beginn des Ausfalls behalten). Für Monitoring ist das Verwerfen der ältesten Daten meist richtig. Für Compliance oder Chargenprotokolle bevorzugen Sie womöglich, keine neuen Daten niedriger Priorität mehr anzunehmen und den Anfang zu bewahren. In beiden Fällen sollten Sie einen Zähler verworfener Nachrichten veröffentlichen, damit die Lücke sichtbar wird.

Wiedergabe und der Retained-Letztwert

Stundenlange Historie auf ein Topic nachzusenden, das Dashboards auch als „aktuellen Wert“ nutzen, kann eine Anzeige durch alte Werte springen lassen. Zwei gängige Lösungen: den Rückstand auf ein separates Topic wie .../history senden oder den Quellzeitstempel mitschicken und Konsumenten ältere Werte bei der Anzeige ignorieren lassen. Mit MQTT 5 können Sie außerdem für Werte, die veraltet wertlos sind, ein Message Expiry Interval setzen, sodass der Broker sie verwirft, statt sie verspätet zuzustellen.

Besonders wichtig ist das in einem Unified Namespace (UNS), in dem die ISA-95-Equipment-Hierarchie (Enterprise → Site → Area → Line → Cell) die Topic-Struktur auf dem MQTT-Broker bildet, etwa acme/werk-stuttgart/druckluft/linie-1/kompressor-1/telemetrie. Der UNS soll den aktuellen Zustand des Werks zeigen; nachgesendete Historie gehört deshalb in einen eigenen Zweig oder muss klar als historisch gekennzeichnet sein, damit MES, OEE-Dashboards und Energiemonitoring nicht mit alten Werten rechnen. Sparkplug B löst das mit dem Flag is_historical, beschrieben in Sparkplug B und Unified Namespace und der Referenz zu den Sparkplug B Datentypen.

Eine Store-and-Forward-Implementierung in Python

Das folgende Skript nutzt SQLite als dauerhafte Queue und paho-mqtt 2.x als Client. Ein Producer-Thread erzeugt jede Sekunde einen Messwert, unabhängig von der Konnektivität. Der Forwarder veröffentlicht die ältesten Zeilen mit QoS 1, wartet auf jede Bestätigung, löscht zugestellte Zeilen und drosselt die Wiedergabe, um unter dem Limit des öffentlichen MQTT-Brokers von 10 Nachrichten pro Sekunde zu bleiben. Der öffentliche MQTT-Broker ist für Tests und Prototypen gedacht; echte Produktionsdaten gehören auf einen privaten Broker.

import json
import sqlite3
import threading
import time
import random
from datetime import datetime, timezone

import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PORT = 1883
CLIENT_ID = "saf-gateway-2b7c"
TOPIC = "connectmq/demo/saf-2b7c/plant1/compressor1/telemetry"
DB_PATH = "saf_queue.db"
MAX_ROWS = 500_000          # Puffergrenze; älteste Zeilen werden darüber hinaus verworfen
REPLAY_RATE = 8             # Nachrichten pro Sekunde (öffentlicher Broker erlaubt 10)
ACK_TIMEOUT = 10            # Sekunden Wartezeit auf PUBACK

db_lock = threading.Lock()
db = sqlite3.connect(DB_PATH, check_same_thread=False)
db.execute("PRAGMA journal_mode=WAL")
db.execute("""CREATE TABLE IF NOT EXISTS queue (
                seq INTEGER PRIMARY KEY AUTOINCREMENT,
                topic TEXT NOT NULL,
                payload TEXT NOT NULL)""")
db.commit()


def enqueue(topic, data):
    with db_lock:
        db.execute("INSERT INTO queue (topic, payload) VALUES (?, ?)",
                   (topic, json.dumps(data)))
        count = db.execute("SELECT COUNT(*) FROM queue").fetchone()[0]
        if count > MAX_ROWS:
            db.execute("DELETE FROM queue WHERE seq IN "
                       "(SELECT seq FROM queue ORDER BY seq LIMIT ?)", (count - MAX_ROWS,))
        db.commit()


def oldest(n=50):
    with db_lock:
        return db.execute("SELECT seq, topic, payload FROM queue ORDER BY seq LIMIT ?",
                          (n,)).fetchall()


def delete(seq):
    with db_lock:
        db.execute("DELETE FROM queue WHERE seq = ?", (seq,))
        db.commit()


def producer():
    """Simulierte Datenquelle: blockiert nie auf dem Netzwerk."""
    while True:
        enqueue(TOPIC, {
            "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
            "pressure_bar": round(7 + random.gauss(0, 0.1), 3),
            "motor_temp_c": round(65 + random.gauss(0, 0.5), 2),
        })
        time.sleep(1)


def on_connect(client, userdata, flags, reason_code, properties):
    print("Connected:", reason_code)


def on_disconnect(client, userdata, flags, reason_code, properties):
    print("Disconnected:", reason_code, "- buffering locally")


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.reconnect_delay_set(min_delay=1, max_delay=60)
# Produktion: client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD"); client.tls_set(); PORT = 8883
client.connect_async(BROKER, PORT, keepalive=30)
client.loop_start()

threading.Thread(target=producer, daemon=True).start()

interval = 1.0 / REPLAY_RATE
while True:
    if not client.is_connected():
        time.sleep(1)
        continue
    rows = oldest()
    if not rows:
        time.sleep(0.5)
        continue
    for seq, topic, payload in rows:
        data = json.loads(payload)
        data["seq"] = seq                      # ermöglicht Konsumenten das Deduplizieren
        try:
            info = client.publish(topic, json.dumps(data), qos=1)
            info.wait_for_publish(timeout=ACK_TIMEOUT)
        except (RuntimeError, ValueError) as exc:
            print("Publish failed:", exc)
            break
        if not info.is_published():
            print("No PUBACK, will retry from seq", seq)
            break                              # Zeile behalten, später in Reihenfolge erneut versuchen
        delete(seq)
        time.sleep(interval)
    with db_lock:
        backlog = db.execute("SELECT COUNT(*) FROM queue").fetchone()[0]
    if backlog > 1:
        print("Backlog:", backlog)

Zum Testen starten Sie das Skript, beobachten das Topic in einem zweiten Terminal und trennen dann für ein, zwei Minuten Ihre Netzwerkverbindung:

mosquitto_sub -h broker.connectmq.com -p 1883 -i saf-2b7c-viewer -t "connectmq/demo/saf-2b7c/#" -v

Offline gibt das Skript die Trennung aus und puffert weiter. Kehrt das Netzwerk zurück, verbindet sich der Client automatisch neu, und der Rückstand wird in Sequenzreihenfolge abgearbeitet. Das Feld ts zeigt, wann jeder Wert gemessen wurde, und seq steigt lückenlos an. Sie können das Skript während des Ausfalls auch komplett beenden; nach dem Neustart ist die SQLite-Queue noch da.

Hinweise zur Implementierung

• Jeweils nur eine Nachricht unterwegs hält die Reihenfolge trivial und ist für die meisten Gateways schnell genug. Für höheren Durchsatz veröffentlichen Sie ein kleines Fenster und löschen Zeilen, sobald ihre Bestätigungen eintreffen.
• At least once, nicht exactly once. Geht das PUBACK verloren, nachdem der Broker die Nachricht angenommen hat, wird die Zeile erneut gesendet. Über das Feld seq kann ein Historian oder Subscriber Duplikate verwerfen. MQTT QoS 2 würde Duplikate auf der Leitung vermeiden, nicht aber über einen Gateway-Neustart hinweg, daher lohnen sich IDs auf Anwendungsebene trotzdem. Die Abwägungen finden Sie unter MQTT-QoS-Stufen erklärt.
• Der WAL-Modus in SQLite erlaubt gleichzeitige Lese- und Schreibvorgänge bei guter Dauerhaftigkeit.
• Gedrosselte Wiedergabe schützt den Broker und nachgelagerte Konsumenten. Gemeinsam genutzte Tarife haben Limits pro Client, und der kostenlose Serverless-Plan von ConnectMQ unterstützt QoS 1, also genau das, was dieses Muster braucht.

Checkliste für Store-and-Forward mit MQTT

1. Schreiben Sie jede Nachricht auf die Festplatte, bevor Sie versuchen, sie zu senden.
2. Fügen Sie jedem Payload einen Quellzeitstempel und eine Sequenznummer hinzu.
3. Löschen Sie aus dem Puffer erst, wenn der Broker bestätigt hat.
4. Senden Sie die ältesten Nachrichten zuerst nach, gedrosselt unter den Limits von Broker und Konsumenten.
5. Legen Sie eine maximale Puffergröße und eine Verwerfungsregel fest und melden Sie verworfene Nachrichten.
6. Entscheiden Sie, wie nachgesendete Historie mit „aktueller Wert“-Topics zusammenspielt.
7. Testen Sie mit echten Ausfällen: Kabel ziehen, Gateway mitten im Ausfall neu starten, Puffer volllaufen lassen.

Store-and-Forward bei Mobilfunk und abgelegenen Standorten

Abgelegene Anlagen mit Mobilfunkanbindung verlieren häufiger die Verbindung und zahlen pro Byte. Dort lässt sich Store-and-Forward gut mit Batching (mehrere Messwerte pro Nachricht) und Kompression kombinieren. Der Leitfaden MQTT über Mobilfunk behandelt die Abstimmung des Keep Alive und Datenbudgets für solche Verbindungen, und der Leitfaden zu Industrial IoT mit MQTT zeigt, wo die Pufferung in einer vollständigen Werksarchitektur ihren Platz hat.

Ein zuverlässiger Broker auf der Gegenseite

Store-and-Forward schützt die Edge-Seite der Verbindung. Auf der Cloud-Seite brauchen Sie einen Broker, der erreichbar, mit TLS abgesichert und für Wiedergabe-Schübe dimensioniert ist. Ein verwalteter MQTT-Broker nimmt Ihnen den Broker-Betrieb ab, und der Leitfaden zum Cloud-MQTT-Broker erklärt, wie Sie Optionen vergleichen.

Starten Sie mit ConnectMQ

Richten Sie Ihre puffernden Gateways auf ConnectMQ aus: TLS auf Port 8883, ausgestellte Zugangsdaten, Message Expiry aus MQTT 5 und ein Live-Topic-Explorer, in dem Sie verfolgen, wie ein Rückstand nach einem Ausfall abgearbeitet wird. Erstellen Sie unter app.connectmq.com einen kostenlosen Serverless-Broker, ganz ohne Kreditkarte.

Häufig gestellte Fragen

Hat MQTT ein eingebautes Store-and-Forward?

Nicht für Publisher, die den Broker nicht erreichen. QoS und persistente Sessions wirken, sobald eine Nachricht den Broker erreicht hat; die Pufferung davor ist Aufgabe der Edge-Anwendung.

Sollte ich im Arbeitsspeicher oder auf der Festplatte puffern?

Auf der Festplatte. Puffer im Arbeitsspeicher gehen bei einem Neustart oder Stromausfall verloren, und solche Ereignisse treten oft genau dann auf, wenn auch das Netzwerk ausfällt.

Wie gehen Konsumenten nach der Wiedergabe mit doppelten Nachrichten um?

Fügen Sie jedem Payload eine Sequenznummer oder eindeutige ID hinzu und lassen Sie Konsumenten bereits gespeicherte IDs ignorieren. Historians deduplizieren oft zusätzlich anhand von Zeitstempel und Tag.

Überlastet eine große Wiedergabe den Broker?

Wird alles auf einmal gesendet, kann sie Ratenlimits auslösen oder Subscriber überfordern. Drosseln Sie die Wiedergabe auf eine feste Rate, wie im Beispiel, und richten Sie die Rate an Ihrem Tarif aus.