MQTT Shared Subscriptions: Load Balancing mit $share

Im normalen MQTT erhält jeder Subscriber, dessen Filter auf ein Topic passt, eine eigene Kopie jeder Nachricht. Für Fan-out ist das perfekt, aber unpraktisch, wenn sich mehrere Backend-Worker einen Nachrichtenstrom aufteilen sollen. Shared Subscriptions, standardisiert in MQTT 5, lösen genau das: Subscriber treten einer benannten Gruppe bei, und der Broker stellt jede Nachricht nur an ein Mitglied dieser Gruppe zu.

Die $share-Syntax

Eine Shared Subscription ist eine normale Subscribe-Anfrage mit einem speziellen Topic-Filter:

$share/<ShareName>/<TopicFilter>

• $share – ein festes Präfix, das die Subscription als geteilt kennzeichnet.
• ShareName – der Gruppenname. Er muss mindestens ein Zeichen lang sein und darf weder / noch + oder # enthalten.
• TopicFilter – ein gewöhnlicher Filter, Wildcards inklusive. Siehe MQTT-Topics und Wildcards.

So steckt zum Beispiel $share/ingest/factory/+/telemetry den Subscriber für jedes Topic factory/<line>/telemetry in die Gruppe ingest. Publisher tun nichts Besonderes: Sie publishen weiterhin an factory/line-3/telemetry und bekommen das $share-Präfix nie zu sehen. Um zu prüfen, welche Topics ein Filter trifft, füge den Teil nach dem Gruppennamen in den MQTT-Topic-Matcher ein.

Wie Nachrichten verteilt werden

Die Zustellregeln lassen sich am besten an einem Beispiel zeigen. Angenommen, es gibt diese Subscriptions:

Jede Gruppe bekommt also eine Kopie, und innerhalb einer Gruppe verarbeitet sie ein Mitglied. Wie dieses Mitglied ausgewählt wird, legt die Spezifikation nicht fest. Broker bieten Strategien wie Round Robin, Zufall, Sticky (so lange an dasselbe Mitglied senden, bis es verschwindet) oder Hashing über Client-ID oder Topic an – bei EMQX lässt sich das zum Beispiel konfigurieren. Jedes Mitglied behält seine eigene Subscription-QoS, und Nachrichten werden wie gewohnt pro Mitglied bestätigt.

Broker-Unterstützung

Shared Subscriptions sind Teil des MQTT-5-Standards, und ein Broker signalisiert die Unterstützung über die Property Shared Subscription Available im CONNACK. Unterstützt ein Broker sie nicht, lehnt er das Subscribe mit dem Reason Code 0x9E (Shared Subscriptions not supported) ab – siehe die Referenz der MQTT-Reason-Codes.

• Mosquitto unterstützt $share seit Version 1.6, für MQTT-5- und MQTT-3.1.1-Clients.
• EMQX unterstützt $share und zusätzlich eine ältere Kurzform, $queue/<filter>, die sich wie eine Shared Subscription in einer einzigen Standardgruppe verhält.
• HiveMQ unterstützt $share für beide Protokollversionen.

Managed- und Cloud-Broker unterscheiden sich hier, also prüfe die Doku, bevor du dich darauf verlässt. Unser MQTT-Broker-Vergleich behandelt die wichtigsten Optionen.

Beispiel: Zwei Worker teilen sich einen Stream

Öffne drei Terminals. Die ersten beiden treten derselben Gruppe bei; das dritte publisht zehn Nachrichten:

mosquitto_sub -V mqttv5 -h broker.hivemq.com -q 1 \
  -t '$share/workers/testmqtt/share-demo/jobs' -v

for i in $(seq 1 10); do
  mosquitto_pub -V mqttv5 -h broker.hivemq.com -q 1 \
    -t 'testmqtt/share-demo/jobs' -m "job $i"
done

Jeder Job erscheint in genau einem Worker-Terminal. Beachte die einfachen Anführungszeichen um den Filter: In den meisten Shells würde $share in doppelten Anführungszeichen als Variable expandiert. Im Code funktioniert das genauso:

import mqtt from 'mqtt';

const client = mqtt.connect('wss://broker.hivemq.com:8884/mqtt', { protocolVersion: 5 });

client.on('connect', () => {
  client.subscribe('$share/workers/testmqtt/share-demo/jobs', { qos: 1 });
});

