MQTT-Lasttest-Tools: emqtt-bench, MQTTX Bench und Locust im Vergleich

MQTT-Lasttest-Tools zeigen Ihnen, ob Ihr System standhält, wenn sich Tausende Geräte gleichzeitig verbinden, ein Firmware-Fehler ein Topic flutet oder ein Backend-Subscriber nicht hinterherkommt. Dieser Leitfaden vergleicht emqtt-bench, MQTTX Bench und Locust, zeigt die Befehle für jedes Tool und ergänzt eine kleine Python-Sonde, die End-to-End-Latenz und Nachrichtenverlust misst. Außerdem erfahren Sie, welche Metriken zählen und wie Sie verantwortungsvoll testen – immer gegen Ihren eigenen privaten Broker, niemals gegen einen geteilten öffentlichen.

Erste Regel: Testen Sie Ihren eigenen Broker

Aus den verfügbaren MQTT-Lasttest-Tools zu wählen ist der einfache Teil; entscheidend ist, sie auf das richtige Ziel zu richten. Lasttests erzeugen Verkehr, der wie ein Angriff aussieht. Führen Sie sie nur gegen Infrastruktur aus, die Ihnen gehört oder die Sie ausdrücklich testen dürfen.

• Keine Lasttests gegen den öffentlichen MQTT-Broker. broker.connectmq.com ist ein geteilter Testdienst mit strengen Limits pro Client (10 Nachrichten pro Sekunde und Client, 10 Subscriptions pro Client, Nachrichten bis 64 KB). Ein Lasttest dort misst diese Limits, nicht Ihr System, und verschlechtert den Dienst für alle anderen.
• Einen privaten, für den Test dimensionierten Broker verwenden. Legen Sie bei ConnectMQ einen eigenen Broker für Lasttests an, damit produktive Geräte nicht beeinträchtigt werden.
• Die Kontingente Ihres Plans vor dem Start kennen. Der kostenlose Serverless-Plan erlaubt bis zu 25 Verbindungen und QoS bis 1; der kostenpflichtige Serverless-Plan bis zu 1.000 Verbindungen, QoS 2 und Nachrichten bis 1 MB; Dedicated läuft auf einem isolierten Cluster. Eine Kontingentüberschreitung ist kein Broker-Ausfall: MQTT-5-Clients erhalten den Reason Code 0x97 (Quota exceeded), den Ihr Test getrennt von echten Fehlern erfassen sollte.
• Regeln für Client IDs prüfen. ConnectMQ bindet Client IDs an die ausgestellten Zugangsdaten, während Lasttest-Tools IDs aus einem Präfix und einem Zähler erzeugen. Prüfen Sie im Portal, welche Client IDs Ihre Test-Zugangsdaten zulassen, und konfigurieren Sie die Client-ID-Option des Tools entsprechend – sonst wird jede Verbindung als nicht autorisiert abgewiesen.
• Schutzmechanismen beachten. Verwaltete MQTT-Broker wenden Flood Protection an und sperren IPs nach wiederholten fehlgeschlagenen Logins. Ein falsch konfigurierter Test mit falschen Zugangsdaten kann die IP Ihrer Testmaschine sperren lassen; validieren Sie die Zugangsdaten daher zuerst mit einem einzelnen Client.

Was Sie messen sollten

Durchsatz allein ist eine irreführende Zahl. Erfassen Sie Folgendes und legen Sie die Bestehenskriterien fest, bevor Sie irgendetwas starten:

Behalten Sie immer den Lastgenerator im Blick. Liegt seine CPU bei 100 Prozent, spiegeln die Latenzwerte Ihren Laptop wider, nicht den Broker. Verteilen Sie große Tests auf mehrere Maschinen.

emqtt-bench

emqtt-bench ist ein schlankes Erlang-Tool aus dem EMQX-Projekt, das darauf ausgelegt ist, viele Verbindungen mit sehr wenig Speicher pro Client zu öffnen. Es hat drei Modi: conn, pub und sub.

# 500 Verbindungen hochfahren, eine alle 10 ms
emqtt_bench conn -h YOUR_BROKER_HOST -p 1883 -c 500 -i 10 \
  -u YOUR_USERNAME -P YOUR_PASSWORD

