MQTT Shared Subscriptions: Lastverteilung auf Message-Consumer

Mit MQTT Shared Subscriptions können mehrere Clients als Gruppe abonnieren, sodass jede Nachricht eines Topics nur an ein Mitglied dieser Gruppe zugestellt wird statt an alle. Nach diesem Artikel verstehen Sie die Syntax $share/group/topic, wissen, wie die Zustellung funktioniert, wenn Worker hinzukommen und wegfallen, wie Sie damit Backend-Consumer skalieren und wie Sie prüfen, ob Ihr Broker die Funktion unterstützt.

Warum normale Subscriptions Consumer nicht skalieren

Beim klassischen Publish/Subscribe erhält jeder Subscriber, dessen Filter passt, eine eigene Kopie jeder Nachricht. Genau das wollen Sie für Dashboards: Jeder Bildschirm zeigt jeden Messwert. Genau das wollen Sie aber nicht für die Backend-Verarbeitung. Betreiben Sie drei Instanzen eines Dienstes, der Telemetrie in eine Datenbank schreibt, bedeuten drei normale Subscriptions, dass jeder Messwert dreimal geschrieben wird.

Vor MQTT 5 waren die Workarounds umständlich: ein einzelner Consumer, der zum Flaschenhals wurde, eine manuelle Aufteilung der Topics auf Instanzen oder eine Bridge von MQTT in ein separates Queueing-System. Shared Subscriptions lösen das Problem direkt im MQTT-Broker.

So funktionieren MQTT Shared Subscriptions

Eine Shared Subscription verwendet ein spezielles, in MQTT 5 standardisiertes Filterformat:

$share/<ShareName>/<TopicFilter>

• $share ist das feste Präfix, das die Subscription als geteilt kennzeichnet.
• <ShareName> benennt die Consumer-Gruppe. Er darf kein /, + oder # enthalten.
• <TopicFilter> ist ein normaler Filter, Wildcards eingeschlossen.

Jeder Client, der denselben $share/<ShareName>/<TopicFilter> abonniert, tritt derselben Gruppe bei. Für jede passende Nachricht wählt der Broker ein Mitglied der Gruppe aus und stellt ihm die Nachricht zu. Unterschiedliche Share-Namen sind unabhängige Gruppen, und normale Subscriber des einfachen Topics erhalten weiterhin ihre eigenen Kopien.

Die Spezifikation überlässt die Wahl des Mitglieds dem Broker. Round-Robin, Zufall und „am wenigsten ausgelastet“ sind gängige Strategien; verlassen Sie sich also weder auf eine bestimmte Reihenfolge noch darauf, dass Nachrichten eines Geräts immer beim selben Worker landen.

Zustellsemantik, die Sie kennen müssen

1. Die Reihenfolge gilt pro Mitglied, nicht pro Topic. Jeder Worker sieht seine eigenen Nachrichten in der richtigen Reihenfolge, aber zwei aufeinanderfolgende Messwerte desselben Sensors können an verschiedene Worker gehen und parallel verarbeitet werden. Ist die Reihenfolge wichtig, nehmen Sie eine Sequenznummer oder einen Zeitstempel auf und lösen Sie sie in Ihrer Datenbank auf.
2. QoS gilt weiterhin. Trennt sich ein Worker bei QoS 1, bevor er eine Nachricht bestätigt hat, stellen Broker sie typischerweise beim Reconnect erneut derselben Session zu oder leiten sie an ein anderes Mitglied weiter; das Verhalten unterscheidet sich je nach Broker. Machen Sie Handler idempotent.
3. Retained Messages werden beim Abonnieren nicht gesendet. Laut Spezifikation werden Retained Messages beim Anlegen einer Shared Subscription nicht zugestellt, weil unklar wäre, welches Mitglied sie erhalten soll.
4. No Local ist verboten. Die Subscription-Option No Local bei einer Shared Subscription zu setzen, ist ein Protokollfehler.
5. Sessions gelten pro Mitglied. Jeder Worker hat seine eigene Client-ID und Session. Eine Persistent Session auf einem Worker hält dessen Subscription aufrecht, während er offline ist. Wie sich das auf die Pufferung auswirkt, lesen Sie unter MQTT Persistent Sessions.

