MQTT-Client-Performance optimieren: Inflight, Batching und Buffer

Die meisten „MQTT ist langsam“-Probleme liegen nicht am Broker. Es ist ein Publisher, der auf jede Bestätigung wartet, ein Inflight-Fenster von 20 auf einer 100-ms-Strecke oder ein Subscriber-Callback, der in eine Datenbank schreibt, bevor er zurückkehrt. MQTT-Client-Performance-Tuning läuft auf eine Handvoll Einstellungen hinaus: Max Inflight, die Art zu publizieren, Batching, Queue-Größen, Keep-Alive und der TCP-Socket selbst.

Dieser Leitfaden erklärt jede Stellschraube, zeigt die Einstellung in gängigen Bibliotheken und liefert ein Skript, mit dem du den Effekt vorher und nachher misst.

Warum die Round-Trip-Time die Obergrenze setzt

Bei QoS 1 braucht jede Nachricht ein PUBACK. Erlaubt der Client nur N unbestätigte Nachrichten, ist der Durchsatz pro Verbindung grob begrenzt auf:

max. Nachrichten/s ≈ Inflight-Fenster ÷ Round-Trip-Time

Bei 20 Inflight-Nachrichten und 50 ms Round Trip sind das etwa 400 Nachrichten pro Sekunde — egal, wie schnell der Broker ist. Erhöhst du das Fenster auf 200, steigt die Obergrenze auf etwa 4.000. Im LAN mit 1 ms Round Trip ist das Standardfenster selten das Limit; über Regionen hinweg oder im Mobilfunk fast immer.

QoS 0 kennt keine Bestätigungen, das Inflight-Fenster greift dort nicht — die Grenzen sind Socket-Buffer, CPU und die Limits pro Client auf dem Broker. QoS 2 braucht zwei Round Trips pro Nachricht und erreicht die Obergrenze früher. Siehe QoS 0 vs 1 vs 2 im Benchmark für die Abwägung.

Die Stellschrauben im Überblick

Das letzte Wort hat der Broker. Bei MQTT 5 teilt das Receive Maximum des Brokers im CONNACK dem Client mit, wie viele QoS-1/2-Nachrichten er gleichzeitig unterwegs haben darf; ein konformer Client darf diesen Wert nicht überschreiten, egal was lokal eingestellt ist. Ändert eine höhere Client-Einstellung nichts, prüfe diesen Wert — Flow Control und Receive Maximum erklärt, wie du ihn ausliest.

1. Max Inflight erhöhen — mit Vorsicht

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="testmqtt-tuned-pub")
client.max_inflight_messages_set(100)   # Standard 20
client.max_queued_messages_set(10000)   # Speicher begrenzen statt unbegrenzt

MqttConnectOptions opts = new MqttConnectOptions();
opts.setMaxInflight(100);               // Standard 10
opts.setKeepAliveInterval(30);

Ein größeres Fenster ist nicht umsonst. Jede Inflight-Nachricht bleibt bis zur Bestätigung im Speicher des Clients, und nach einem Reconnect werden alle erneut gesendet — ein großes Fenster macht aus einem kurzen Netzaussetzer einen Schwall doppelter Nachrichten. Erhöhe es schrittweise und hör auf, sobald der Durchsatz nicht mehr steigt.

2. Asynchron publizieren

Der größte Fehler ist, nach jedem Publish auf dessen Abschluss zu warten, bevor der nächste gesendet wird:

for reading in readings:
    info = client.publish("testmqtt/tuning/data", reading, qos=1)
    info.wait_for_publish()             # serialisiert alles auf die RTT

publish() in paho-mqtt reiht die Nachricht bereits ein und kehrt sofort zurück. Lass es dabei und verfolge den Abschluss in on_publish oder sammle die MQTTMessageInfo-Objekte und prüfe sie am Ende.