# 50 Subscriber auf eine Wildcard, QoS 1
emqtt_bench sub -h YOUR_BROKER_HOST -p 1883 -c 50 -i 10 \
  -t 'loadtest/#' -q 1 -u YOUR_USERNAME -P YOUR_PASSWORD

# 200 Publisher, die jeweils alle 1000 ms eine 256-Byte-Nachricht an ihr eigenes Topic senden
emqtt_bench pub -h YOUR_BROKER_HOST -p 1883 -c 200 -i 10 -I 1000 \
  -t 'loadtest/device/%i' -s 256 -q 1 -u YOUR_USERNAME -P YOUR_PASSWORD

Wichtige Flags: -c Anzahl der Clients, -i Intervall zwischen neuen Verbindungen in Millisekunden, -I Intervall zwischen Publishes pro Client, -t Topic mit den Platzhaltern %i (Sequenznummer des Clients) und %c (Client ID), -s Payload-Größe, -q QoS. Ergänzen Sie --ssl und Port 8883, um TLS zu testen – was Sie tun sollten, denn TLS-Handshakes sind bei Reconnect-Stürmen oft die eigentlichen Kosten. Das Tool gibt alle paar Sekunden Raten aus.

Stärken: sehr effizient pro Verbindung, ideal für Tests der Verbindungsanzahl und von Reconnect-Stürmen. Schwächen: Die Payloads sind synthetisch und die Latenzmessung ist begrenzt; kombinieren Sie es daher mit einer Sonde (siehe unten).

MQTTX Bench

MQTTX CLI enthält einen bench-Befehl mit denselben drei Modi und freundlicherer Ausgabe. Nutzt Ihr Team bereits MQTTX, ist das der einfachste Einstieg.

mqttx bench conn -c 200 -i 10 -h YOUR_BROKER_HOST -p 8883 -l mqtts \
  -u YOUR_USERNAME -P YOUR_PASSWORD

mqttx bench sub -c 20 -t 'loadtest/#' -q 1 -h YOUR_BROKER_HOST -p 8883 -l mqtts \
  -u YOUR_USERNAME -P YOUR_PASSWORD

mqttx bench pub -c 100 -i 10 -im 1000 -t 'loadtest/device/%i' -q 1 \
  -m '{"temp": 21.5}' -h YOUR_BROKER_HOST -p 8883 -l mqtts \
  -u YOUR_USERNAME -P YOUR_PASSWORD

Hier ist -i das Verbindungsintervall und -im das Nachrichtenintervall pro Client. Da das Tool auf Node.js läuft, kostet jeder Client mehr Speicher als bei emqtt-bench; sehr große Verbindungszahlen brauchen also mehr Generator-Maschinen. Prüfen Sie mqttx bench pub --help in Ihrer Version, da sich die Optionen weiterentwickeln.

Locust

Locust ist ein Python-Framework für Lasttests. Es unterstützt MQTT nicht nativ, aber Sie können eine User-Klasse um paho-mqtt herum schreiben und die Ergebnisse an die Web-Oberfläche von Locust melden. Das macht es zur flexibelsten Option: Realistische Payloads, Verhalten pro Gerät, Ramp-Profile und Assertions sind schlicht Python.

# locustfile.py  (pip install locust paho-mqtt)
import json
import os
import random
import time
import uuid

import paho.mqtt.client as mqtt
from locust import User, between, events, task

HOST = os.environ.get("MQTT_HOST", "YOUR_BROKER_HOST")
USER = os.environ.get("MQTT_USER", "YOUR_USERNAME")
PASS = os.environ.get("MQTT_PASS", "YOUR_PASSWORD")


