MQTT 5 Funktionen, die in der Praxis zählen: Reason Codes, Shared Subscriptions, Expiry und mehr

MQTT 5 ist die aktuelle Version des MQTT-Standards und behebt viele Schwachstellen, auf die Teams beim Betrieb von MQTT 3.1.1 in der Produktion stoßen. In diesem Leitfaden erfahren Sie, welche Funktionen von MQTT 5 die tägliche Entwicklungsarbeit tatsächlich verändern, sehen sie im direkten Vergleich mit 3.1.1 und probieren sie mit den Mosquitto-CLI-Tools und Python selbst gegen den öffentlichen MQTT-Broker broker.connectmq.com aus.

Wenn das Protokoll neu für Sie ist, beginnen Sie mit Was ist ein MQTT-Broker? und kehren Sie dann hierher zurück.

Warum es MQTT 5 gibt

MQTT 3.1.1 ist klein, stabil und überall verbreitet. Es lässt aber wichtige Fragen offen. Warum wurde meine Verbindung abgelehnt? Warum ist mein Abonnement fehlgeschlagen? Wie verteile ich Nachrichten eines Topics auf mehrere Worker? Wie lange soll der Broker meine Session oder einen alten Befehl aufbewahren?

In 3.1.1 lauteten die Antworten meist „Der Broker hat den Socket geschlossen“ oder „Schreib es selbst in die Payload“. MQTT 5, 2019 von OASIS standardisiert, ergänzt das Protokoll um explizite Rückmeldungen, Metadaten und Limits und behält dabei dasselbe Publish/Subscribe-Modell bei. Ihre Topics, QoS-Stufen und Retained Messages funktionieren genau wie bisher. Die vollständige Spezifikation erhalten Sie bei OASIS.

MQTT 3.1.1 vs. MQTT 5 im Überblick

Reason Codes: Wissen, warum etwas fehlgeschlagen ist

Die mit Abstand nützlichste Neuerung von MQTT 5 für den Betrieb sind Reason Codes. CONNACK, PUBACK, PUBREC, SUBACK, UNSUBACK und DISCONNECT tragen jeweils einen Ein-Byte-Code, und der Server kann einen menschenlesbaren Reason String mitschicken.

Codes unter 0x80 bedeuten Erfolg, 0x80 und höher stehen für Fehler. Für Geräte im Feld ist das ein großer Fortschritt: Die Firmware kann auf „falsches Passwort“ (anhalten und alarmieren) anders reagieren als auf „Server busy“ (Backoff und erneut versuchen). ConnectMQ gibt Reason Codes wie 0x97 Quota exceeded und 0x89 Server busy zurück, sodass Ihre Clients ein Quota-Problem von einer vorübergehenden Überlast unterscheiden können.

Session Expiry und Message Expiry

Session Expiry Interval

In 3.1.1 lebt eine persistente Session (Clean Session = false), bis der Client zurückkehrt – womöglich für immer. MQTT 5 ersetzt das durch zwei Einstellungen: Clean Start (neu beginnen oder fortsetzen) und Session Expiry Interval (wie viele Sekunden der Broker die Session nach dem Trennen aufbewahrt). Ein Sensor, der zehn Minuten schläft, kann eine Session von 15 Minuten anfordern und erhält beim Aufwachen trotzdem seine aufgelaufenen QoS-1-Nachrichten, während verwaiste Sessions von selbst verschwinden.

Bei ConnectMQ ist Session Expiry pro Plan begrenzt; auf dem öffentlichen MQTT-Broker, der für Tests und Prototypen gedacht ist, liegt das Maximum bei 30 Minuten.

Message Expiry Interval

Ein Publisher kann jeder Nachricht eine Lebensdauer in Sekunden mitgeben. Wartet die Nachricht nach Ablauf des Intervalls noch in einer Warteschlange oder ist sie als Retained Message gespeichert, verwirft der Broker sie. Genau das wollen Sie bei Befehlen wie „Ventil öffnen“, deren Ausführung eine Stunde zu spät gefährlich wäre.

mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -i demo-a1b2-pub \
  -t "connectmq/demo/a1b2/commands/valve1" -m "open" -q 1 \
  -D publish message-expiry-interval 60

Shared Subscriptions: Integrierte Lastverteilung

Bei einem normalen Abonnement erhält jeder Subscriber jede Nachricht. Das wird zum Problem, wenn eine einzelne Backend-Instanz mit Tausenden Geräten nicht Schritt halten kann. Mit Shared Subscriptions in MQTT 5 können mehrere Clients einer Gruppe beitreten; der Broker stellt jede Nachricht nur einem Mitglied der Gruppe zu.

Die Syntax lautet $share/{group}/{filter}. Publisher müssen nichts ändern: Sie veröffentlichen weiterhin auf das normale Topic.

Ausprobieren: Zwei Worker teilen sich ein Topic

Öffnen Sie zwei Terminals und starten Sie zwei Consumer in derselben Gruppe workers. Ersetzen Sie a1b2 durch Ihre eigene Zufalls-ID, denn der öffentliche MQTT-Broker ist für alle lesbar.

# Terminal 1
mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 -i demo-a1b2-w1 \
  -t '$share/workers/connectmq/demo/a1b2/jobs' -v

# Terminal 2
mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 -i demo-a1b2-w2 \
  -t '$share/workers/connectmq/demo/a1b2/jobs' -v

Veröffentlichen Sie nun aus einem dritten Terminal sechs Jobs:

for i in 1 2 3 4 5 6; do
  mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -i demo-a1b2-pub \
    -t "connectmq/demo/a1b2/jobs" -m "job-$i" -q 1
done

Als wir das gegen broker.connectmq.com ausgeführt haben, erhielt Worker 1 job-1, job-3, job-5 und Worker 2 job-2, job-4, job-6. Jeder Job wurde genau einmal verarbeitet, und ein dritter Worker verteilt die Last weiter, ohne dass die Publisher angepasst werden müssen. Die einfachen Anführungszeichen um das Topic sind in der Bash wichtig, damit die Shell $share nicht expandiert.

Typische Einsatzgebiete sind horizontal skalierte Ingestion-Dienste, Befehlsverarbeitung und jeder Consumer, den Sie als mehrere Container-Replikate betreiben.

Topic Aliases: Kleinere Pakete auf schmalen Verbindungen

Industrielle Topics können lang sein, zum Beispiel acme/izmir/machining/line1/cnc1/telemetry. Diesen String in jedem PUBLISH mitzuschicken, verschwendet auf Mobilfunk- oder Satellitenverbindungen Bandbreite. Mit einem Topic Alias sendet der Client das vollständige Topic einmal zusammen mit einem ganzzahligen Alias und danach nur noch die Alias-Nummer. Jede Seite kündigt in CONNECT bzw. CONNACK an, wie viele Aliases sie akzeptiert (Topic Alias Maximum). Die meisten Client-Bibliotheken übernehmen die Aliases für Sie, sobald Sie sie aktivieren.

Solch lange Topics sind typisch für einen Unified Namespace (UNS), dessen Topic-Baum die ISA-95-Hierarchie Enterprise → Site → Area → Line → Cell abbildet, etwa acme/werk-stuttgart/zerspanung/linie-1/cnc-1/telemetrie. Gerade dort, wo ein Werk oder eine Außenanlage per Mobilfunk angebunden ist, spart ein Topic Alias bei jeder Nachricht spürbar Bytes, ohne dass Sie die sprechende UNS-Struktur aufgeben müssen.

User Properties: Metadaten, ohne die Payload anzufassen

User Properties sind beliebige UTF-8-Schlüssel/Wert-Paare, die an ein Paket angehängt werden. Nutzen Sie sie für Hinweise zum Content-Type, Schemaversionen, Trace-IDs oder die Firmware-Version des Senders, während die Payload ein sauberer Sensorwert bleibt.

mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -i demo-a1b2-pub \
  -t "connectmq/demo/a1b2/sensors/t1" -m '{"temperature":22.4}' \
  -D publish user-property schema v2 -D publish user-property fw 1.4.0

Ein Subscriber kann sie mit mosquitto_sub -V mqttv5 -F "%t %p %P" ausgeben.

Request/Response mit Response Topic und Correlation Data

Viele IoT-Abläufe sind eigentlich RPC-Aufrufe: „Lies deine Konfiguration“, „Neustart“, „Gib mir die letzten 10 Werte“. MQTT 5 standardisiert das:

1. Der Anfragende abonniert sein eigenes Antwort-Topic.
2. Er veröffentlicht die Anfrage mit einer Property Response Topic, die auf dieses Antwort-Topic zeigt, und mit Correlation Data, die die Anfrage identifizieren.
3. Der Antwortende antwortet auf dem Response Topic und kopiert die Correlation Data zurück.

Da das Response Topic an die Subscriber weitergegeben wird, braucht der Antwortende kein fest codiertes Routing-Wissen. So sehen Sie die eintreffenden Properties:

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 -i demo-a1b2-s \
  -t "connectmq/demo/a1b2/#" -F "%t %p | response=%R props=%P"

Flusskontrolle und vom Server angekündigte Limits

Receive Maximum

Receive Maximum begrenzt, wie viele QoS-1- und QoS-2-Nachrichten gleichzeitig unterwegs sein dürfen (gesendet, aber noch nicht bestätigt). Der Client teilt dem Server sein Limit im CONNECT mit, der Server dem Client seines im CONNACK. Ein kleiner Mikrocontroller kann einen niedrigen Wert setzen, damit er nie überflutet wird, und der Broker kann sich vor einem Client schützen, der schneller veröffentlicht, als er bestätigen kann. Mehr zu Bestätigungen lesen Sie in MQTT QoS-Stufen erklärt.

Limits im CONNACK

In 3.1.1 hat ein Client Limits erst durch Fehlschläge entdeckt. In MQTT 5 kündigt der Server sie in den CONNACK-Properties an, darunter Maximum QoS, Retain Available, Wildcard Subscription Available, Shared Subscription Available, Receive Maximum und Topic Alias Maximum. Ein gut geschriebener Client liest diese Werte einmal aus und passt sich an, etwa indem er auf QoS 1 heruntergeht, wenn QoS 2 im aktuellen Plan nicht angeboten wird.

MQTT 5 in Python mit Paho 2.x

Das folgende Skript verbindet sich mit MQTT 5, gibt die im CONNACK angekündigten Limits aus und führt einen vollständigen Request/Response-Durchlauf mit Message Expiry, User Properties, Response Topic und Correlation Data aus.

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

BROKER = "broker.connectmq.com"
PREFIX = "connectmq/demo/a1b2"          # eigene Zufalls-ID verwenden
REQ_TOPIC = f"{PREFIX}/rpc/request"
RESP_TOPIC = f"{PREFIX}/rpc/response"


def on_connect(client, userdata, flags, reason_code, properties):
    print("CONNACK:", reason_code)
    for name in ("MaximumQoS", "RetainAvailable", "ReceiveMaximum",
                 "TopicAliasMaximum", "SharedSubscriptionAvailable"):
        if hasattr(properties, name):
            print(f"  {name} = {getattr(properties, name)}")
    client.subscribe(REQ_TOPIC, qos=1)
    client.subscribe(RESP_TOPIC, qos=1)


def on_message(client, userdata, msg):
    props = msg.properties
    if msg.topic == REQ_TOPIC:
        # Antwortende Seite: auf dem angefragten Topic antworten, Correlation Data zurückgeben
        reply = Properties(PacketTypes.PUBLISH)
        reply.CorrelationData = props.CorrelationData
        client.publish(props.ResponseTopic, b'{"uptime": 3600}', qos=1,
                       properties=reply)
    elif msg.topic == RESP_TOPIC:
        print("Response:", msg.payload.decode(),
              "correlation:", props.CorrelationData.decode())


