Garantiert MQTT die Reihenfolge von Nachrichten? Teste es selbst

MQTT garantiert die Reihenfolge von Nachrichten nur in einem engen Rahmen: Nachrichten eines Publishers auf einem Topic mit einer QoS-Stufe werden jedem Subscriber in der Reihenfolge zugestellt, in der der Broker sie empfangen hat. Über Topics und Publisher hinweg sowie in Shared Subscriptions gibt es keine Reihenfolgegarantie — und selbst im garantierten Rahmen können QoS-1-Wiederholungen verspätete Duplikate erzeugen.

Dieser Artikel erklärt genau, was die Spezifikation verspricht, wo die Reihenfolge in der Praxis bricht, und liefert zwei kurze Python-Skripte, die vertauschte und doppelte Nachrichten auf deinem eigenen Broker messen.

Was die Spezifikation sagt

Sowohl MQTT 3.1.1 als auch MQTT 5 (Abschnitt 4.6, "Message ordering") definieren jedes Topic standardmäßig als Ordered Topic. Leitet ein Broker Nachrichten auf einem geordneten Topic weiter, muss er PUBLISH-Pakete an Subscriber "für dasselbe Topic und dieselbe QoS" in der Reihenfolge senden, in der er sie von einem Client empfangen hat. Clients wiederum müssen unbestätigte Nachrichten erneut senden und Bestätigungen in Reihenfolge verschicken.

Die Spezifikation nennt auch den Haken. Bei QoS 1 kann ein Publisher, der 1, 2, 3, 4 sendet, dazu führen, dass ein Subscriber 1, 2, 3, 2, 3, 4 sieht: Nach einem Reconnect kommen die erneut gesendeten Kopien von 2 und 3 nach den Originalen an. Die letzte Kopie jeder Nachricht ist geordnet, aber ein Duplikat kann nach seinem Nachfolger auftauchen. Ist nie mehr als eine Nachricht gleichzeitig in flight, kann das nicht passieren.

MQTT-QoS-Stufen erklärt die Bestätigungsabläufe hinter diesen Regeln.

Wo die Reihenfolge in der Praxis bricht

• Mehrere Nachrichten in flight und Reconnects. Clients und Broker erlauben meist viele unbestätigte QoS-1/2-Nachrichten gleichzeitig (in MQTT 5 wird das Limit über Receive Maximum ausgehandelt). Bricht die Verbindung mit mehreren davon ab, kann die Wiederholung nach neueren Nachrichten ankommen.
• Shared Subscriptions. $share/group/topic verteilt Nachrichten auf mehrere Consumer. Jeder Consumer sieht seinen Anteil eventuell geordnet, aber die Verarbeitung läuft parallel, sodass das Gesamtergebnis ungeordnet ist. Siehe Shared Subscriptions.
• Mehrere Publisher. Zwei Gateways, die dasselbe Gerät melden, erzeugen zwei unabhängig geordnete Ströme. Der Broker darf sie beliebig verschränken.
• Bridges und Cluster. Eine Bridge ist nur ein weiterer Client, eine Bridge-Verbindung hält also die Reihenfolge; mehrere Pfade zwischen Brokern nicht. Geclusterte Broker erhalten auf einem Topic meist die Reihenfolge pro Publisher, aber prüfe die Dokumentation deines Brokers.
• Dein eigener Consumer. Die häufigste Ursache überhaupt — Nachrichten kommen geordnet an und werden dann nebenläufig verarbeitet (Thread-Pools, async-Handler, Worker-Queues).

Test 1: Reihenfolge auf deinem Broker messen

Der Publisher nummeriert Nachrichten pro Topic und pro Publisher-ID. Der Subscriber merkt sich die höchste gesehene Sequenz pro (Topic, Publisher) und meldet alles, was rückwärts läuft, jedes Duplikat und alles, was fehlt.

import os, sys, time
from collections import Counter, defaultdict
import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
RUN = sys.argv[1]                      # dieselbe Run-ID wie beim Publisher
QOS = int(os.getenv("MQTT_QOS", "1"))
last = defaultdict(lambda: -1)         # (Topic, Publisher) -> höchste Sequenz
seen = defaultdict(set)
stats = Counter()

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"testmqtt/order/{RUN}/#", qos=QOS)

def on_message(client, userdata, msg):
    pub_id, seq = msg.payload.decode().split(":")
    key, seq = (msg.topic, pub_id), int(seq)
    stats["received"] += 1
    stats["dup_flag"] += msg.dup
    if seq in seen[key]:
        stats["duplicates"] += 1
        return
    seen[key].add(seq)
    if seq < last[key]:
        stats["out_of_order"] += 1
        print(f"OUT OF ORDER {key}: {seq} after {last[key]}")
    last[key] = max(last[key], seq)