In MQTT.js 5 liefert publishAsync ein Promise. Jedes einzeln in einer Schleife abzuwarten hat dasselbe Problem; alle auf einmal abzufeuern hat gar keine Grenze. Ein kleines Fenster liefert Tempo und Back-Pressure zugleich:

import mqtt from "mqtt";

const client = await mqtt.connectAsync("mqtt://broker.connectmq.com:1883", {
  clientId: `testmqtt-tune-${process.pid}`,
  keepalive: 30,
});

const WINDOW = 100;                      // eigenes Inflight-Limit
const N = 5000;
const payload = Buffer.alloc(256, "x");
const pending = new Set();
const started = performance.now();

for (let i = 0; i < N; i++) {
  const p = client.publishAsync("testmqtt/tuning/js", payload, { qos: 1 })
    .finally(() => pending.delete(p));
  pending.add(p);
  if (pending.size >= WINDOW) await Promise.race(pending);
}
await Promise.all(pending);

const secs = (performance.now() - started) / 1000;
console.log(`${N} msgs in ${secs.toFixed(2)} s = ${(N / secs).toFixed(0)} msg/s`);
await client.endAsync();

3. Kleine Nachrichten bündeln

Jede MQTT-Nachricht trägt einen Fixed Header, den Topic-Namen und — bei QoS 1 — ein 4-Byte-PUBACK auf dem Rückweg. Bei einem 20-Byte-Messwert kann der Topic größer sein als die Daten. Batching schickt eine Nachricht mit vielen Messwerten:

import json, time

buffer, last_flush = [], time.monotonic()

def add_reading(client, value):
    global last_flush
    buffer.append({"ts": time.time(), "v": value})
    if len(buffer) >= 50 or time.monotonic() - last_flush >= 1.0:
        client.publish("testmqtt/tuning/batch", json.dumps(buffer), qos=1)
        buffer.clear()
        last_flush = time.monotonic()

Leere den Batch nach Größe oder Zeit, damit auch ein ruhiger Sensor innerhalb einer Sekunde liefert. Die Kompromisse sind echt: Die Latenz wächst bis zum Flush-Intervall, ein verlorener Batch nimmt viele Messwerte mit, und Consumer müssen Arrays verstehen. Mit dem Paketgrößen-Rechner siehst du, wie viel Overhead du tatsächlich sparst; halte Batches deutlich unter der maximalen Paketgröße des Brokers. Mit MQTT 5 sind Topic Aliases eine leichtere Alternative, die lange Topic-Namen verkürzt, ohne das Payload-Format zu ändern.

4. Buffer und Queues bewusst dimensionieren

Eine unbegrenzte Ausgangs-Queue versteckt Probleme: Ist das Netz langsamer als dein Producer, wächst der Speicher, bis der Prozess beendet wird. Eine begrenzte Queue macht daraus einen expliziten Fehler, den du behandeln kannst. In paho-mqtt hat das von publish() gelieferte MQTTMessageInfo nach Erreichen von max_queued_messages_set(n) den Wert rc == MQTT_ERR_QUEUE_SIZE, und nichts wird eingereiht — logge das und entscheide, ob du verwirfst, samplest oder drosselst.

Auf der Empfangsseite entspricht dem die Verarbeitungsgeschwindigkeit des Subscribers. Callbacks laufen im Netzwerk-Thread oder Event Loop des Clients; solange dein on_message beschäftigt ist, liest der Client nicht vom Socket, sendet keine PUBACKs und beantwortet kein Keep-Alive. Gib die Arbeit an eine Queue ab und kehre sofort zurück:

import queue, threading

work = queue.Queue(maxsize=10000)

def on_message(client, userdata, msg):
    work.put_nowait((msg.topic, msg.payload))   # hier niemals blockieren

def worker():
    while True:
        topic, payload = work.get()
        save_to_database(topic, payload)        # langsame Arbeit außerhalb des Netzwerk-Threads