Beispiel: Python-Worker skalieren

Der folgende Worker tritt einer Gruppe namens db-writers bei und verarbeitet Telemetrie. Starten Sie ihn in zwei oder drei Terminals mit unterschiedlichen Worker-Nummern, um die Lastverteilung zu beobachten.

import sys
import json
import paho.mqtt.client as mqtt

WORKER = sys.argv[1] if len(sys.argv) > 1 else "1"
FILTER = "$share/db-writers/connectmq/demo/s7h3/factory/+/telemetry"

def on_connect(client, userdata, flags, reason_code, properties):
    print(f"Worker {WORKER} connected: {reason_code}")
    client.subscribe(FILTER, qos=1)

def on_subscribe(client, userdata, mid, reason_code_list, properties):
    for rc in reason_code_list:
        if rc.is_failure:
            print(f"Subscribe rejected: {rc}")  # z. B. Shared Subscriptions not supported
        else:
            print(f"Joined group, granted {rc}")

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    print(f"Worker {WORKER} got {msg.topic}: {data}")
    # hier in die Datenbank schreiben; idempotent halten

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"db-writer-s7h3-{WORKER}",
                     protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_subscribe = on_subscribe
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

Starten Sie zwei Worker:

python worker.py 1
python worker.py 2

Veröffentlichen Sie dann eine Folge von Testmesswerten:

for i in $(seq 1 10); do
  mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -i sh-pub-s7h3 -q 1 \
    -t "connectmq/demo/s7h3/factory/line$((i % 3))/telemetry" \
    -m "{\"seq\": $i, \"temp\": 70}"
done

Unterstützt der Broker Shared Subscriptions, werden die zehn Nachrichten auf die Worker aufgeteilt, statt dass jeder Worker alle zehn ausgibt. Ersetzen Sie s7h3 durch eine eigene Zufallszeichenfolge; der öffentliche MQTT-Broker ist für Tests gedacht und für alle lesbar, veröffentlichen Sie also nur Dummy-Daten.

Dieselbe Subscription sieht in MQTT.js so aus:

const mqtt = require("mqtt");
const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId: "db-writer-node-s7h3-1",
  protocolVersion: 5,
});
client.on("connect", () => {
  client.subscribe("$share/db-writers/connectmq/demo/s7h3/factory/+/telemetry",
    { qos: 1 }, (err, granted) => console.log(err || granted));
});
client.on("message", (topic, payload) => console.log(topic, payload.toString()));

So prüfen Sie die Unterstützung für Shared Subscriptions

Die Unterstützung unterscheidet sich zwischen Brokern und Tarifen; prüfen Sie also, statt es vorauszusetzen:

1. Lesen Sie das SUBACK. In MQTT 5 liefert ein Broker ohne Unterstützung für Shared Subscriptions den Reason Code 0x9E (Shared Subscriptions not supported). Der obige Callback on_subscribe gibt ihn aus.
2. Prüfen Sie das CONNACK. MQTT-5-Broker können im CONNACK Shared Subscription Available = 0 angeben. Fehlt die Property, wird Unterstützung angenommen.
3. Führen Sie den Aufteilungstest durch. Zwei Mitglieder, zehn Nachrichten. Geben beide Mitglieder alle zehn aus, hat der Broker $share/... als gewöhnlichen Topic-Filter behandelt und nicht als Gruppe.
4. Prüfen Sie den Tarif. Verwaltete MQTT-Broker koppeln die Funktion oft an eine Tarifstufe. Bei ConnectMQ sind Shared Subscriptions ab dem Tarif Serverless (Paid) enthalten.
5. Testen Sie MQTT 3.1.1 separat. Manche Broker akzeptieren $share von 3.1.1-Clients als Erweiterung, andere verlangen MQTT 5. Verlassen Sie sich nur darauf, wenn Ihr Broker es dokumentiert.

Shared Subscriptions im Vergleich zu anderen Ansätzen