class MqttDevice(User):
    wait_time = between(1, 2)

    def on_start(self):
        self.device_id = uuid.uuid4().hex[:8]
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                                  client_id=f"lt-{self.device_id}",
                                  protocol=mqtt.MQTTv5)
        self.client.username_pw_set(USER, PASS)
        self.client.tls_set()
        start = time.perf_counter()
        try:
            self.client.connect(HOST, 8883, keepalive=30)
            self.client.loop_start()
            events.request.fire(request_type="MQTT", name="connect",
                                response_time=(time.perf_counter() - start) * 1000,
                                response_length=0, exception=None, context={})
        except Exception as exc:
            events.request.fire(request_type="MQTT", name="connect",
                                response_time=(time.perf_counter() - start) * 1000,
                                response_length=0, exception=exc, context={})
            raise

    @task
    def publish_telemetry(self):
        payload = json.dumps({"id": self.device_id,
                              "temp": round(random.uniform(18, 26), 2),
                              "ts": time.time()})
        start = time.perf_counter()
        info = self.client.publish(f"loadtest/device/{self.device_id}", payload, qos=1)
        exc = None
        try:
            info.wait_for_publish(timeout=5)  # wartet bei QoS 1 auf PUBACK
            if not info.is_published():
                exc = TimeoutError("no PUBACK within 5 s")
        except Exception as e:
            exc = e
        events.request.fire(request_type="MQTT", name="publish qos1",
                            response_time=(time.perf_counter() - start) * 1000,
                            response_length=len(payload), exception=exc, context={})

    def on_stop(self):
        self.client.loop_stop()
        self.client.disconnect()

Starten Sie es mit locust -f locustfile.py --headless -u 100 -r 5 --run-time 10m, wobei -u die Anzahl simulierter Geräte und -r die Spawn-Rate pro Sekunde ist. Jedes Gerät betreibt seinen eigenen paho-Netzwerk-Thread, weshalb Locust pro Verbindung schwerer ist als emqtt-bench; nutzen Sie es für verhaltensreiche Tests mit Hunderten Geräten statt für reine Verbindungszahlen. Die Publish-Zeitmessung erfasst hier den PUBACK-Roundtrip – ein guter Indikator für die Reaktionsfähigkeit des Brokers.

Eine Latenz- und Verlustsonde in Python

Keines der obigen Tools liefert standardmäßig verlässliche End-to-End-Latenz- und Verlustwerte. Lassen Sie diese Sonde parallel zu jedem Lasttest laufen: Sie veröffentlicht nummerierte, mit Zeitstempel versehene Nachrichten und misst, was ankommt.

# probe.py  (pip install paho-mqtt)
import json, os, statistics, time, uuid
import paho.mqtt.client as mqtt

HOST, PORT = os.environ.get("MQTT_HOST", "YOUR_BROKER_HOST"), 8883
USER, PASS = os.environ.get("MQTT_USER", "YOUR_USERNAME"), os.environ.get("MQTT_PASS", "YOUR_PASSWORD")
TOPIC = f"loadtest/probe/{uuid.uuid4().hex[:6]}"
COUNT, INTERVAL = 600, 0.5

latencies, seen = [], set()

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    latencies.append((time.time() - data["ts"]) * 1000)
    seen.add(data["seq"])

def make(name):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"probe-{name}-{uuid.uuid4().hex[:6]}")
    c.username_pw_set(USER, PASS)
    c.tls_set()
    c.connect(HOST, PORT, keepalive=30)
    c.loop_start()
    return c

sub = make("sub")
sub.on_message = on_message
sub.subscribe(TOPIC, qos=1)
time.sleep(1)
pub = make("pub")

for seq in range(COUNT):
    pub.publish(TOPIC, json.dumps({"seq": seq, "ts": time.time()}), qos=1)
    time.sleep(INTERVAL)

time.sleep(5)
q = statistics.quantiles(latencies, n=100)
print(f"received {len(seen)}/{COUNT}, lost {COUNT - len(seen)}")
print(f"latency ms p50={q[49]:.1f} p95={q[94]:.1f} p99={q[98]:.1f} max={max(latencies):.1f}")

Publisher und Subscriber laufen auf derselben Maschine, sodass Uhrenabweichungen die Latenz nicht verfälschen. Führen Sie die Sonde einmal auf einem unbelasteten Broker als Baseline aus und dann während jeder Laststufe; die Differenz ist das, was die Last kostet. Lassen Ihre Zugangsdaten nur bestimmte Client IDs zu, ersetzen Sie die generierten IDs durch Ihre ausgestellten.

