Mit einem MQTT-Gerätesimulator bauen Sie Dashboards, Speicher-Pipelines und Alarmregeln, bevor auch nur ein echter Sensor installiert ist. In diesem Tutorial schreiben Sie einen Python-Simulator, der Hunderte virtuelle Sensoren mit glaubwürdigem Rauschen, Drift, Fehlern und Aussetzern modelliert und sie an den öffentlichen MQTT-Broker unter broker.connectmq.com veröffentlicht, ohne dessen Rate-Limits zu verletzen.
Warum Sie Geräte simulieren sollten, bevor Hardware da ist
Hardware ist meist der langsamste Teil eines IoT-Projekts. Platinen sind bestellt, die Firmware ist halb fertig und der Installateur war noch nicht vor Ort, während die Leute, die Dashboard und Datenbank bauen, bereits Daten brauchen. Ein Simulator beseitigt diese Abhängigkeit. Er liefert jeder nachgelagerten Komponente einen realistischen Datenstrom und lässt Sie Situationen proben, die sich mit echter Ausrüstung nur schwer provozieren lassen: einen Sensor, der auf einen unmöglichen Wert springt, ein Gerät, das zehn Minuten lang schweigt, oder eine ganze Linie, die auf einmal online geht. In Industrie-4.0-Projekten kommt hinzu, dass Maschinen oft erst bei der Inbetriebnahme im Werk verfügbar sind, MES-Anbindung, OEE-Dashboards und Alarmierung aber schon vorher stehen müssen.
Einen zweiten Vorteil entdecken Teams erst später. Sobald die echte Flotte existiert, wird derselbe Simulator zu Ihrem Regressionswerkzeug. Sie können ihn nach jeder Änderung an Ihrem Ingestion-Code gegen einen Staging-Broker laufen lassen und sofort sehen, ob ein Parser oder Dashboard kaputtgegangen ist.
Was einen nützlichen Simulator von einem Spielzeug unterscheidet, ist Realismus. Eine Schleife, die jede Sekunde random.uniform(20, 25) veröffentlicht, erzeugt Daten, die überhaupt nicht wie die eines echten Temperaturfühlers aussehen. Echte Signale ändern sich langsam, folgen Tageszyklen, tragen Messrauschen, driften über Wochen und fallen gelegentlich aus. Der folgende Simulator modelliert jedes dieser Verhaltensmuster mit wenigen Codezeilen.
Aufbau des MQTT-Gerätesimulators
Legen Sie vor dem Programmieren die Topic-Struktur und das Payload-Format fest, denn alles, was Sie nachgelagert bauen, hängt davon ab. Dieses Tutorial verwendet ein Topic pro Gerät und eine kompakte JSON-Payload:
connectmq/demo/<run-id>/<line>/<device>/telemetry
{"device": "sensor-0042", "ts": 1760600000123, "temperature": 21.84, "humidity": 44.2}
Die <run-id> ist eine zufällige Zeichenfolge, die beim Start erzeugt wird. Der öffentliche MQTT-Broker ist für alle lesbar und für Tests und Prototypen gedacht, daher hält ein eindeutiges Präfix Ihren Datenstrom von den Experimenten anderer getrennt und macht es leicht, genau Ihre Daten zu abonnieren. Die Struktur bleibt bei sechs Topic-Ebenen, deutlich unter der Grenze des öffentlichen MQTT-Brokers von acht.
Der Simulator besteht aus drei Teilen:
- Ein Sensormodell, das den Zustand pro Gerät hält (Basiswert, Drift, Phase des Tageszyklus, Fehler-Flag) und auf Anfrage einen neuen Messwert liefert.
- Ein Scheduler, der entscheidet, welches Gerät als Nächstes veröffentlicht. Die Geräte werden gleichmäßig über das Intervall verteilt, statt alle im selben Moment zu senden.
- Ein Rate-Limiter, der den veröffentlichenden Client unter dem Nachrichtenlimit pro Client des Brokers hält.
Die Rate-Limits des Brokers einhalten
Der öffentliche MQTT-Broker erlaubt 10 Nachrichten pro Sekunde pro Client, 10 Abonnements pro Client und 64 KB pro Nachricht. Hunderte Geräte, die jede Sekunde veröffentlichen, würden das innerhalb von Augenblicken überschreiten, daher muss der Simulator zuerst etwas rechnen. Die Gesamtrate ist einfach die Anzahl der Geräte geteilt durch das Publish-Intervall.
| Virtuelle Geräte | Intervall pro Gerät | Gesamtrate | Passt in einen öffentlichen Client? |
|---|---|---|---|
| 50 | 10 s | 5 msg/s | Ja |
| 200 | 30 s | 6,7 msg/s | Ja |
| 300 | 30 s | 10 msg/s | Nein, am Limit |
| 500 | 60 s | 8,3 msg/s | Ja |
| 1.000 | 60 s | 16,7 msg/s | Nein, auf mehrere Clients aufteilen oder einen kostenpflichtigen Plan nutzen |
Der Code begrenzt sich selbst auf 8 Nachrichten pro Sekunde, um Spielraum für Wiederholungen und für parallel laufende abonnierende Werkzeuge zu lassen. Brauchen Sie eine höhere Rate, haben Sie zwei ehrliche Optionen: Geräte auf mehrere Client-Verbindungen verteilen, jede mit eigener ID und eigenem Budget, oder zu einem privaten Broker wechseln, bei dem die Limits von Ihrem Plan statt vom geteilten öffentlichen Dienst bestimmt werden.
Der vollständige Python-Simulator
Installieren Sie die einzige Abhängigkeit:
pip install "paho-mqtt>=2.0"
Speichern Sie Folgendes als simulator.py. Es verwendet die Callback-API von paho-mqtt 2.x und ansonsten nur die Standardbibliothek.
import argparse
import heapq
import json
import math
import random
import time
import uuid
import paho.mqtt.client as mqtt
class VirtualSensor:
"""Ein simulierter Temperatur- und Feuchtesensor mit persistentem Zustand."""
def __init__(self, index: int, fault_rate: float):
self.device_id = f"sensor-{index:04d}"
self.line = f"line{index % 4 + 1}"
self.base_temp = random.uniform(19.0, 24.0)
self.humidity = random.uniform(35.0, 55.0)
self.phase = random.uniform(0, 2 * math.pi)
self.drift = 0.0
self.faulty = random.random() < fault_rate
def read(self, now: float) -> dict:
# Langsamer Tageszyklus von +/- 2,5 Grad, pro Gerät verschoben
daily = 2.5 * math.sin(2 * math.pi * (now % 86400) / 86400 + self.phase)
# Begrenzter Random Walk, der Kalibrierungsdrift nachahmt
self.drift = max(-3.0, min(3.0, self.drift + random.gauss(0, 0.05)))
# Gaußsches Messrauschen
temp = self.base_temp + daily + self.drift + random.gauss(0, 0.15)
self.humidity = max(10.0, min(95.0, self.humidity + random.gauss(0, 0.3)))
# Fehlerhafte Geräte erzeugen gelegentlich physikalisch unplausible Spitzen
if self.faulty and random.random() < 0.05:
temp += random.choice([-40.0, 60.0])
return {
"device": self.device_id,
"ts": int(now * 1000),
"temperature": round(temp, 2),
"humidity": round(self.humidity, 1),
}
def build_client(args, run_id: str) -> mqtt.Client:
client_id = args.client_id or f"sim-{run_id}"
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
if args.username:
client.username_pw_set(args.username, args.password)
if args.tls:
client.tls_set() # System-CA-Speicher; der Broker nutzt ein Let's-Encrypt-Zertifikat
def on_connect(client, userdata, flags, reason_code, properties):
print(f"connected: {reason_code}")
def on_disconnect(client, userdata, flags, reason_code, properties):
print(f"disconnected: {reason_code}")
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.reconnect_delay_set(min_delay=1, max_delay=30)
return client
def main():
parser = argparse.ArgumentParser(description="MQTT device simulator")
parser.add_argument("--host", default="broker.connectmq.com")
parser.add_argument("--port", type=int, default=1883)
parser.add_argument("--tls", action="store_true")
parser.add_argument("--username")
parser.add_argument("--password")
parser.add_argument("--client-id")
parser.add_argument("--prefix", help="topic prefix, default connectmq/demo/<run-id>")
parser.add_argument("--devices", type=int, default=200)
parser.add_argument("--interval", type=float, default=30.0)
parser.add_argument("--max-rate", type=float, default=8.0)
parser.add_argument("--fault-rate", type=float, default=0.02)
parser.add_argument("--dropout", type=float, default=0.01)
args = parser.parse_args()
run_id = uuid.uuid4().hex[:8]
prefix = args.prefix or f"connectmq/demo/{run_id}"
expected = args.devices / args.interval
if expected > args.max_rate:
raise SystemExit(f"{expected:.1f} msg/s exceeds --max-rate {args.max_rate}")
print(f"publishing to {prefix}/# at about {expected:.1f} msg/s")
sensors = [VirtualSensor(i, args.fault_rate) for i in range(args.devices)]
client = build_client(args, run_id)
client.connect_async(args.host, args.port, keepalive=30)
client.loop_start()
start = time.time()
# Erste Messwerte gleichmäßig über ein Intervall verteilen
schedule = [(start + i * args.interval / args.devices, i) for i in range(args.devices)]
heapq.heapify(schedule)
min_gap = 1.0 / args.max_rate
last_send = 0.0
try:
while True:
due, i = heapq.heappop(schedule)
now = time.time()
wait = max(due - now, last_send + min_gap - now)
if wait > 0:
time.sleep(wait)
sensor = sensors[i]
now = time.time()
if random.random() >= args.dropout and client.is_connected():
topic = f"{prefix}/{sensor.line}/{sensor.device_id}/telemetry"
client.publish(topic, json.dumps(sensor.read(now)), qos=0)
last_send = time.time()
heapq.heappush(schedule, (due + args.interval, i))
except KeyboardInterrupt:
print("stopping")
finally:
client.loop_stop()
client.disconnect()
if __name__ == "__main__":
main()
Starten Sie ihn mit den Standardwerten:
python simulator.py --devices 200 --interval 30
Das Skript gibt das gewählte Präfix aus, zum Beispiel connectmq/demo/3f9a1c2e/#. Öffnen Sie ein zweites Terminal und beobachten Sie den Datenstrom:
mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-3f9a1c2e -t "connectmq/demo/3f9a1c2e/#" -v
So entsteht der Realismus
Jeder Messwert ist die Summe aus vier Komponenten, und jede existiert, um einen bestimmten Teil Ihrer Pipeline zu testen.
- Tageszyklus. Eine Sinuswelle mit 24-Stunden-Periode und zufälliger Phase pro Gerät. Ihre Diagramme bekommen eine sichtbare Form, und Aggregationsabfragen wie Stundenmittelwerte liefern aussagekräftige Schwankungen statt einer flachen Linie.
- Begrenzte Drift. Ein Random Walk, begrenzt auf plus/minus drei Grad. Er ahmt Sensoren nach, die langsam ihre Kalibrierung verlieren, und genau das muss eine Drifterkennung sehen.
- Messrauschen. Gaußsches Rauschen mit kleiner Standardabweichung. Glättung, gleitende Mittelwerte und Totband-Filter ergeben nur Sinn, wenn die Eingabe verrauscht ist.
- Fehler und Aussetzer. Etwa 2 Prozent der Geräte werden als fehlerhaft markiert und senden gelegentlich Spitzen von plus 60 oder minus 40 Grad. Unabhängig davon wird 1 Prozent der geplanten Messwerte einfach übersprungen. Spitzen testen Validierung und Alarmschwellen; Lücken testen die Erkennung veralteter Daten und wie Ihre Diagramme fehlende Daten darstellen.
Da der Zustand in jedem VirtualSensor-Objekt liegt, sind die Werte eines Geräts über die Zeit stetig. Das ist wichtig, wenn Sie etwas testen, das den aktuellen Wert mit dem vorherigen vergleicht, etwa Alarme auf die Änderungsrate.
Häufige Fallstricke und wie Sie sie vermeiden
Stoßweises Veröffentlichen. Wenn jedes Gerät in Sekunde null jedes Intervalls veröffentlicht, sieht der Broker einen Schub von Hunderten Nachrichten, gefolgt von Stille. Echte Flotten sind selten so synchronisiert, und ein Schub trifft die Limits pro Client. Der Scheduler staffelt den ersten Messwert jedes Geräts über das Intervall, was die Rate gleichmäßig hält.
Verwirrung bei der Uhrzeit. Die Payload trägt ihren eigenen Millisekunden-Zeitstempel. Nachgelagerte Werkzeuge wie Telegraf oder eine Python-Bridge können ihn statt der Ankunftszeit nutzen, wodurch Diagramme auch bei verzögerten Nachrichten korrekt bleiben. Stellen Sie sicher, dass sich alle Consumer über die Einheit einig sind: hier Millisekunden, nicht Sekunden.
Wiederverwendete Client-IDs. Starten Sie zwei Kopien des Simulators mit derselben --client-id, trennt der Broker die ältere Session jedes Mal, wenn sich die neuere verbindet, und die beiden bekämpfen sich gegenseitig. Lassen Sie das Skript auf dem öffentlichen MQTT-Broker eine zufällige ID erzeugen, oder verwenden Sie auf einem privaten Broker die für Ihre Zugangsdaten ausgestellte ID.
Sensible Daten an einen öffentlichen MQTT-Broker senden. Alles auf dem öffentlichen MQTT-Broker kann jeder lesen, der das Topic errät oder per Wildcard abonniert. Simulierte Daten sind dort in Ordnung, echte Kunden- oder Anlagendaten nicht.
Veröffentlichen ohne Verbindung. Die Schleife prüft client.is_connected() und überspringt Messwerte, solange die Verbindung unterbrochen ist, genau wie ein echter Sensor ohne Pufferung. Wenn Sie Pufferverhalten testen möchten, lesen Sie den Leitfaden zu paho-MQTT-Reconnects und ergänzen Sie eine Offline-Warteschlange.
Den Simulator auf Ihren eigenen Broker richten
Der öffentliche MQTT-Broker ist ideal für einen schnellen Start, doch für dauerhafte Lasttests mit Prototypen wollen Sie eine isolierte Umgebung. Mit einem verwalteten MQTT-Broker von ConnectMQ erhalten Sie eigene Zugangsdaten und einen TLS-Endpunkt:
python simulator.py \
--host broker.connectmq.com --port 8883 --tls \
--username YOUR_USERNAME --password YOUR_PASSWORD --client-id YOUR_CLIENT_ID \
--prefix connectmq/factory1/production \
--devices 500 --interval 60 --max-rate 9
Port 8883 nutzt TLS mit einem Let's-Encrypt-Zertifikat, daher genügt tls_set() mit dem System-Trust-Store. Arbeiten Sie mit einem Unified Namespace (UNS), setzen Sie --prefix auf den ISA-95-Pfad eines Bereichs, etwa acme/werk-stuttgart/montage; der Simulator hängt Linie und Gerät an, sodass die simulierten Daten genau dort im UNS landen, wo später die echten Maschinen publizieren. Auf einem verwalteten MQTT-Broker sind Client-IDs an die ausgestellten Zugangsdaten gebunden, übergeben Sie also die ID aus dem Portal, statt eine zu erfinden. Der Browser-Client und der Live-Topic-Explorer im Portal sind ein schneller Weg zu prüfen, ob die simulierten Geräte unter den erwarteten Topics erscheinen. Der Plan Serverless (Free) unterstützt bis zu 25 Verbindungen mit QoS 0 oder 1, was genügt, um eine große simulierte Flotte auf mehrere Prozesse aufzuteilen, sofern jeder seine eigenen ausgestellten Zugangsdaten verwendet.
Wohin mit den simulierten Daten?
Ein Simulator wird erst interessant, wenn etwas seine Daten verarbeitet. Naheliegende nächste Schritte:
- Parsen Sie die Payloads in Python, wie unter MQTT-JSON in Python verarbeiten gezeigt, und berechnen Sie Statistiken in pandas.
- Speichern Sie den Datenstrom mit Telegraf und InfluxDB über den TIG-Stack mit Docker Compose und bauen Sie darauf Grafana-Panels.
- Vergleichen Sie dieses Skript in unserer Übersicht der MQTT-Lasttest-Werkzeuge mit spezialisierten Benchmark-Tools, wenn Sie den Broker-Durchsatz messen statt realistische Daten erzeugen wollen.
Einen breiteren Blick auf Python- und andere Client-Optionen bietet der Leitfaden zu MQTT-Client-Bibliotheken, der die wichtigsten Bibliotheken nach Sprache vergleicht, und die paho-mqtt-Dokumentation behandelt jede oben verwendete Client-Option.
Starten Sie mit ConnectMQ
Beginnen Sie mit dem kostenlosen öffentlichen MQTT-Broker unter broker.connectmq.com, um Ihre simulierte Flotte in wenigen Minuten zu sehen. Wenn Sie bereit für private Zugangsdaten, TLS und eigene Kontingente sind, betreibt ConnectMQ einen verwalteten MQTT-Broker in der Cloud. Erstellen Sie unter app.connectmq.com einen kostenlosen Serverless-Broker ohne Kreditkarte, kopieren Sie den generierten Benutzernamen, das Passwort und die Client-ID und richten Sie den Simulator darauf. Wenn Sie Hosting-Optionen noch vergleichen, erklärt unser Leitfaden zum Cloud-MQTT-Broker, worauf Sie achten sollten.
Häufig gestellte Fragen
Wie viele Geräte kann ein MQTT-Gerätesimulator über einen Client veröffentlichen?
Das hängt vom Intervall und vom Limit pro Client des Brokers ab. Halten Sie auf dem öffentlichen MQTT-Broker Geräte geteilt durch Intervall unter 10 Nachrichten pro Sekunde, zum Beispiel 200 Geräte alle 30 Sekunden oder 500 Geräte alle 60 Sekunden.
Sollte der Simulator QoS 0 oder QoS 1 verwenden?
QoS 0 ist für die meisten Telemetrie-Simulationen ausreichend und entspricht dem, was viele echte Sensoren tun. Verwenden Sie QoS 1, wenn Sie gezielt die Verarbeitung von Bestätigungen oder die Duplikaterkennung nachgelagert testen möchten.
Kann ich Geräte simulieren, die offline gehen?
Ja. Erhöhen Sie --dropout für zufällige Lücken, oder ergänzen Sie pro Client eine Last-Will-Nachricht, damit Subscriber einen Offline-Status erhalten, wenn ein Simulatorprozess beendet wird.
Ist es sicher, den Simulator auf dem öffentlichen MQTT-Broker zu betreiben?
Ja, solange die Daten synthetisch sind. Der öffentliche MQTT-Broker ist für jeden lesbar, veröffentlichen Sie dort also niemals echte Kunden-, Anlagen- oder personenbezogene Daten.
