MQTT-Lasttests mit Locust

Locust ist ein Python-Framework für Lasttests mit Live-Web-Dashboard, verteilten Workern und Szenarien, die als normaler Code geschrieben werden. Standardmäßig spricht es HTTP, aber seine User-Klasse kann jedes Protokoll steuern – MQTT-Lasttests mit Locust bedeuten also, paho-mqtt in eine User-Klasse zu packen und jedes Connect und jedes Publish an die Locust-Statistik zu melden.

Dieser Leitfaden liefert ein vollständiges locustfile.py: einen wiederverwendbaren MQTT-Basis-User, Sensor-User, die Telemetrie publizieren, einen Dashboard-User, der die End-to-End-Latenz misst, und die Befehle, um alles im Browser, headless in CI und verteilt über mehrere Maschinen auszuführen.

Warum Locust für MQTT?

Locust ist die richtige Wahl, wenn sich ein „User" wie ein Gerät verhalten soll: verbinden, nach Zeitplan publizieren, auf Nachrichten reagieren, neu verbinden – und du diese Logik in Python schreiben willst. Für eine schnelle Durchsatzobergrenze ist ein CLI-Benchmark einfacher; siehe MQTT-Broker-Lasttests.

So funktioniert die Integration

Jeder Locust-User ist ein leichtgewichtiges Greenlet (Locust läuft auf gevent). Unser MQTT-User:

1. Erstellt pro Locust-User einen paho-mqtt-Client – 1.000 User bedeuten also 1.000 MQTT-Verbindungen.
2. Misst die Verbindungszeit von connect() bis zum CONNACK.
3. Misst die Publish-Zeit von publish() bis zum PUBACK bei QoS 1 (QoS 0 wird gemeldet, sobald es geschrieben ist).
4. Meldet jedes Ergebnis mit environment.events.request.fire(...) – genau das, was der HTTP-Client von Locust intern tut. Dashboard, CSV-Export und Fehler-Tab funktionieren deshalb unverändert.

Das locustfile

import json, os, random, threading, time, uuid
import paho.mqtt.client as mqtt
from locust import User, between, constant, task
from locust.exception import StopUser

HOST = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", 1883))
PREFIX = "testmqtt/locust"

class MqttUser(User):
    abstract = True
    subscriptions: list[tuple[str, int]] = []

    def on_start(self):
        self.client_id = f"locust-{uuid.uuid4().hex[:10]}"
        self.pending, self.lock = {}, threading.Lock()
        self.connected, self.connect_rc = threading.Event(), None
        c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=self.client_id)
        if os.environ.get("MQTT_USER"):
            c.username_pw_set(os.environ["MQTT_USER"], os.environ.get("MQTT_PASS"))
        c.on_connect, c.on_publish, c.on_message = self._on_connect, self._on_publish, self.on_message
        self.client = c
        start = time.perf_counter()
        try:
            c.connect(HOST, PORT, keepalive=60)
            c.loop_start()
            if not self.connected.wait(10):
                raise TimeoutError("no CONNACK within 10 s")
            if self.connect_rc.is_failure:
                raise ConnectionError(f"CONNACK: {self.connect_rc}")
            self._report("CONNECT", "connect", start, 0)
        except Exception as e:
            self._report("CONNECT", "connect", start, 0, e)
            c.loop_stop()
            raise StopUser()

    def on_stop(self):
        with self.lock:
            leftovers, self.pending = list(self.pending.values()), {}
        for start, name, size in leftovers:
            self._report("PUBLISH", name, start, size, TimeoutError("no PUBACK before stop"))
        self.client.disconnect()
        self.client.loop_stop()

    # --- reporting -------------------------------------------------------
    def _report(self, kind, name, start, size, exc=None):
        self.environment.events.request.fire(
            request_type=kind, name=name, response_time=(time.perf_counter() - start) * 1000,
            response_length=size, exception=exc, context={"client_id": self.client_id})

    # --- paho callbacks (run in the network loop) -----------------------
    def _on_connect(self, client, userdata, flags, reason_code, properties):
        self.connect_rc = reason_code
        for topic, qos in self.subscriptions:          # also restores them after a reconnect
            client.subscribe(topic, qos)
        self.connected.set()

    def _on_publish(self, client, userdata, mid, reason_code, properties):
        with self.lock:
            item = self.pending.pop(mid, None)
        if item:
            start, name, size = item
            exc = Exception(f"PUBACK: {reason_code}") if reason_code.is_failure else None
            self._report("PUBLISH", name, start, size, exc)

    def on_message(self, client, userdata, msg):
        pass

    # --- API for tasks ---------------------------------------------------
    def publish(self, topic, payload, qos=1, name=None):
        name, start = name or topic, time.perf_counter()
        with self.lock:
            info = self.client.publish(topic, payload, qos=qos)
            if info.rc == mqtt.MQTT_ERR_SUCCESS and qos > 0:
                self.pending[info.mid] = (start, name, len(payload))
                return
        exc = None if info.rc == mqtt.MQTT_ERR_SUCCESS else Exception(mqtt.error_string(info.rc))
        self._report("PUBLISH", name, start, len(payload), exc)

