Python MQTT JSON: Sensordaten mit Zeitstempeln und Schema-Versionen veröffentlichen

JSON-Payloads sind der Standardweg, um mit Python und MQTT Sensordaten zwischen Geräten, Dashboards und Datenbanken zu bewegen, doch kleine Fehler wie lokale Zeitstempel, NaN-Werte oder unversionierte Felder verursachen später große Kopfschmerzen. In diesem Python-MQTT-JSON-Tutorial bauen Sie einen Python-Publisher, der Messwerte in kompaktes, gültiges JSON mit UTC-Zeitstempeln, Sequenznummern und einer Schema-Version serialisiert und sie mit MQTT-5-Content-Type-Metadaten an broker.connectmq.com sendet, sowie einen Subscriber, der das Empfangene validiert. Am Ende haben Sie ein Payload-Format, das Sie auch nutzen können, wenn Ihr Projekt wächst.

Warum JSON für MQTT-Payloads?

MQTT ist egal, was in einer Nachricht steckt; die Payload besteht einfach aus Bytes, bis zur Größengrenze des Brokers. Diese Freiheit bedeutet, dass Sie ein Format wählen müssen. JSON ist die übliche Wahl, weil:

• jede Sprache und jedes Werkzeug es parsen kann, von Node-RED und Telegraf bis zu Datenbanken mit JSON-Spalten;
• es menschenlesbar ist, sodass Sie mit mosquitto_sub oder einem Topic-Explorer debuggen können;
• es selbstbeschreibend ist, sodass ein neues Feld ältere Consumer nicht bricht.

Der Kompromiss ist die Größe. JSON wiederholt die Feldnamen in jeder Nachricht, was bei eingeschränkten Verbindungen ins Gewicht fällt. Für die meisten WLAN- und Ethernet-Geräte sind diese Kosten vernachlässigbar, und die Vorteile beim Debuggen überwiegen. Wenn Sie später etwas Kompakteres brauchen, sind Binärformate wie Protobuf (das Sparkplug B verwendet) eine Option, wie im Leitfaden zu Sparkplug B und Unified Namespace (UNS) beschrieben.

Die Payload entwerfen

Eine gute Sensor-Payload beantwortet fünf Fragen: Welche Version des Formats ist das, welches Gerät hat sie gesendet, wann wurde gemessen, die wievielte Nachricht in der Folge ist es, und welche Werte mit welchen Einheiten enthält sie. Dieses Format verwendet das Tutorial:

{
  "v": 1,
  "device": "room1",
  "ts": "2026-10-16T08:15:30.123Z",
  "seq": 42,
  "temperature_c": 21.4,
  "humidity_pct": 45.2,
  "battery_v": null
}

Die Designentscheidungen hinter jedem Feld:

Einheiten in den Feldnamen vermeiden verschachtelte {"value": 21.4, "unit": "C"}-Objekte für jeden Messwert, was Payloads klein und Abfragen einfach hält. Welche Konvention Sie auch wählen, wenden Sie sie auf allen Geräten einheitlich an. In Industrieanlagen gehört der Kontext ohnehin ins Topic: Im UNS folgt es der ISA-95-Hierarchie, etwa acme/werk-stuttgart/montage/linie-1/zelle-3/telemetry, sodass die Payload nur noch Werte, Zeitstempel und Version tragen muss.

Einen Python-MQTT-JSON-Publisher bauen

Installieren Sie paho-mqtt und speichern Sie den Publisher als json_publisher.py. Ändern Sie k7f3q2 in Ihre eigene zufällige Zeichenfolge, damit Sie von anderen Nutzern des kostenlosen öffentlichen MQTT-Brokers getrennt bleiben (er ist für Tests gedacht, nicht für Produktionsdaten):

pip install "paho-mqtt>=2.0"

import json
import math
import random
import time
import uuid
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from typing import Optional

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

BROKER = "broker.connectmq.com"
PORT = 1883
PREFIX = "connectmq/demo/k7f3q2"
DEVICE = "room1"
SCHEMA_VERSION = 1


@dataclass
class Reading:
    device: str
    ts: str
    seq: int
    temperature_c: Optional[float]
    humidity_pct: Optional[float]
    battery_v: Optional[float] = None
    v: int = SCHEMA_VERSION


def utc_now_iso() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")


def clean(value: Optional[float], digits: int = 2) -> Optional[float]:
    """Floats runden und NaN oder Unendlich in None umwandeln, was zu JSON null wird."""
    if value is None or math.isnan(value) or math.isinf(value):
        return None
    return round(value, digits)


def read_sensor() -> tuple:
    # Durch echten Sensorcode ersetzen. Gelegentlich einen fehlgeschlagenen Lesevorgang simulieren.
    if random.random() < 0.05:
        return float("nan"), float("nan")
    return random.uniform(20, 25), random.uniform(40, 55)


def to_payload(reading: Reading) -> bytes:
    return json.dumps(asdict(reading), separators=(",", ":"), allow_nan=False).encode("utf-8")


