MQTT Shared Subscriptions und Load Balancing testen

Eine Shared Subscription ($share/<gruppe>/<topic>) macht aus mehreren Subscribern einen Worker-Pool: Jede Nachricht auf dem Topic geht an genau ein Mitglied der Gruppe statt an alle. Zum Testen startest du mehrere Worker in derselben Gruppe, veröffentlichst eine nummerierte Folge und prüfst drei Dinge: Jede Nachricht kam genau einmal in der Gruppe an, die Last wurde verteilt, und beim Absturz eines Workers mitten im Strom ging nichts verloren.

Dieser Guide zeigt einen schnellen CLI-Check, einen Python-Test mit Absturzsimulation und die Randfälle, die du prüfen solltest. Die Konzepte erklärt MQTT Shared Subscriptions. Hier wird nur getestet.

Was ein korrekter Broker tut

Die Verteilungsstrategie bestimmt der Broker. Mosquitto nutzt Round Robin. Andere Broker bieten zufällige, Sticky- oder Hash-basierte Strategien, oft konfigurierbar. Dein Test sollte faire Verteilung innerhalb einer Toleranz prüfen, kein exaktes Muster, es sei denn, du verlässt dich auf eine bestimmte Strategie.

Schritt 0: Unterstützt der Broker das überhaupt?

Ein MQTT-5-Broker meldet die Unterstützung im CONNACK mit der Property Shared Subscription Available. Fehlt die Property, gilt sie als vorhanden. Ist sie 0, lehnt der Broker $share-Subscriptions ab oder ignoriert sie, und das SUBACK liefert 0x9E Shared Subscriptions not supported. Der eigentliche Beweis ist in jedem Fall der Verteilungstest weiter unten. Die Reason-Code-Suche entschlüsselt jedes SUBACK, das du bekommst.

Der Gruppenname ist genau eine Topic-Ebene: kein /, + oder #. Der Rest ist ein normaler Topic-Filter, Wildcards funktionieren also nach der Gruppe: $share/workers/testmqtt/share/+/jobs. Prüfe Filter mit dem Topic Matcher, bevor du dem Broker die Schuld gibst.

Schneller Check mit der Mosquitto-CLI

Öffne zwei Terminals mit Mitgliedern derselben Gruppe und ein drittes mit einem normalen Subscriber:

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

mosquitto_sub -h broker.connectmq.com -V mqttv5 -q 1 \
  -t 'testmqtt/share/9b1e/jobs' -v

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

Terminal 1 und 2 teilen sich die zehn Jobs (bei Round Robin ungerade Nummern auf der einen, gerade auf der anderen Seite). Terminal 3 sieht alle zehn. Setze das Topic immer in einfache Anführungszeichen: Ohne Quotes oder in doppelten Anführungszeichen würde deine Shell $share als Variable expandieren.

Automatisierter Verteilungs- und Failover-Test

Dieses paho-mqtt-2.x-Skript startet drei Worker in einer Gruppe, veröffentlicht 100 nummerierte QoS-1-Nachrichten und meldet Anzahl pro Worker, Lücken und Duplikate. Mit --failover lässt es worker-0 auf halber Strecke abstürzen: Der Worker hört auf zu lesen, seine letzten Nachrichten bleiben unbestätigt, dann wird die TCP-Verbindung ohne DISCONNECT gekappt, wie bei einem echten Absturz.

import sys, time, uuid, collections
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
RUN = uuid.uuid4().hex[:8]
TOPIC = f"testmqtt/share/{RUN}/jobs"
GROUP = f"workers{RUN}"
WORKERS, MESSAGES, RATE = 3, 100, 10     # RATE in messages per second
FAILOVER = "--failover" in sys.argv          # crash worker-0 halfway through

got = collections.defaultdict(list)          # worker name -> message ids received

def make_worker(name):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{name}-{RUN}", protocol=mqtt.MQTTv5)

    def on_connect(cl, userdata, flags, rc, props):
        if getattr(props, "SharedSubscriptionAvailable", 1) == 0:
            print(f"{name}: CONNACK says shared subscriptions are not available")
        cl.subscribe(f"$share/{GROUP}/{TOPIC}", qos=1)

    def on_subscribe(cl, userdata, mid, reason_codes, props):
        print(f"{name}: SUBACK {reason_codes[0]}")

    c.on_connect, c.on_subscribe = on_connect, on_subscribe
    c.on_message = lambda cl, userdata, msg: got[name].append(int(msg.payload))
    c.connect(HOST, PORT)
    c.loop_start()
    return c

workers = {f"worker-{i}": make_worker(f"worker-{i}") for i in range(WORKERS)}
time.sleep(2)                                 # let all subscriptions settle

pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"share-pub-{RUN}", protocol=mqtt.MQTTv5)
pub.connect(HOST, PORT)
pub.loop_start()
for i in range(MESSAGES):
    if FAILOVER and i == MESSAGES // 2:
        crashed = workers.pop("worker-0")
        crashed.loop_stop()                   # stop reading: unacked messages stay in flight
        crashed.socket().close()              # drop TCP without DISCONNECT, like a crash
        print(f"crashed worker-0 after message {i - 1}")
    pub.publish(TOPIC, str(i), qos=1)
    time.sleep(1 / RATE)