threading.Thread(target=worker, daemon=True).start()

Läuft die Queue trotzdem voll, ist der Consumer wirklich zu langsam — skaliere ihn mit Shared Subscriptions oder lies langsame Consumer. Im Go-Client von Paho lässt SetOrderMatters(false) Message-Handler parallel statt nacheinander laufen, auf Kosten der Reihenfolgegarantie zwischen den Handlern.

5. Keep-Alive und TCP_NODELAY

Keep-Alive macht nichts schneller. Es bestimmt, wie schnell eine tote Verbindung bemerkt wird: Der Broker trennt einen Client nach etwa dem 1,5-Fachen des Keep-Alive-Intervalls ohne Traffic. 30–60 Sekunden passen für die meisten verbundenen Geräte; bei batteriebetriebenen Verbindungen länger. Die meisten Bibliotheken lassen PINGREQ aus, solange andere Pakete fließen. Details unter MQTT Keep Alive.

Der Nagle-Algorithmus kann kleine Nachrichten verzögern: Das Betriebssystem hält ein kleines Segment zurück, solange frühere Daten unbestätigt sind, und zusammen mit Delayed ACKs auf der Gegenseite entstehen Latenzspitzen von einigen zehn Millisekunden. Mit TCP_NODELAY wird jedes Paket sofort gesendet. Ob deine Bibliothek das bereits setzt, hängt von Bibliothek und Version ab; es selbst zu setzen schadet nicht. In paho-mqtt geht das im Callback on_socket_open:

import socket

def on_socket_open(client, userdata, sock):
    sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

client.on_socket_open = on_socket_open

Vorher und nachher messen

Tuning ohne Messung ist Raten. Dieses Skript publiziert eine feste Anzahl QoS-1-Nachrichten mit einem vorgegebenen Inflight-Fenster und meldet die Rate, mit der sie bestätigt wurden:

import os, sys, threading, time
import paho.mqtt.client as mqtt

N = 5000
INFLIGHT = int(sys.argv[1]) if len(sys.argv) > 1 else 20
acked, done = 0, threading.Event()

def on_publish(client, userdata, mid, reason_code, properties):
    global acked
    acked += 1
    if acked == N:
        done.set()

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"testmqtt-bench-{os.getpid()}")
client.max_inflight_messages_set(INFLIGHT)
client.on_publish = on_publish
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_start()
time.sleep(1)

start = time.perf_counter()
for _ in range(N):
    client.publish("testmqtt/tuning/bench", b"x" * 256, qos=1)
done.wait(timeout=120)
secs = time.perf_counter() - start
print(f"inflight={INFLIGHT}: {acked}/{N} bestätigt in {secs:.2f} s = {acked / secs:.0f} msg/s")
client.loop_stop()
client.disconnect()

Starte es mit python inflight_bench.py 10, dann mit 20, 50, 100, 200. Die Rate sollte steigen und dann abflachen; der Knick der Kurve ist das richtige Fenster für diesen Netzwerkpfad. Wiederhole den Test von dort, wo deine Geräte wirklich stehen — ein Ergebnis direkt neben dem Broker sagt wenig über ein Gerät auf einem anderen Kontinent. Öffne währenddessen den Online-Client und abonniere testmqtt/tuning/#, um zu sehen, dass die Nachrichten ankommen.

Für größere Läufe mit vielen Clients nimm spezialisierte Werkzeuge — der Leitfaden zum Durchsatztest stellt sie vor.

Optimiere in dieser Reihenfolge: nicht mehr auf jeden Publish warten, das Inflight-Fenster erhöhen, bis die Kurve abflacht, nur bündeln, wenn der Overhead pro Nachricht weiterhin dominiert, Queues begrenzen und Callbacks schnell halten. Jeder Schritt ist ein bis zwei Zeilen Code, und das Benchmark-Skript zeigt dir, ob er sich gelohnt hat.

Häufig gestellte Fragen