MQTT und Redis: den aktuellen Gerätestatus cachen

Um den aktuellen Gerätestatus aus MQTT in Redis zu cachen, betreibst du einen kleinen Subscriber, der auf deine Telemetrie-Topics hört und jede Nachricht in einen Redis-Hash schreibt, dessen Key die Geräte-ID ist. Deine APIs, Dashboards und Hintergrundjobs lesen den aktuellen Status dann mit einem einzigen HGETALL, ohne MQTT-Verbindung und ohne auf die nächste Nachricht zu warten.

Dieses Tutorial baut den Subscriber in Python, schützt ihn vor Updates in falscher Reihenfolge, ergänzt einen Redis Stream für die jüngste Historie und vergleicht den Ansatz mit Retained Messages und Redis Pub/Sub.

Wann ein Redis-Cache sinnvoll ist

MQTT hat bereits eine "letzter Wert"-Funktion: Retained Messages. Ein neuer Subscriber erhält sofort die letzte Retained Message jedes Topics. Wozu also Redis?

Brauchen nur MQTT-Clients den letzten Wert, reichen Retained Messages. Sobald Code ohne MQTT den Status braucht, ist ein Redis-Cache einfacher und schneller.

Das Datenmodell

Halte es langweilig und vorhersehbar:

• device:<id>: ein Hash mit den letzten Feldern: temp, humidity, battery, ts (Geräte-Timestamp), topic, seen (wann die Bridge geschrieben hat).
• devices:online: ein optionales Set von Geräte-IDs, gepflegt über Status-Topics.
• telemetry: ein optionaler Stream mit den letzten Rohnachrichten für Replay und Debugging.

Geräte publishen JSON auf testmqtt/redis/<device>/state, zum Beispiel {"ts": 1760700000123, "temp": 21.4, "battery": 87}. Ein Timestamp vom Gerät (oder eine Sequenznummer) ist das, womit du später veraltete Updates ablehnen kannst. Tipps zum Parsen findest du unter JSON-Payloads in Python.

Der Subscriber

Diese Bridge nutzt paho-mqtt 2.x und redis-py. Das Lua-Skript schreibt den Hash nur, wenn der eingehende Timestamp neuer ist als der gespeicherte, und erneuert eine TTL, damit stille Geräte von selbst verschwinden:

import json, os, time
import paho.mqtt.client as mqtt
import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

# KEYS[1] = Geräte-Key; ARGV = ts, ttl, danach Feld/Wert-Paare
UPSERT_IF_NEWER = r.register_script("""
local cur = redis.call('HGET', KEYS[1], 'ts')
if cur and tonumber(cur) >= tonumber(ARGV[1]) then
  return 0
end
redis.call('HSET', KEYS[1], 'ts', ARGV[1], unpack(ARGV, 3))
redis.call('EXPIRE', KEYS[1], ARGV[2])
return 1
""")

TTL_SECONDS = 900   # Geräte vergessen, die 15 Minuten still sind

def on_connect(client, userdata, flags, reason_code, properties):
    print("verbunden:", reason_code)
    client.subscribe("testmqtt/redis/+/state", qos=1)

def on_message(client, userdata, msg):
    device = msg.topic.split("/")[2]
    try:
        data = json.loads(msg.payload)
    except ValueError:
        print("überspringe Nicht-JSON-Payload auf", msg.topic)
        return

    ts = int(data.pop("ts", time.time() * 1000))
    fields = []
    for k, v in data.items():
        plain = isinstance(v, (str, int, float)) and not isinstance(v, bool)
        fields += [k, v if plain else json.dumps(v)]   # bool/None/Objekte als JSON
    fields += ["topic", msg.topic, "seen", int(time.time() * 1000)]

    written = UPSERT_IF_NEWER(keys=[f"device:{device}"], args=[ts, TTL_SECONDS, *fields])
    if not written:
        print(f"veraltetes Update für {device} ignoriert (ts={ts})")

    # optionale kurze Historie, begrenzt auf ~100k Einträge
    r.xadd("telemetry", {"device": device, "payload": msg.payload}, maxlen=100_000, approximate=True)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="redis-state-bridge")
client.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
client.tls_set()
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 8883)
client.loop_forever()

Hinweise zum Design:

• Atomares Compare-and-Set. Lua-Skripte laufen in Redis atomar, also können zwei Bridge-Instanzen (oder eine QoS-1-Neuzustellung) nicht um die Wette einen älteren Status schreiben.
• Flache Felder. Strings und Zahlen gehen unverändert hinein; Booleans, Nullwerte und verschachtelte Objekte werden als JSON-Strings gespeichert. Redis-Hashes haben nur eine Ebene, kämpfe nicht dagegen an.
• TTL als Offline-Signal. Läuft ein Key ab, hat sich das Gerät TTL_SECONDS lang nicht gemeldet. Für ein explizites Signal kombinierst du das mit einer Last-Will-Nachricht auf einem Status-Topic.
• Duplikate schaden nicht. Denselben Status zweimal zu schreiben ändert nichts, dadurch ist dieses Ziel von Natur aus idempotent.

Den Status lesen

Jeder Service kann jetzt ohne MQTT beantworten, was Gerät X gerade tut:

redis-cli HGETALL device:sensor-1
redis-cli HGET device:sensor-1 temp
redis-cli TTL device:sensor-1
redis-cli --scan --pattern 'device:*'

Für ein Dashboard mit Live-Updates lädst du die erste Ansicht über einen REST-Endpoint, der aus Redis liest, und lässt den Browser für alle späteren Änderungen MQTT über Secure WebSockets abonnieren (Port 8084, Pfad /mqtt). Der Cache beantwortet das "Jetzt", MQTT liefert das "Als Nächstes".

Redis Streams für die jüngste Historie

Die XADD-Zeile führt ein begrenztes Log der Rohnachrichten. Streams unterstützen Consumer Groups, sodass mehrere Worker Updates mit Bestätigung verarbeiten können, und beim Debuggen kannst du einen Zeitraum erneut abspielen:

redis-cli XLEN telemetry
redis-cli XREVRANGE telemetry + - COUNT 5

redis-cli XGROUP CREATE telemetry alerts $ MKSTREAM
redis-cli XREADGROUP GROUP alerts worker-1 COUNT 10 BLOCK 5000 STREAMS telemetry '>'

Halte den Stream kurz (MAXLEN), denn er liegt im Speicher. Für langfristige Historie schreibst du in eine Zeitreihendatenbank oder nach Postgres; siehe MQTT nach PostgreSQL und Aufbewahrung und Speicherung von MQTT-Daten.

MQTT vs Redis Pub/Sub

Redis hat auch Pub/Sub, und es ist verlockend, MQTT ganz wegzulassen. Die beiden sind aber nicht austauschbar:

• Redis Pub/Sub ist Fire-and-Forget zwischen Prozessen, die gerade mit Redis verbunden sind. Es gibt kein QoS, keinen Retained-Wert, kein Last Will, kein für Geräte gedachtes ACL-Modell pro Topic, und Clients müssen eine Redis-Verbindung halten.
• MQTT ist für Geräte in unzuverlässigen Netzen gebaut: QoS 1 und 2, persistente Sessions, Retained Messages, Keep Alive, Last Will und schlanke Clients für Mikrocontroller.

Eine gängige Aufteilung ist MQTT am Edge und Redis im Backend. Der ausführliche Vergleich steht in MQTT vs Redis Pub/Sub.

Testen

Starte Redis lokal und führe die Bridge aus:

docker run -d --name redis -p 6379:6379 redis:7
python mqtt_redis_state.py

Publishe einen neuen Status und danach einen älteren mit kleinerem Timestamp:

mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
  -t testmqtt/redis/sensor-1/state -m '{"ts": 2000, "temp": 22.5, "battery": 90}'

mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
  -t testmqtt/redis/sensor-1/state -m '{"ts": 1000, "temp": 19.0, "battery": 95}'

redis-cli HGETALL device:sensor-1

temp muss weiterhin 22.5 sein, und die Bridge sollte veraltetes Update für sensor-1 ignoriert loggen. Du kannst die Payloads auch von Hand aus dem Online-MQTT-Client senden und mit redis-cli MONITOR jeden Befehl verfolgen, den die Bridge absetzt.

Test-Status aus dem Browser publishen

Ein Redis-Statuscache ist eines der günstigsten Upgrades für ein MQTT-System: ein Subscriber, ein Hash pro Gerät, und jeder Service hat sofort Zugriff auf das "Jetzt". Bau die Timestamp-Prüfung von Anfang an ein, halte die Historie aus dem Hash heraus und lass MQTT weiter das tun, was es am besten kann: die Updates transportieren.

Häufig gestellte Fragen