time.sleep(5)

all_ids = [m for ids in got.values() for m in ids]
missing = sorted(set(range(MESSAGES)) - set(all_ids))
dupes = len(all_ids) - len(set(all_ids))
for name in sorted(got):
    print(f"{name}: {len(got[name]):4d} messages, in order: {got[name] == sorted(got[name])}")
print(f"delivered {len(set(all_ids))}/{MESSAGES}, missing {len(missing)}, duplicates {dupes}")
if missing:
    print("first missing ids:", missing[:10])
for c in workers.values():
    c.disconnect(); c.loop_stop()
pub.disconnect(); pub.loop_stop()

Ein Lauf gegen einen Round-Robin-Broker sieht so aus:

worker-0: SUBACK Granted QoS 1
worker-1: SUBACK Granted QoS 1
worker-2: SUBACK Granted QoS 1
crashed worker-0 after message 49
worker-0:   17 messages, in order: True
worker-1:   42 messages, in order: True
worker-2:   41 messages, in order: True
delivered 100/100, missing 0, duplicates 0

So liest du das Ergebnis:

• Ohne --failover sollten die Zahlen etwa gleich sein (33/33/34 bei Round Robin). Eine stark schiefe Verteilung deutet auf eine Sticky- oder Hash-Strategie hin oder auf einen langsamen Worker, den der Broker überspringt.
• Mit --failover heißt missing 0, dass der Broker die Arbeit des abgestürzten Mitglieds umgeleitet hat. Fehlen IDs genau um den Absturzzeitpunkt, wurden In-Flight-Nachrichten mit der Session verworfen. Manche Broker warten zuerst auf einen Reconnect des Clients, teste also mit deinen echten Session-Einstellungen.
• Duplikate sind bei QoS 1 erlaubt: Eine Nachricht kann an den abgestürzten Worker geliefert und an einen anderen erneut gesendet werden. Worker in einer Shared Group müssen idempotent sein. Siehe doppelte MQTT-Nachrichten.
• Fehlende Nachrichten ganz ohne Absturz bedeuten meist ein Rate- oder Queue-Limit auf Broker-Seite. Senke RATE und starte erneut. Geteilte öffentliche Endpoints drosseln besonders häufig.

Randfälle, die du prüfen solltest

1. Reihenfolge. Jedes Mitglied sieht seine eigenen Nachrichten in Reihenfolge, aber über die Gruppe gibt es keine globale Ordnung. Nachricht 41 kann auf einem anderen Worker vor Nachricht 40 verarbeitet werden. Zählt die Reihenfolge pro Gerät, nutze eine Hash-by-Topic-Strategie, wo verfügbar, oder partitioniere die Topics. Der Message-Ordering-Test zeigt, wie man das misst.
2. Retained Messages. Veröffentliche eine Retained Message und starte dann einen Shared Subscriber: Er darf nichts erhalten. Worker, die einen Anfangszustand brauchen, müssen ihn anders beziehen.
3. Gemischtes QoS. Das zugestellte QoS ist das Minimum aus Publish-QoS und dem gewährten QoS jedes Mitglieds. Ein Mitglied mit QoS-0-Subscription verliert beim Absturz Nachrichten, auch wenn die anderen QoS 1 nutzen.
4. Subscription-Optionen. No Local auf einer Shared Subscription ist ein Protokollfehler. Die übrigen Optionen erklärt MQTT Subscription Options.
5. Gruppe skalieren. Füge mitten im Lauf einen vierten Worker hinzu: Er sollte sofort seinen Anteil bekommen. Entferne einen: Die anderen fangen die Last auf.
6. Persistente Mitglieder. Mit Session Expiry größer 0 behält ein getrenntes Mitglied seinen Platz in der Gruppe, und manche Broker schicken ihm weiter Nachrichten, die sich offline aufstauen. Entscheide, ob Worker überhaupt persistent sein sollen.

Den Durchsatzgewinn testen

Shared Subscriptions gibt es, um Consumer zu skalieren. Miss, ob die Skalierung real ist: Führe das Skript mit 1, 2 und 4 Workern aus, die pro Nachricht eine feste Verarbeitungszeit warten, und vergleiche die Gesamtzeit zum Abarbeiten eines Bursts. Der Durchsatz sollte fast linear wachsen, bis Broker oder Netzwerk zum Limit werden. Ein Fan-in-Test ist die natürliche Ergänzung: viele Publisher, die eine Gruppe speisen.

Die Verteilung live beobachten

Öffne den Online-MQTT-Client in zwei Browser-Tabs, abonniere in beiden denselben $share/...-Filter und veröffentliche aus einem dritten Tab. Du siehst die Aufteilung in Echtzeit, ein guter Plausibilitätscheck vor dem Skript.

Test-Jobs aus dem Browser veröffentlichen

Ein Shared-Subscription-Test dauert wenige Minuten und beantwortet die Fragen, die in Produktion zählen: Wird jeder Job verarbeitet, verteilt sich die Last wirklich, und was passiert mit laufender Arbeit, wenn ein Worker stirbt? Nimm den Failover-Lauf in deine CI-Suite auf und wiederhole ihn bei jedem Broker- oder Session-Wechsel.

Häufig gestellte Fragen