client.on('message', (topic, payload) => {
  // topic is the real topic, without the $share prefix
  console.log(topic, payload.toString());
});

Den Publisher kannst du auch im Browser spielen: Öffne den Client unten und publishe an testmqtt/share-demo/jobs, während deine Worker laufen.

Test-Jobs im Online-Client publishen

Anwendungsfälle

• Horizontale Skalierung von Backend-Consumern. Betreibe N Instanzen eines Ingestion-Services und lass den Broker die Telemetrie auf sie verteilen – ohne separate Message Queue.
• Hochverfügbarkeit. Stürzt ein Consumer ab, erhalten die übrigen Mitglieder weiterhin Nachrichten.
• Anbindung an Datenbanken oder Stream-Processing. Mehrere Writer können sich die Last teilen, Nachrichten in eine Time-Series-Datenbank zu schreiben oder an Kafka weiterzuleiten.
• Work Queues. Befehle wie „skaliere dieses Bild“ kann derjenige Worker abarbeiten, der sie gerade erhält.

Ein Setup mit Shared Subscriptions entwerfen

Gruppennamen pro Consumer-Typ wählen

Benenne Gruppen nach der Aufgabe, nicht nach der Instanz: $share/ingest/..., $share/alerts/.... Jede Instanz des Ingest-Services tritt ingest bei, jede Instanz des Alerting-Services alerts. Jeder Service sieht dann als Ganzes jede Nachricht genau einmal, während sich seine Instanzen die Arbeit teilen. Gibst du versehentlich jeder Instanz einen eigenen Gruppennamen, bist du wieder beim einfachen Fan-out, und jede Instanz verarbeitet jede Nachricht.

Sessions und QoS für Worker

Worker abonnieren meist mit QoS 1, damit eine Nachricht erneut zugestellt wird, falls ein Worker abstürzt, bevor er sie bestätigt hat. Ob Nachrichten für eine Gruppe gepuffert werden, während alle Mitglieder offline sind, hängt von deren Sessions ab: Mit einer persistenten Session kann der Broker weiter puffern; mit Clean Sessions werden Nachrichten, die gepublisht werden, während niemand aus der Gruppe verbunden ist, dieser Gruppe einfach nicht zugestellt.

Shared Subscriptions vs. Message Queue

Shared Subscriptions bieten dir einfache Competing Consumers innerhalb von MQTT. Features dedizierter Queueing-Systeme wie Replay ab einem Offset, Dead-Letter-Queues oder Reihenfolgegarantien pro Partition bekommst du damit nicht. Für viele IoT-Backends ist dieser Kompromiss in Ordnung; wenn du Replay brauchst, leite den Stream aus einer Shared Subscription in ein Log wie Kafka weiter und verarbeite ihn dort.

Stolperfallen und Einschränkungen

• Keine globale Reihenfolge. MQTT garantiert die Reihenfolge pro Topic für einen einzelnen Subscriber, aber sobald Nachrichten auf mehrere Worker verteilt werden, laufen sie parallel. Wenn die Reihenfolge pro Gerät wichtig ist, nutze eine Broker-Strategie mit Hashing über das Topic oder partitioniere selbst nach Topic.
• Keine Retained Messages. Laut MQTT-5-Spezifikation werden Retained Messages beim Anlegen einer Shared Subscription nicht gesendet. Nutze für Zustand wie Retained-Status-Topics eine normale Subscription.
• No Local ist nicht erlaubt. Die Option No Local bei einer Shared Subscription zu setzen, ist in MQTT 5 ein Protokollfehler.
• Mit dem vollständigen Filter abbestellen. Um die Gruppe zu verlassen, musst du exakt den String $share/group/filter unsubscriben, nicht nur den einfachen Filter.
• ACLs. Broker unterscheiden sich darin, ob Zugriffsregeln auf den $share/...-String oder auf den zugrunde liegenden Filter angewendet werden. Teste deine Berechtigungen mit der echten Syntax.
• Duplikate sind weiterhin möglich. QoS 1 heißt mindestens einmal, und die erneute Zustellung an ein anderes Mitglied nach einem Disconnect kann Duplikate erzeugen. Baue Handler idempotent; siehe MQTT-QoS-Stufen.

Häufig gestellte Fragen