class Sensor(MqttUser):
    weight = 20
    wait_time = between(4, 6)                          # one reading every ~5 s

    @task(10)
    def telemetry(self):
        payload = json.dumps({"ts": time.time(), "temp": round(random.gauss(21, 0.5), 2)})
        self.publish(f"{PREFIX}/{self.client_id}/telemetry", payload, qos=1, name="telemetry")

    @task(1)
    def heartbeat(self):
        self.publish(f"{PREFIX}/{self.client_id}/status", "ok", qos=0, name="status")

class Dashboard(MqttUser):
    weight = 1
    wait_time = constant(1)
    subscriptions = [(f"{PREFIX}/+/telemetry", 1)]

    def on_message(self, client, userdata, msg):
        sent = json.loads(msg.payload)["ts"]
        self.environment.events.request.fire(
            request_type="RECEIVE", name="telemetry end-to-end",
            response_time=(time.time() - sent) * 1000, response_length=len(msg.payload),
            exception=None, context={})

    @task
    def idle(self):
        pass                                           # messages arrive via on_message

Einige wichtige Details:

• name="telemetry" fasst die Publishes aller Sensoren in einer Zeile zusammen. Ohne das wird das eindeutige Topic jedes Clients zu einer eigenen Zeile in der Statistiktabelle, und das Dashboard ist nicht mehr lesbar.
• Der Lock um publish() und pending verhindert eine Race Condition, bei der ein sehr schnelles PUBACK verarbeitet wird, bevor die Message-ID erfasst wurde.
• Gewichte – mit weight = 20 und weight = 1 startet ein Test mit 210 Usern etwa 200 Sensoren und 10 Dashboards. Jedes Dashboard empfängt die Telemetrie aller Sensoren, die Dashboards testen also auch Fan-out.
• Die End-to-End-Latenz vergleicht time.time() auf zwei Clients. Auf einer Maschine ist das genau; über mehrere Maschinen hinweg müssen die Uhren per NTP synchronisiert sein, sonst enthalten die Zahlen die Uhrabweichung.

Den Test ausführen

pip install locust "paho-mqtt>=2"
export MQTT_HOST=your-broker-host    # your own broker instance, not the shared endpoint
export MQTT_USER=username MQTT_PASS=password
ulimit -n 8192                         # one socket per user
locust -f locustfile.py

Öffne http://localhost:8089, gib die Anzahl der User und die Spawn-Rate (gestartete User pro Sekunde) ein und starte. Der Tab Statistics zeigt eine Zeile pro Name – connect, telemetry, status, telemetry end-to-end – mit Anfragerate, Fehlern und Perzentilen. Charts zeichnet Anfragen pro Sekunde, Antwortzeiten und User-Zahl live; Failures gruppiert Fehler wie The client is not currently connected. oder no CONNACK within 10 s.