def missing():
    return sum(last[k] + 1 - len(seen[k]) for k in seen)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
if os.getenv("MQTT_USER"):
    client.username_pw_set(os.getenv("MQTT_USER"), os.getenv("MQTT_PASS"))
client.on_connect, client.on_message = on_connect, on_message
client.connect(HOST, 1883)
client.loop_start()
try:
    while True:
        time.sleep(5)
        print(dict(stats), "missing:", missing())
except KeyboardInterrupt:
    print("FINAL", dict(stats), "missing:", missing())

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

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
RUN, N = sys.argv[1], int(sys.argv[2]) if len(sys.argv) > 2 else 10000
PUB_ID = os.getenv("PUB_ID", "p1")
TOPICS = int(os.getenv("TOPICS", "4"))
QOS = int(os.getenv("MQTT_QOS", "1"))

pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
if os.getenv("MQTT_USER"):
    pub.username_pw_set(os.getenv("MQTT_USER"), os.getenv("MQTT_PASS"))
pub.max_inflight_messages_set(int(os.getenv("INFLIGHT", "20")))
pub.connect(HOST, 1883)
pub.loop_start()

seq, infos = [0] * TOPICS, []
for i in range(N):
    t = i % TOPICS
    infos.append(pub.publish(f"testmqtt/order/{RUN}/t{t}", f"{PUB_ID}:{seq[t]}", qos=QOS))
    seq[t] += 1
for info in infos:
    info.wait_for_publish(timeout=60)
pub.disconnect()
pub.loop_stop()
print(f"published {N} messages on {TOPICS} topics as {PUB_ID}")

pip install "paho-mqtt>=2"
python order_sub.py run42            # Terminal 1, zuerst starten
python order_pub.py run42 10000      # Terminal 2

Auf einem gesunden Broker und in einem stabilen Netzwerk sollte die Baseline mit out_of_order: 0, duplicates: 0 und missing: 0 enden. Vor den Skripten kannst du den Strom live verfolgen, indem du im Online-MQTT-Client testmqtt/order/# abonnierst, während der Publisher läuft.

Test 2: Grenzfälle provozieren

Ändere jeweils nur eine Variable und vergleiche mit der Baseline:

Für den Shared-Subscription-Lauf änderst du den Subscribe-Aufruf auf den $share/...-Filter; ob ein Filter zu deinen Topics passt, prüfst du mit dem Topic Matcher. Duplikate sind kein Broker-Bug — das ist QoS 1, das seine Arbeit macht; doppelte Nachrichten zeigt, wie du Consumer idempotent machst.

Die Reihenfolge nicht im eigenen Code verlieren

Die meisten Meldungen à la "MQTT hat meine Nachrichten vertauscht" stammen vom Consumer. In Node.js wartet ein async-Message-Handler nicht auf die vorherige Nachricht:

client.on("message", async (topic, payload) => {
  await saveToDb(topic, payload); // Nachricht 2 kann vor Nachricht 1 fertig sein
});

Verkette die Arbeit pro Topic, um die Reihenfolge pro Topic zu erhalten und verschiedene Topics trotzdem parallel zu verarbeiten:

const chains = new Map();
client.on("message", (topic, payload) => {
  const prev = chains.get(topic) ?? Promise.resolve();
  const next = prev.then(() => saveToDb(topic, payload)).catch(console.error);
  chains.set(topic, next);
});

Dasselbe gilt für Thread-Pools in Python oder Java und für Worker-Queues hinter dem Broker: Partitioniere die Arbeit nach Topic oder Geräte-ID, niemals per Round-Robin.

Auf Reihenfolge hin entwerfen

• Lege Ereignisse, die geordnet bleiben müssen, auf ein Topic eines einzigen Publishers — zum Beispiel site/device42/events, nicht ein Topic pro Ereignistyp.
• Füge der Payload eine Sequenznummer oder einen Quell-Zeitstempel hinzu, damit Consumer veraltete Updates verwerfen ("last writer wins") oder kleine Fenster umsortieren können.
• Mach Handler idempotent, denn QoS 1 wird Duplikate liefern.
• Betrachte Shared Subscriptions als grundsätzlich ungeordnet; partitioniere nach Gerät, wenn die Reihenfolge wichtig ist.
• Nutze Retained Messages für den "aktuellen Zustand", bei dem nur der letzte Wert zählt und die Reihenfolge weniger wichtig ist.

Die Reihenfolgegarantie von MQTT ist echt, aber eng: ein Publisher, ein Topic, eine QoS. Zehn Minuten mit einem Sequenznummern-Test zeigen dir, wie sich dein Broker und dein Netzwerk verhalten, und die Varianten zeigen genau, wo dein Design übernehmen muss — meist mit einer Sequenznummer in der Payload und einem Consumer, der jedes Topic der Reihe nach verarbeitet.

Häufig gestellte Fragen