Vergleich im Überblick

Ein verantwortungsvoller Testplan

1. Ziel formulieren: zum Beispiel „2.000 Geräte verbinden sich binnen 2 Minuten neu; die p95-Latenz bleibt unter einem vereinbarten Schwellenwert“. In der Fertigung ist das typische Szenario ein Netzausfall im Werk, nach dem alle Edge-Gateways und SPS-Anbindungen einer Halle gleichzeitig zurückkehren.
2. Mit einem Client validieren: Zugangsdaten, TLS, Topic-Struktur, Regeln für Client IDs. Nutzen Sie dieselbe Topic-Hierarchie wie in der Produktion, bei Industrie-4.0-Projekten also meist den Unified Namespace (UNS) nach ISA-95, etwa acme/werk-muenchen/montage/linie-1/zelle-3/..., damit Wildcard-Subscriptions realistisch belastet werden.
3. Baseline: die Sonde auf einem unbelasteten Broker ausführen.
4. Stufenweise steigern: 10, 25, 50, 100 Prozent des Ziels, jede Stufe mehrere Minuten halten.
5. Die unangenehmen Fälle testen: alle verbinden sich gleichzeitig neu, ein Subscriber hört auf zu lesen, übergroße Payloads.
6. Kontingent-Antworten getrennt erfassen: 0x97 bedeutet, dass Sie ein Planlimit erreicht haben – eine Information, kein Absturz.
7. Aufräumen: Retained-Testnachrichten löschen und alle Clients trennen.

Zeigen die Ergebnisse, dass Sie mehr Verbindungen oder Isolation brauchen, als Ihr aktueller Plan bietet, ist das das Signal für eine höhere Stufe, etwa von Serverless zu Dedicated. Die Übersicht zum verwalteten MQTT-Broker und der Leitfaden zur Auswahl eines Cloud-MQTT-Brokers erläutern die Abwägungen. Für Funktionstests vor dem Lasttest lesen Sie MQTT-Clients testen, und für realistischen Verkehr in geringem Umfang nutzen Sie einen MQTT-Gerätesimulator. Die Grundlagen der Kommandozeile behandelt MQTT-CLI-Tools, und der Leitfaden zu MQTT-Client-Bibliotheken hilft, wenn Sie einen eigenen Lastgenerator in einer anderen Sprache bauen möchten.

Starten Sie mit ConnectMQ

Erstellen Sie unter app.connectmq.com einen eigenen Broker für Ihre Tests im kostenlosen Serverless-Plan, ganz ohne Kreditkarte, und validieren Sie Ihre Testskripte in kleinem Maßstab. Wenn Sie für größere Läufe bereit sind, rüsten Sie dasselbe Broker-Setup auf und behalten Sie das Dashboard des Portals offen, um die Verbindungen während des Tests zu beobachten.

Häufig gestellte Fragen

Darf ich den öffentlichen MQTT-Broker einem Lasttest unterziehen?

Nein. Öffentliche Broker sind geteilt und pro Client ratenbegrenzt; die Ergebnisse sind daher bedeutungslos, und andere Nutzer werden beeinträchtigt. Verwenden Sie Ihren eigenen privaten Broker.

Welches MQTT-Lasttest-Tool ist das beste?

emqtt-bench für große Verbindungszahlen, MQTTX Bench für schnelle, gut lesbare Tests und Locust, wenn Sie realistisches Geräteverhalten brauchen. Viele Teams nutzen emqtt-bench für die Skalierung plus eine kleine Sonde für die Latenz.

Warum werden meine Lasttest-Clients mit „Quota exceeded“ abgewiesen?

Sie haben ein Limit Ihres Plans erreicht, etwa die maximale Anzahl an Verbindungen. MQTT-5-Clients erhalten den Reason Code 0x97; reduzieren Sie die Last oder wechseln Sie in einen Plan mit höheren Limits.

Sollten Lasttests TLS verwenden?

Ja, wenn die Produktion TLS nutzt. Handshakes verursachen bei Reconnect-Stürmen auf beiden Seiten erheblichen CPU-Aufwand, und Tests ohne TLS verschleiern diese Kosten.