connect_props = Properties(PacketTypes.CONNECT)
connect_props.SessionExpiryInterval = 600   # Session 10 Minuten behalten
connect_props.ReceiveMaximum = 10           # max. 10 unbestätigte eingehende Nachrichten

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id="demo-a1b2-py", protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, clean_start=True, properties=connect_props)
client.loop_start()
time.sleep(2)

request = Properties(PacketTypes.PUBLISH)
request.ResponseTopic = RESP_TOPIC
request.CorrelationData = uuid.uuid4().hex.encode()
request.MessageExpiryInterval = 30
request.UserProperty = [("schema", "v1"), ("app", "blog-demo")]
client.publish(REQ_TOPIC, b'{"cmd": "status"}', qos=1, properties=request)

time.sleep(3)
client.loop_stop()
client.disconnect()

In einem echten System sind Anfragender und Antwortender verschiedene Clients, der Code bleibt aber derselbe. Für die Produktion verbinden Sie sich mit YOUR_USERNAME, YOUR_PASSWORD und YOUR_CLIENT_ID aus dem Portal mit Ihrem eigenen Broker, idealerweise über TLS auf Port 8883 (siehe MQTT-Sicherheit: Best Practices).

MQTT-5-Unterstützung in ConnectMQ

ConnectMQ-Broker sprechen MQTT 3.1.1 und MQTT 5.0 über TCP, TLS und WebSocket, sodass sich alte und neue Geräte denselben Broker teilen können. Folgende MQTT-5-Funktionen sind derzeit implementiert:

Die ConnectMQ-Erweiterung für VS Code zeigt die Reason Codes von CONNACK, SUBACK und PUBACK direkt an – praktisch, wenn Sie herausfinden wollen, warum ein Abonnement abgelehnt wurde.

Starten Sie mit ConnectMQ

Alle Beispiele in diesem Artikel können Sie ohne Konto auf broker.connectmq.com ausprobieren. Wenn Sie bereit für Ihren eigenen, isolierten Broker mit Zugangsdaten sind, registrieren Sie sich unter app.connectmq.com: Der Serverless-Plan ist kostenlos, benötigt keine Kreditkarte und stellt Ihnen in wenigen Minuten einen Broker bereit. Die Optionen vergleichen wir in Was ist ein verwalteter MQTT-Broker?.

Häufig gestellte Fragen

Ist MQTT 5 abwärtskompatibel zu MQTT 3.1.1?

Das Übertragungsformat unterscheidet sich, daher wählt ein Client beim Verbinden eine Version. Ein Broker, der beide unterstützt, wie ConnectMQ, ermöglicht es Clients mit MQTT 3.1.1 und MQTT 5, auf denselben Topics Nachrichten auszutauschen.

Brauche ich MQTT 5 für Shared Subscriptions?

Shared Subscriptions sind im MQTT-5-Standard definiert. Verbinden Sie sich mit MQTT 5 (zum Beispiel mit -V mqttv5 in den Mosquitto-Tools) und abonnieren Sie $share/{group}/{filter}, um eine lastverteilte Zustellung zu erhalten.

Was ist der Unterschied zwischen Session Expiry und Message Expiry?

Session Expiry steuert, wie lange der Broker die Session und die aufgelaufenen Nachrichten eines getrennten Clients aufbewahrt. Message Expiry steuert, wie lange eine einzelne Nachricht gültig bleibt, bevor der Broker sie verwirft.

Welche MQTT-5-Funktion sollte ich zuerst einführen?

Beginnen Sie mit Reason Codes, denn sie machen Fehler ohne Zusatzaufwand sichtbar. Ergänzen Sie dann Session Expiry für schlafende Geräte und Shared Subscriptions für Backend-Dienste, die skalieren müssen.