Es gibt andere Wege, Arbeit aufzuteilen, und es hilft zu wissen, wann welcher passt. Manuelle Topic-Partitionierung (Worker 1 abonniert line1, Worker 2 line2) bietet eine strikte Reihenfolge pro Partition, erfordert aber jedes Mal eine Neukonfiguration, wenn Sie einen Worker oder eine Linie hinzufügen. Ein einzelner Consumer, der eine interne Warteschlange speist, hält die Reihenfolge einfach, macht diesen einen Prozess aber zum Flaschenhals und Single Point of Failure. Shared Subscriptions brauchen keine Koordination zwischen Workern und skalieren, indem Sie einfach weitere Instanzen starten, um den Preis der Reihenfolge pro Topic über Worker hinweg. Für die meisten Szenarien der Telemetrieaufnahme und Alarmverarbeitung ist das der richtige Kompromiss.

Muster und Fallstricke

• Eine Gruppe pro Zweck. Verwenden Sie einen Share-Namen pro unabhängiger Aufgabe (db-writers, alerting, exporter). Jede Aufgabe erhält dann jede Nachricht genau einmal über ihre Worker verteilt.
• Autoscaling. Einen Worker hinzuzufügen heißt einfach, einen weiteren Client mit demselben Filter zu verbinden; einen zu entfernen heißt, ihn zu trennen. Geben Sie jeder Instanz eine eindeutige Client-ID, wie in den Best Practices für die MQTT Client-ID erklärt.
• Flusskontrolle. Kombinieren Sie Shared Subscriptions auf jedem Worker mit einem kleinen Receive Maximum aus MQTT 5, damit einer langsamen Instanz keine große Welle unbestätigter Nachrichten zugeteilt wird.
• Wildcards sind kein Problem. $share/group/factory/# verteilt einen ganzen Zweig. In einem Unified Namespace (UNS), dessen Topics der ISA-95-Hierarchie folgen, kann eine Historian-Gruppe so etwa $share/historian/acme/werk-stuttgart/# abonnieren und die Maschinendaten eines ganzen Werks auf mehrere Instanzen verteilen. Es gelten die üblichen Matching-Regeln, erklärt unter MQTT-Wildcards.
• Nicht für Fan-out verwenden. Wenn jeder Consumer jede Nachricht sehen muss, verwenden Sie normale Subscriptions.

Neu bei MQTT? Das MQTT-Tutorial behandelt Subscriptions, QoS und Sessions von Anfang an.

Consumer mit ConnectMQ skalieren

ConnectMQ ist ein verwalteter MQTT-Broker mit Unterstützung für MQTT 3.1.1 und 5.0; Shared Subscriptions sind ab dem Tarif Serverless (Paid) verfügbar, zusammen mit QoS 2, Nachrichten bis 1 MB und bis zu 1.000 Verbindungen. Wenn Sie einen isolierten Cluster benötigen, stehen die Optionen Dedicated und BYOC zur Verfügung. Registrieren Sie sich und starten Sie mit dem kostenlosen Serverless-Tarif ohne Kreditkarte, und wechseln Sie in einen höheren Tarif, wenn Ihr Backend Consumer-Gruppen benötigt. Verbinden Sie Ihre Worker mit YOUR_USERNAME, YOUR_PASSWORD und YOUR_CLIENT_ID.

Häufig gestellte Fragen

Wie lautet die Syntax einer MQTT Shared Subscription?

Verwenden Sie $share/<ShareName>/<TopicFilter>, zum Beispiel $share/workers/sensors/+/temperature. Alle Clients mit demselben Share-Namen und Filter bilden eine Gruppe.

Funktionieren Shared Subscriptions mit MQTT 3.1.1?

Sie sind in MQTT 5 definiert. Manche Broker akzeptieren die $share-Syntax von 3.1.1-Clients als Erweiterung, Sie sollten das aber in der Dokumentation Ihres Brokers bestätigen, bevor Sie sich darauf verlassen.

Werden Retained Messages an Shared Subscriptions zugestellt?

Nein. Beim Anlegen einer Shared Subscription werden ihr keine Retained Messages gesendet. Lesen Sie den retained Zustand über eine separate normale Subscription, wenn ein Worker ihn benötigt.

Wie entscheidet der Broker, welcher Subscriber eine Nachricht erhält?

Die MQTT-Spezifikation überlässt das dem Broker. Round-Robin und zufällige Auswahl sind üblich; verlassen Sie sich also nie darauf, dass ein bestimmtes Mitglied eine bestimmte Nachricht erhält.