MQTT-Fan-out-Test: ein Publisher, tausende Subscriber

Bei Fan-out geht eine Nachricht an viele Empfänger: eine Konfigurationsänderung für die ganze Flotte, ein Preis-Update auf 10.000 Dashboards, eine Ankündigung „Firmware verfügbar“. Der Publisher sendet ein PUBLISH; der Broker sendet eines pro passender Subscription. Ein Fan-out-Test misst, wie lange es dauert, bis der letzte Subscriber die Nachricht hat – und ob überhaupt jeder sie bekommen hat.

Diese Anleitung baut einen Benchmark 1 → 10.000 auf, ergänzt eine Sonde für die Zustellstreuung und rechnet die Bandbreite durch, damit dich die Netzwerkrechnung nicht überrascht. Für allgemeine Ratentests siehe den MQTT-Durchsatztest, für das umgekehrte Muster den Fan-in-Test.

Warum Fan-out ein eigener Test ist

Publish-Durchsatztests koppeln meist jeden Publisher mit einem Subscriber. Fan-out belastet etwas anderes: die Fähigkeit des Brokers, eine Nachricht in tausende ausgehende Queues zu kopieren und diese an tausende Sockets zu leeren.

Die Frage lautet nicht „wie viele Nachrichten pro Sekunde“, sondern „wie lange, bis jeder Subscriber sie hat“ – die Zustellstreuung – bei deiner realen Flottengröße.

Zuerst die Bandbreitenrechnung

Rechne vor dem Start: Ein Fan-out-Test kann eine Netzwerkverbindung auslasten, lange bevor der Broker ins Schwitzen kommt. Ein PUBLISH auf der Leitung ist ungefähr fester Header + 2 Bytes Topic-Länge + Topic + Packet Identifier (QoS > 0) + Properties + Payload. Details unter MQTT-Bandbreiten-Overhead.

Beispiel: 500-Byte-Payload auf testmqtt/fanout/broadcast (25 Bytes) mit QoS 0 ≈ 530 Bytes pro PUBLISH.

Dazu kommt der TCP/IP- und TLS-Overhead. Die letzte Zeile ist mehr, als viele einzelne Lastgenerator-Maschinen empfangen können – deshalb verteilt man Fan-out-Tests meist auf mehrere Maschinen. Der Paketgrößen-Rechner liefert dir die exakte Größe pro Paket für dein Topic und deinen Payload.

Schritt 1: Subscriber öffnen

emqtt-bench kann tausende Subscriber-Verbindungen halten. Verteile 10.000 auf vier Maschinen (oder Container auf verschiedenen Hosts), um nicht an File-Descriptor- und Ephemeral-Port-Limits zu stoßen:

ulimit -n 65535
docker run --rm -it --ulimit nofile=65535:65535 emqx/emqtt-bench sub \
  -h broker.connectmq.com -p 1883 \
  -c 2500 -i 5 \
  -t testmqtt/fanout/broadcast -q 0 \
  -u "$MQTT_USER" -P "$MQTT_PASS"

-i 5 öffnet alle 5 ms eine Verbindung (200 pro Sekunde und Maschine), die vier Maschinen brauchen also gut 12 Sekunden, bis alle verbunden sind. Verbindungen schneller zu öffnen ist ein eigener Test – siehe Verbindungs-Stresstest – und sollte nicht mit Fan-out-Messungen vermischt werden. Warte, bis alle vier die volle Anzahl melden, bevor du publishst.

Schritt 2: Publishen und sondieren

Die Subscriber-Ausgabe von emqtt-bench zeigt die aggregierte Empfangsrate, aber nicht, wie spät der langsamste Subscriber war. Dieses paho-mqtt-2.x-Skript fügt in einem Prozess 50 Sonden-Subscriber hinzu, publisht Nachrichten mit Zeitstempel auf dasselbe Broadcast-Topic und meldet Latenz und Zustellstreuung über die Sonden. Da Publisher und Sonden dieselbe Uhr nutzen, sind die Werte direkt vergleichbar:

import json, threading, time, uuid
from collections import defaultdict
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
TOPIC = "testmqtt/fanout/broadcast"
PROBES, MESSAGES, RATE, QOS = 50, 60, 1, 0
lat = defaultdict(list)          # seq -> Latenzen (ms) über alle Sonden
lock = threading.Lock()

def on_message(client, userdata, msg):
    try:
        m = json.loads(msg.payload)
    except ValueError:
        return                   # Nicht-Sonden-Traffic ignorieren
    with lock:
        lat[m["seq"]].append((time.perf_counter() - m["t"]) * 1000)

def new_client(name):
    return mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                       client_id=f"{name}-{uuid.uuid4().hex[:6]}")

