MQTT-Fan-in-Test: tausende Geräte, ein Backend

Fan-in ist die Alltagsform von IoT: Tausende Geräte publishen jeweils ein wenig Telemetrie, und ein Backend abonniert devices/+/telemetry, um alles zu speichern oder zu verarbeiten. Einzeln sind die Geräte leise; zusammen schicken sie alles durch eine Subscription, eine TCP-Verbindung und einen Prozess. Ein Fan-in-Test findet heraus, wie viele Geräte dieser Consumer aufnehmen kann, bevor sich ein Backlog bildet – und wie du ihn skalierst, wenn er es nicht schafft.

Diese Anleitung simuliert eine Flotte mit emqtt-bench, misst Backlog und Verzögerung auf Consumer-Seite mit einem Python-Skript und zeigt, wie Shared Subscriptions die Last verteilen. Das umgekehrte Muster behandelt der Fan-out-Test.

Wo sich Fan-in-Engpässe verstecken

Der Broker kommt mit dem Fan-in-Ingress meist gut zurecht – viele kleine Nachrichten an einen Subscriber zu routen ist billig. Das Problem liegt dahinter:

Die letzten beiden sind Broker-Limits für eine einzelne Session – etwa max_queued_messages und max_inflight_messages bei Mosquitto. Fällt der Consumer zurück, puffert der Broker Nachrichten für ihn, und ist die Queue voll, verwirft er oder trennt die Verbindung, je nach Broker. Unter langsame Consumer siehst du, wie sich verschiedene Broker verhalten.

Schritt 1: die Flotte simulieren

emqtt-bench kann von einer Maschine aus tausende Geräte spielen. Gib jedem Client mit %i (dem Client-Index) ein eigenes Topic, damit die Topic-Struktur der Produktion entspricht. Das Intervall -I gilt pro Client in Millisekunden, 5.000 Geräte mit -I 5000 senden also je eine Nachricht alle 5 Sekunden – insgesamt 1.000 msg/s:

docker run --rm -it --ulimit nofile=65535:65535 emqx/emqtt-bench pub \
  -h broker.connectmq.com -p 1883 \
  -c 5000 -i 5 -I 5000 \
  -t 'testmqtt/fanin/%i/telemetry' -s 200 -q 1 \
  -u "$MQTT_USER" -P "$MQTT_PASS"

Um die Last zu erhöhen, senke -I (5000 → 2000 → 1000 ms ergibt 2.500 und 5.000 msg/s), statt Verbindungen hinzuzufügen: Du testest den Consumer, nicht die Verbindungsanzahl. Prüfe vor dem Start mit dem Topic-Matcher, ob deine Wildcard die Geräte-Topics wirklich trifft – ein Tippfehler im Filter sieht genauso aus wie ein Consumer, der nichts empfängt.

Schritt 2: ein Consumer, der seinen Backlog zeigt

Ein produktionsnaher Consumer sollte nie langsame Arbeit in on_message erledigen: Das blockiert die Netzwerkschleife, die dann den Socket nicht mehr liest und den Backlog in den Broker verschiebt, wo du ihn nicht siehst. Dieser paho-mqtt-2.x-Consumer übergibt jede Nachricht an eine Queue, verarbeitet sie in Worker-Threads mit simulierten Kosten pro Nachricht und gibt jede Sekunde Empfangsrate, Verarbeitungsrate und Backlog aus:

import json, queue, sys, threading, time, uuid
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
FILTER = "testmqtt/fanin/+/telemetry"
SHARED = "--shared" in sys.argv
TOPIC = f"$share/ingest/{FILTER}" if SHARED else FILTER
WORK_MS = 0.5        # simulierte Kosten pro Nachricht (Parsen + DB-Insert)
WORKERS = 4

inbox = queue.Queue(maxsize=100_000)
counts = {"rx": 0, "done": 0}
lags = []

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(TOPIC, qos=1)

def on_message(client, userdata, msg):
    counts["rx"] += 1
    inbox.put(msg.payload)       # nie langsame Arbeit im Netzwerk-Thread

def worker():
    while True:
        payload = inbox.get()
        if payload[:1] == b"{":  # Sonden-Nachrichten mit Zeitstempel
            try:
                lags.append((time.time() - json.loads(payload)["t"]) * 1000)
            except (ValueError, KeyError):
                pass
        time.sleep(WORK_MS / 1000)
        counts["done"] += 1

for _ in range(WORKERS):
    threading.Thread(target=worker, daemon=True).start()

c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                client_id=f"ingest-{uuid.uuid4().hex[:6]}", protocol=mqtt.MQTTv5)
c.on_connect, c.on_message = on_connect, on_message
c.connect(HOST, PORT)
c.loop_start()