client = mqtt.Client(
    mqtt.CallbackAPIVersion.VERSION2,
    client_id=f"json-pub-{uuid.uuid4().hex[:8]}",
    protocol=mqtt.MQTTv5,
)
# Privater ConnectMQ-Broker: client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD"); client.tls_set(); Port 8883
client.connect(BROKER, PORT, keepalive=60)
client.loop_start()

props = Properties(PacketTypes.PUBLISH)
props.ContentType = "application/json"
props.PayloadFormatIndicator = 1  # Payload ist UTF-8-Text

topic = f"{PREFIX}/sensors/{DEVICE}/telemetry"
seq = 0
try:
    while True:
        temperature, humidity = read_sensor()
        reading = Reading(
            device=DEVICE,
            ts=utc_now_iso(),
            seq=seq,
            temperature_c=clean(temperature),
            humidity_pct=clean(humidity, 1),
        )
        payload = to_payload(reading)
        client.publish(topic, payload, qos=1, properties=props)
        print(f"{len(payload):>4} bytes  {payload.decode()}")
        seq += 1
        time.sleep(2)
except KeyboardInterrupt:
    pass
finally:
    client.disconnect()
    client.loop_stop()

Führen Sie python json_publisher.py aus. Alle zwei Sekunden wird eine Zeile wie diese ausgegeben, und etwa jeder zwanzigste Messwert zeigt null-Werte, wo der simulierte Sensor ausgefallen ist:

 116 bytes  {"device":"room1","ts":"2026-10-16T08:15:30.123Z","seq":42,"temperature_c":21.4,"humidity_pct":45.2,"battery_v":null,"v":1}

Warum jedes Detail zählt

Kompakte Trennzeichen

separators=(",", ":") entfernt die Leerzeichen, die json.dumps standardmäßig einfügt. Bei einer Payload dieser Größe spart das kostenlos rund zehn Prozent. Hübsche Formatierung gehört in Ihre Debugging-Werkzeuge, nicht auf die Leitung.

allow_nan=False und die NaN-Falle

Pythons json.dumps schreibt standardmäßig bereitwillig NaN und Infinity, doch das ist kein gültiges JSON. Viele Consumer, darunter strikte Parser in JavaScript, Go und Datenbanken, verwerfen die gesamte Nachricht. Sensoren erzeugen ständig NaN, wenn ein Lesevorgang fehlschlägt, daher wandelt der Helfer clean() diese in None um, und allow_nan=False sorgt dafür, dass jeder durchrutschende Wert auf Ihrer Seite einen Fehler auslöst, statt nachgelagert Daten zu verfälschen.

UTC-Zeitstempel mit Millisekunden

Zeitstempel ohne Zeitzone sind mehrdeutig, und lokale Zeiten springen durch die Sommerzeit zweimal im Jahr. ISO 8601 in UTC mit Z-Suffix lässt sich als String korrekt sortieren und überall parsen. Hat Ihr Gerät keine Echtzeituhr, synchronisieren Sie es vor dem Veröffentlichen per NTP, oder lassen Sie ts weg und den Consumer die Empfangszeit erfassen.

Sequenznummern

seq kostet ein paar Bytes und gibt Consumern die Möglichkeit, fehlende Nachrichten (ein Sprung von 41 auf 43) und Duplikate (QoS 1 kann eine Nachricht zweimal zustellen) zu erkennen. Setzen Sie die Nummer beim Neustart zurück und lassen Sie Consumer einen Abfall auf null als Neustart behandeln.

MQTT-5-Content-Type

Die Properties ContentType und PayloadFormatIndicator teilen jedem MQTT-5-Consumer ohne Raten mit, dass die Payload UTF-8-JSON ist. Sie sind optional, erleichtern aber den Umgang mit generischen Werkzeugen und Topic-Bäumen mit gemischten Formaten. Unser Artikel zu den MQTT-5-Funktionen behandelt die übrigen Properties.

Ihr Schema versionieren

Ihre Payload wird sich ändern: neue Sensoren, umbenannte Felder, andere Einheiten. Planen Sie das ab der ersten Nachricht ein:

1. Additive Änderungen (ein neues optionales Feld) behalten dasselbe v. Gut geschriebene Consumer ignorieren Felder, die sie nicht kennen.
2. Inkompatible Änderungen (Umbenennen, Entfernen, Ändern von Einheiten oder Typen) erhöhen v auf 2.
3. Veröffentlichen Sie während einer Migration beide Versionen oder halten Sie Consumer in der Lage, beide zu lesen, mit Fallunterscheidung über v.
4. Verwenden Sie eine Versionsnummer niemals für ein anderes Format wieder.
5. Dokumentieren Sie jede Version neben Ihrer Topic-Struktur, damit alle denselben Vertrag nutzen.

Ein validierender Subscriber

Consumer sollten Payloads nie blind vertrauen. Dieser Subscriber parst, prüft die Version und die Pflichtfelder und weist alles andere zurück, ohne abzustürzen:

import json
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/k7f3q2"
REQUIRED = {"v", "device", "ts", "seq"}


def parse_reading(raw: bytes):
    try:
        data = json.loads(raw)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError(f"not valid JSON: {exc}")
    if not isinstance(data, dict) or not REQUIRED <= data.keys():
        raise ValueError("missing required fields")
    if data["v"] != 1:
        raise ValueError(f"unsupported schema version {data['v']}")
    return data


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/sensors/+/telemetry", qos=1)


def on_message(client, userdata, msg):
    try:
        r = parse_reading(msg.payload)
    except ValueError as exc:
        print(f"Rejected message on {msg.topic}: {exc}")
        return
    print(f"{r['device']} #{r['seq']} at {r['ts']}: {r.get('temperature_c')} C, {r.get('humidity_pct')} %")


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"json-sub-{uuid.uuid4().hex[:8]}")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

JSON-Nachrichten untersuchen

Für schnelle Prüfungen gegen den öffentlichen Broker leiten Sie den Kommandozeilen-Client an jq weiter, um eine formatierte, farbige Ausgabe zu erhalten:

mosquitto_sub -h broker.connectmq.com -p 1883 -i json-inspect-k7f3q2 -t 'connectmq/demo/k7f3q2/sensors/+/telemetry' | jq .

Auf einem privaten Broker zeigt der Live-Topic-Explorer im ConnectMQ-Portal jedes Topic Ihres Baums mit seiner neuesten Payload, sodass Sie Feldnamen, den Umgang mit null und Zeitstempel prüfen können, ohne einen Consumer zu schreiben.

Größengrenzen und Batching

Behalten Sie die Nachrichtengröße im Blick. Der öffentliche Broker akzeptiert bis zu 64 KB pro Nachricht, der kostenlose Serverless-Plan bis zu 256 KB. Einzelne Messwerte liegen weit darunter, doch wenn Sie bündeln, etwa eine Minute an Samples als Array senden, prüfen Sie vor dem Veröffentlichen len(payload) und teilen Sie große Batches auf. Eine weitere Möglichkeit, den Verkehr gering zu halten, ist, nur dann zu veröffentlichen, wenn sich ein Wert um mehr als einen festgelegten Schwellenwert ändert, plus einen periodischen Heartbeat, damit Consumer einen ruhigen Sensor von einem toten unterscheiden können. Dieser Ansatz reduziert das Nachrichtenvolumen bei sich langsam ändernden Werten wie der Raumtemperatur oft drastisch, ohne aussagekräftige Informationen zu verlieren.

Wie es weitergeht

Speichern Sie diese Payloads mit dem Tutorial MQTT zu SQLite in Python, dessen json_extract-Abfragen direkt mit diesem Format funktionieren, oder stellen Sie sie live mit dem Streamlit-MQTT-Dashboard dar. Für asynchrone Services lässt sich derselbe Payload-Code in einen aiomqtt-asyncio-Client übernehmen. Die Übersicht der MQTT-Client-Bibliotheken zeigt, wie andere Sprachen dasselbe JSON parsen, und die Dokumentation des Python-json-Moduls behandelt jede Encoder-Option.

Starten Sie mit ConnectMQ

ConnectMQ ist ein verwalteter MQTT-Broker mit MQTT-5-Unterstützung für Properties wie den Content-Type, einem Live-Topic-Explorer zur Prüfung von JSON-Payloads und TLS auf Port 8883. Entwickeln Sie Prototypen auf broker.connectmq.com und verlagern Sie Ihre Geräte dann auf einen privaten Cloud-MQTT-Broker. Registrieren Sie sich für den kostenlosen Serverless-Plan, ohne Kreditkarte.

Häufig gestellte Fragen

Wie sende ich JSON über MQTT in Python?

Serialisieren Sie Ihre Daten mit json.dumps(), kodieren Sie sie in UTF-8-Bytes und übergeben Sie sie als Payload an client.publish(). Auf der Empfängerseite rufen Sie json.loads(msg.payload) auf.

Gehören Zeitstempel in die Payload oder fügt der Consumer sie hinzu?

Nehmen Sie einen UTC-Messzeitstempel in die Payload auf, wenn das Gerät eine zuverlässige Uhr hat. Wenn nicht, lassen Sie den Consumer die Empfangszeit erfassen und ts weg.

Warum enthält meine JSON-Payload NaN?

Pythons json.dumps schreibt für Float-NaN-Werte standardmäßig NaN, was ungültiges JSON ist. Wandeln Sie NaN vor dem Kodieren in None um und setzen Sie allow_nan=False, um verbleibende Fälle abzufangen.

Wohin gehört die Schema-Version, in die Payload oder ins Topic?

Beides funktioniert. Ein v-Feld in der Payload ist einfacher und hält Topics stabil; ein Versionssegment im Topic ermöglicht es Consumern, nur die Versionen zu abonnieren, die sie verstehen.