probes = []
for n in range(PROBES):
    c = new_client(f"fanout-probe-{n}")
    c.on_message = on_message
    c.connect(HOST, PORT)
    c.subscribe(TOPIC, qos=QOS)
    c.loop_start()
    probes.append(c)

pub = new_client("fanout-pub")
pub.connect(HOST, PORT)
pub.loop_start()
time.sleep(3)                    # Subscriptions sich setzen lassen

for seq in range(MESSAGES):
    body = {"seq": seq, "t": time.perf_counter(), "pad": "x" * 450}
    pub.publish(TOPIC, json.dumps(body), qos=QOS)
    time.sleep(1 / RATE)
time.sleep(5)

pct = lambda xs, q: xs[min(len(xs) - 1, int(len(xs) * q))]
all_lat = sorted(v for vals in lat.values() for v in vals)
spread = sorted(max(v) - min(v) for v in lat.values() if len(v) > 1)
print(f"delivered {len(all_lat)}/{PROBES * MESSAGES}")
if all_lat:
    print(f"latency p50 {pct(all_lat, .5):.1f}  p99 {pct(all_lat, .99):.1f}  max {all_lat[-1]:.1f} ms")
if spread:
    print(f"spread  p50 {pct(spread, .5):.1f}  max {spread[-1]:.1f} ms")

Lass es dreimal laufen: ohne Bench-Subscriber (Baseline), mit 2.500 und mit allen 10.000. Die Baseline zeigt den Anteil des Netzwerks; der Zuwachs danach sind die Fan-out-Kosten des Brokers. Wiederhole anschließend mit QOS = 1 und höherer RATE, um zu sehen, wo die Latenz zu steigen beginnt.

Die Ergebnisse lesen

• Zugestellt < erwartet bei den Sonden unter QoS 0 heißt, dass der Broker für manche Verbindungen Nachrichten verworfen hat – meist, weil deren ausgehende Queues voll waren. Wiederhole mit QoS 1 und prüfe die Broker-Logs.
• p99-Latenz wächst mit der Subscriber-Zahl, p50 bleibt flach ist bis zu einem gewissen Punkt normal: Der Broker schreibt nacheinander in die Sockets. Entscheidend ist, ob p99 innerhalb deiner Anforderung bleibt.
• Die Streuung steigt bei höheren Publish-Raten stark an heißt, dass der Broker (oder sein Netzwerk) nicht alle Kopien loswird, bevor die nächste Nachricht kommt. Du bist über der nachhaltigen Fan-out-Rate.
• Bench-Maschinen melden niedrigere Empfangsraten als erwartet – prüfe zuerst deren CPU und Netzwerk. 100.000 msg/s zu empfangen ist echte Arbeit für einen Lastgenerator.
• Einige Subscriber fallen zurück und werden getrennt – siehe langsame Consumer. In Produktion darf ein langsames Gerät nicht alle anderen ausbremsen.

Topics für Fan-out entwerfen

Der Test endet meist in einer Designdiskussion. Einige Muster bewähren sich:

1. Broadcast nur, wo wirklich alle es brauchen. fleet/all/config für Nachrichten, die jedes Gerät erhalten muss; halte sie klein.
2. Gruppen-Topics statt eines Riesen-Broadcasts. fleet/region/eu/config oder fleet/model/x200/firmware senken den Fan-out pro Nachricht und erlauben schrittweise Rollouts.
3. Den aktuellen Zustand retainen. Publishe Konfiguration mit dem Retain-Flag, damit Geräte, die sich später verbinden, sofort den letzten Wert bekommen – siehe Retained Messages. Teste, dass ein später Subscriber ihn erhält.
4. QoS nach Konsequenz wählen. Ersetzbare Updates mit QoS 0; Befehle, die ankommen müssen, mit QoS 1 – so dimensioniert, dass das Tracking pro Subscriber im Broker bezahlbar bleibt.
5. Hier keine Shared Subscriptions. Sie verteilen Last; sie broadcasten nicht.

Eine kurze Checkliste

1. Ausgehende Bandbreite für Zielflotte und Publish-Rate berechnen.
2. Subscriber auf genügend Lastmaschinen verteilen; File-Descriptor-Limits erhöhen.
3. Erst alle Subscriber verbinden, dann publishen – Verbindungstests getrennt halten.
4. Latenz und Streuung mit Sonden messen, nicht nur mit aggregierten Empfangsraten.
5. Mit QoS 0 und QoS 1 wiederholen, mit und ohne TLS.
6. CPU, Speicher und ausgehendes Netzwerk des Brokers neben den Client-Werten festhalten.

Fan-out macht aus einer kleinen Nachricht einen großen Schub an Arbeit und Traffic. Rechne zuerst die Bandbreite durch, verbinde die Subscriber vor dem Publishen und beurteile das Ergebnis am langsamsten Subscriber – genau den bemerken deine Nutzer.

Häufig gestellte Fragen