while True:
    time.sleep(1)
    l = sorted(lags); lags.clear()
    p99 = f"{l[int(len(l) * 0.99) - 1]:.0f} ms" if l else "-"
    print(f"rx {counts['rx']:>6}/s  done {counts['done']:>6}/s  "
          f"backlog {inbox.qsize():>6}  probe lag p99 {p99}")
    counts.update(rx=0, done=0)

emqtt-bench-Payloads bestehen aus Zufallsbytes und können keinen Zeitstempel tragen. Um die End-to-End-Verzögerung zu messen, starte auf derselben Maschine wie den Consumer (damit beide dieselbe Uhr nutzen) eine kleine Sonde, die JSON mit Zeitstempel in denselben Topic-Baum publisht:

import json, time
import paho.mqtt.client as mqtt

p = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
p.connect("broker.connectmq.com", 1883)
p.loop_start()
while True:
    p.publish("testmqtt/fanin/probe/telemetry", json.dumps({"t": time.time()}), qos=1)
    time.sleep(0.1)

Die Zahlen lesen

Fahre die Flotte auf jeder Stufe und beobachte die Consumer-Ausgabe ein paar Minuten lang:

• rx ≈ done, Backlog nahe null, Verzögerung flach – der Consumer hält mit. Weiter zur nächsten Stufe.
• rx ≈ Ingress, aber done < rx, Backlog wächst – die Netzwerkseite ist in Ordnung, die Verarbeitung ist der Engpass. Mehr Worker, Batch-Schreibvorgänge oder ein schnelleres Ziel.
• rx < Ingress, Backlog klein – die Grenze ist die einzelne Verbindung oder die Session im Broker (Inflight-Fenster, Queue pro Client). Hier hilft eine Shared Subscription.
• Verzögerung steigt stetig – du liegst über der nachhaltigen Rate, auch wenn noch nichts verloren geht; die Queue wird nur länger.
• Consumer unter Last getrennt – der Broker hat sein Limit pro Client erreicht, oder der Keep-Alive ist abgelaufen, weil der Netzwerk-Thread blockiert war.

Schritt 3: mit Shared Subscriptions skalieren

Ist eine einzelne Consumer-Verbindung die Grenze, starte mehrere Consumer in derselben Shared-Subscription-Gruppe. Der Broker stellt dann jede Nachricht nur einem Mitglied von $share/ingest/... zu:

python ingest.py --shared   # Terminal 1
python ingest.py --shared   # Terminal 2 (und so weiter)

Wiederhole die Stufe, die vorher gescheitert ist. Zu prüfen:

1. Die Summe von done über alle Consumer entspricht dem Ingress der Flotte – keine Nachricht verloren.
2. Keine Duplikate – ohne $share bekäme jeder Consumer jede Nachricht, und du würdest alles doppelt verarbeiten. Zähle eindeutige IDs, falls dein Payload welche enthält.
3. Gleichmäßige Verteilung – Broker nutzen unterschiedliche Strategien (Round Robin, zufällig, sticky pro Client). Prüfe, dass die Last ungefähr ausgeglichen ist.
4. Failover – beende einen Consumer während des Laufs; die anderen sollten seinen Anteil übernehmen. Nachrichten, die gerade an den beendeten Consumer unterwegs waren, werden je nach QoS und Broker erneut zugestellt oder gehen verloren.
5. Reihenfolge – die Reihenfolge pro Gerät ist über die Gruppe hinweg nicht garantiert. Wenn sie zählt, plane dafür.

Shared Subscriptions erklärt die Semantik, der Shared-Subscription-Test behandelt Verteilungs- und Failover-Tests ausführlicher.

Über einen Prozess hinaus

Ist das Ziel der Engpass und sind Bursts unvermeidlich – eine Flotte, die sich nach einem Ausfall neu verbindet, schickt viele gepufferte Daten auf einmal –, setze einen dauerhaften Puffer zwischen MQTT und Speicher. MQTT nach Kafka zu bridgen ist ein verbreitetes Muster: siehe MQTT zu Kafka. Teste dann gezielt den Burst-Fall: Stoppe den Consumer fünf Minuten, während die Flotte weiterläuft, starte ihn neu und miss, wie lange der Backlog zum Abbauen braucht. Für realistischeres Geräteverhalten als bei emqtt-bench nutze einen Gerätesimulator mit deinen echten Payloads.

Vor einem großen Lauf bestätigt ein kurzer Check im Online-MQTT-Client – testmqtt/fanin/# abonnieren und zusehen, wie Nachrichten eintreffen –, dass die Flotte dorthin publisht, wo du es erwartest.

Fan-in bringt selten den Broker zu Fall, sondern den Consumer. Miss Backlog und Verzögerung statt roher Empfangsraten, halte langsame Arbeit vom Netzwerk-Thread fern und skaliere mit Shared Subscriptions, sobald eine Verbindung nicht mehr reicht.

Häufig gestellte Fragen