Für CI oder lange unbeaufsichtigte Läufe nutzt du den Headless-Modus und bewahrst die Reports auf:

locust -f locustfile.py --headless \
  --users 2000 --spawn-rate 50 --run-time 15m \
  --csv results/mqtt --html results/mqtt-report.html

--spawn-rate ist deine Verbindungsrate. Halte sie bei einem Wert, den deine echte Flotte plausibel erzeugen könnte; 2.000 User mit 50 pro Sekunde sind nach 40 Sekunden hochgefahren. Eine sehr hohe Spawn-Rate macht aus dem Lasttest einen Reconnect-Sturm-Test – nützlich, aber eine andere Fragestellung.

Bevor du hochskalierst, starte mit einer Handvoll Usern und beobachte die Topics im Online-MQTT-Client (testmqtt/locust/#), um zu prüfen, ob Payloads und Topic-Namen stimmen. Solche Smoke-Tests sind auf dem geteilten Endpunkt broker.connectmq.com in Ordnung, er begrenzt aber die Rate – setze MQTT_HOST für den eigentlichen Lasttest deshalb auf deine eigene Broker-Instanz.

Hochskalieren: Worker und Prozesse

Ein Locust-Prozess nutzt einen CPU-Kern. Ist der Lastgenerator ausgelastet, protokolliert Locust CPU usage above 90%! This may constrain your throughput and may even give inconsistent response time measurements! – ab dann spiegeln die Antwortzeiten deinen Laptop wider, nicht den Broker. Füge Prozesse hinzu:

# all cores on one machine (Linux/macOS)
locust -f locustfile.py --processes -1

# or a master plus workers on several machines
locust -f locustfile.py --master
locust -f locustfile.py --worker --master-host 10.0.0.5

Die Worker nutzen dasselbe locustfile; der Master fasst die Statistiken in einem Dashboard zusammen. Verteile die Worker auf mehrere Quell-IPs, wenn du über Zehntausende Verbindungen hinausgehst – Limits für ephemere Ports und Dateideskriptoren auf einem einzelnen Host sind eine häufige falsche Obergrenze.

Locust-Ergebnisse für MQTT lesen

Vergleiche die RPS von PUBLISH telemetry mit RECEIVE geteilt durch die Zahl der Dashboards: Fällt der Empfang hinter das Publizieren zurück, stauen sich Nachrichten oder werden verworfen. QoS verändert den Paketfluss und damit die Zahlen – MQTT-QoS-Level erklärt, warum die QoS-1-Latenz einen vollständigen Roundtrip enthält. Wie du Perzentile strukturiert in eine Kapazitätsentscheidung übersetzt, zeigt MQTT-Benchmark-Ergebnisse interpretieren.

Sinnvolle Varianten

• TLS – rufe c.tls_set() vor connect() auf und setze MQTT_PORT=8883, damit TLS-Handshakes in die Verbindungszeit eingehen.
• MQTT 5 – übergib protocol=mqtt.MQTTv5 an mqtt.Client, um MQTT-5-Broker zu testen; PUBACK-Reason-Codes erscheinen dann als Fehler im Dashboard.
• Lastprofile – eine LoadTestShape-Klasse kann stufenweise hochfahren oder automatisch eine Morgenspitze simulieren.
• Befehle – füge einen Backend-User hinzu, der auf testmqtt/locust/<id>/cmd publiziert, und lass die Sensoren bestätigen, um auch den Rückkanal unter Last zu testen.

Die Stärke von Locust für MQTT: Ein User ist einfach Python – realistisches Geräteverhalten, ein Dashboard, das das ganze Team lesen kann, und dieselbe Datei auf dem Laptop, in CI oder auf einer Flotte von Workern. Fang klein an, prüfe die Topics, fahr dann hoch und lass die Perzentile zeigen, wo die Grenzen liegen.

Häufig gestellte Fragen