MQTT-Request-Response-Muster mit MQTT 5

Mit dem MQTT-Request-Response-Muster kann ein Client eine Frage stellen und eine gezielte Antwort zurückbekommen, obwohl MQTT selbst ein Publish/Subscribe-Protokoll ist. MQTT 5 hat zwei Properties eingeführt, Response Topic und Correlation Data, die das sauber und interoperabel machen. In diesem Leitfaden bauen Sie einen funktionierenden RPC-artigen Aufruf in Python und Node.js, testen ihn auf dem öffentlichen MQTT-Broker und lernen, mit Timeouts und Fehlern umzugehen.

Warum Request/Response über MQTT

Der meiste MQTT-Verkehr verläuft in eine Richtung: Sensoren publizieren, Dashboards abonnieren. Reale Systeme brauchen aber auch Fragen mit Antworten:

„Welche Firmware-Version läuft bei dir?“
„Lies Register 40001 und nenne mir den Wert.“
„Hat sich das Ventil tatsächlich bewegt?“
„Führe einen Selbsttest aus und melde das Ergebnis.“
„Übernimm den neuen Rezeptparameter aus dem MES und bestätige ihn.“

Sie könnten auf jedem Gerät einen HTTP-Endpunkt öffnen, doch Geräte sitzen oft hinter NAT oder in Mobilfunknetzen, in denen eingehende Verbindungen unmöglich sind. Da das Gerät ohnehin eine ausgehende MQTT-Verbindung hält, ist es einfacher, Anfrage und Antwort über den Broker zu leiten, und es bleibt bei einem Sicherheitsmodell. Der Vergleich von MQTT und HTTP geht genauer darauf ein, warum dieses rein ausgehende Modell wichtig ist.

So funktioniert das MQTT-Request-Response-Muster in MQTT 5

Das Muster nutzt zwei in MQTT 5 definierte PUBLISH-Properties:

Property	Gesetzt von	Zweck
Response Topic	Anfragender	Topic, auf dem der Antwortende die Antwort publizieren soll
Correlation Data	Anfragender	Opake Bytes, die der Antwortende unverändert zurückgibt

Der Ablauf:

Der Anfragende abonniert sein eigenes Antwort-Topic, zum Beispiel connectmq/demo/k3p8/rpc/replies/<requester-id>.
Er publiziert die Anfrage auf das Anfrage-Topic des Antwortenden, mit Response Topic auf dieses Antwort-Topic und Correlation Data auf eine eindeutige ID gesetzt.
Der Antwortende empfängt die Anfrage, erledigt die Arbeit und publiziert das Ergebnis auf das Response Topic, wobei er die Correlation Data übernimmt.
Der Anfragende ordnet die eingehende Correlation Data der offenen Anfrage zu und schließt sie ab.

Der Broker verfolgt Anfragen überhaupt nicht. Er leitet lediglich zwei gewöhnliche Nachrichten weiter. Alles, was daraus „RPC“ macht, lebt in den Clients – deshalb skaliert das Muster gut und funktioniert auf jedem MQTT-5-Broker.

Topic-Design für Request/Response

Gute Topics machen das Muster leicht abzusichern und zu debuggen:

Anfragen: <prefix>/rpc/<device-id>/req, sodass jedes Gerät nur seine eigenen Anfragen abonniert.
Antworten: <prefix>/rpc/replies/<requester-id>, sodass jeder Aufrufer nur seine eigenen Antworten erhält.
Verwenden Sie nie ein gemeinsames Antwort-Topic für viele Aufrufer. Jeder Aufrufer würde jede Antwort erhalten und die meisten verwerfen müssen.

In einem Unified Namespace (UNS), dessen Topics der ISA-95-Hierarchie folgen, hängen Sie die RPC-Topics am besten unter den Knoten der jeweiligen Anlage, etwa acme/werk-stuttgart/montage/linie-1/pumpe-07/rpc/req. So liegen Befehle und Telemetrie einer Maschine im selben Zweig, und Zugriffsregeln pro Werk oder Linie greifen automatisch auch für Anfragen.

Einen breiteren Blick auf die Strukturierung von Namen bieten die MQTT-Konventionen für Topic-Namen.

Antwortender in Python

Dieser Antwortende übernimmt die Rolle eines Geräts. Er wartet auf Anfragen und sendet ein Ergebnis über Response Topic und Correlation Data der Anfrage zurück. Er nutzt paho-mqtt 2.x mit MQTT 5 auf dem öffentlichen MQTT-Broker, der für Tests gedacht ist, nicht für Produktionsdaten:

import json
import uuid
import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

BROKER = "broker.connectmq.com"
REQ_TOPIC = "connectmq/demo/k3p8/rpc/pump-07/req"

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(REQ_TOPIC, qos=1)
    print("responder ready")

def on_message(client, userdata, msg):
    props = msg.properties
    response_topic = getattr(props, "ResponseTopic", None)
    correlation = getattr(props, "CorrelationData", None)
    if not response_topic:
        print("request without response topic, ignoring")
        return

    request = json.loads(msg.payload)
    if request.get("method") == "getStatus":
        result = {"ok": True, "rpm": 1480, "running": True}
    else:
        result = {"ok": False, "error": "unknown method"}

    reply_props = Properties(PacketTypes.PUBLISH)
    if correlation is not None:
        reply_props.CorrelationData = correlation
    client.publish(response_topic, json.dumps(result), qos=1, properties=reply_props)

client = mqtt.Client(
    mqtt.CallbackAPIVersion.VERSION2,
    client_id=f"responder-{uuid.uuid4().hex[:8]}",
    protocol=mqtt.MQTTv5,
)
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=30)
client.loop_forever()

Anfragender in Node.js mit Timeouts

Der Anfragende führt eine Map offener Aufrufe, indiziert nach Correlation-ID, und lehnt jeden Aufruf ab, der innerhalb des Timeouts keine Antwort erhält. Er nutzt MQTT.js v5 über TLS:

const mqtt = require("mqtt");
const { randomUUID } = require("crypto");

const clientId = "requester-" + randomUUID().slice(0, 8);
const replyTopic = `connectmq/demo/k3p8/rpc/replies/${clientId}`;
const pending = new Map();

const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId,
  protocolVersion: 5,
});

function call(deviceId, method, params = {}, timeoutMs = 5000) {
  const id = randomUUID();
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => {
      pending.delete(id);
      reject(new Error(`timeout calling ${method} on ${deviceId}`));
    }, timeoutMs);
    pending.set(id, { resolve, timer });
    client.publish(
      `connectmq/demo/k3p8/rpc/${deviceId}/req`,
      JSON.stringify({ method, params }),
      {
        qos: 1,
        properties: {
          responseTopic: replyTopic,
          correlationData: Buffer.from(id),
          messageExpiryInterval: Math.ceil(timeoutMs / 1000),
        },
      }
    );
  });
}

client.on("message", (topic, payload, packet) => {
  const corr = packet.properties && packet.properties.correlationData;
  if (!corr) return;
  const id = corr.toString();
  const entry = pending.get(id);
  if (!entry) return; // verspätete oder unbekannte Antwort
  clearTimeout(entry.timer);
  pending.delete(id);
  entry.resolve(JSON.parse(payload.toString()));
});

client.on("connect", () => {
  client.subscribe(replyTopic, { qos: 1 }, async (err) => {
    if (err) return console.error(err);
    try {
      const status = await call("pump-07", "getStatus");
      console.log("reply:", status);
    } catch (e) {
      console.error(e.message);
    } finally {
      client.end();
    }
  });
});

Starten Sie den Python-Antwortenden und führen Sie dann den Node.js-Anfragenden aus. Sie sollten reply: { ok: true, rpm: 1480, running: true } sehen. Stoppen Sie den Antwortenden und starten Sie den Anfragenden erneut, um den Timeout-Pfad zu sehen.

Beachten Sie, dass die Anfrage zusätzlich ein Message Expiry in Höhe des Timeouts trägt. Ist das Gerät mit persistenter Session offline, wird die Anfrage verworfen, sobald der Aufrufer aufgegeben hat, statt Minuten später ausgeführt zu werden. Die Details stehen in unserem Artikel über das MQTT Message Expiry Interval.

Checkliste für das Request/Response-Design

Abonnieren Sie das Antwort-Topic, bevor Sie die erste Anfrage senden, sonst verpassen Sie womöglich eine schnelle Antwort.
Verwenden Sie pro Anfrage eine eindeutige Correlation-ID, etwa eine UUID.
Implementieren Sie immer einen Timeout und räumen Sie offene Einträge auf.
Setzen Sie Message Expiry für Anfragen, damit veraltete Aufrufe nie ausgeführt werden.
Geben Sie strukturierte Fehler zurück (ok: false, Fehlercode), statt zu schweigen.
Gestalten Sie Aktionen nach Möglichkeit idempotent, weil QoS 1 eine Anfrage doppelt zustellen kann. Siehe Doppelte MQTT-Nachrichten.
Senden Sie Anfragen und Antworten nicht retained.

Den Antwort-Payload gestalten

Der Transport ist nur die halbe Vereinbarung; das Antwortformat zählt genauso, sobald mehrere Teams dieselben Geräte aufrufen. Ein kleiner, einheitlicher Umschlag hält Aufrufer einfach:

{"ok": false, "code": "BUSY", "message": "self-test already running", "retryAfter": 10}

Einige Konventionen, die sich in der Praxis bewähren:

Antworten Sie immer, auch bei Fehlern. Ein schweigender Antwortender ist von einem Offline-Gerät nicht zu unterscheiden, und der Aufrufer kann nur in den Timeout laufen.
Verwenden Sie maschinenlesbare Codes wie BUSY, INVALID_PARAMS oder UNKNOWN_METHOD, plus eine menschenlesbare Meldung für Logs.
Versionieren Sie Ihre Methoden. getStatus.v2 oder ein version-Feld erlaubt es, die Antwortstruktur zu ändern, ohne ältere Aufrufer zu brechen.
Trennen Sie Annahme und Abschluss. Bei allem, was länger als ein, zwei Sekunden dauert, antworten Sie sofort mit {"ok": true, "accepted": true, "jobId": "..."} und publizieren Fortschritt und Endergebnis auf einem Job-Topic. Der Timeout des Aufrufers deckt dann nur die Annahme ab.
Kennzeichnen Sie den Content Type. Wenn Sie die MQTT-5-Property Content Type bei Anfragen und Antworten auf application/json setzen, können generische Tools sie korrekt anzeigen.

Arbeiten mit MQTT-3.1.1-Clients

Sprechen manche Geräte nur MQTT 3.1.1, stehen die Properties nicht zur Verfügung. Der übliche Workaround ist, dieselben Informationen in den Payload zu legen, zum Beispiel {"replyTo": "...", "corrId": "...", "method": "getStatus"}. Das funktioniert, aber alle Beteiligten müssen sich auf die Feldnamen einigen, während MQTT-5-Properties standardisiert sind und den Payload sauber halten. Gemischte Flotten betreiben oft beides, bis ältere Firmware ausgemustert ist.

Fehlersuche

Es kommt keine Antwort an. Prüfen Sie, ob der Antwortende das Response Topic tatsächlich sieht. Verbindet sich eine der beiden Seiten mit MQTT 3.1.1, gehen die Properties verloren. Prüfen Sie außerdem, ob der Anfragende genau das richtige Antwort-Topic abonniert hat.

Antworten landen beim falschen Aufrufer. Ihr Antwort-Topic wird vermutlich geteilt. Nehmen Sie die Client ID des Anfragenden darin auf.

Correlation Data stimmt nicht überein. Sie ist binär. Vergleichen Sie Bytes oder dekodieren Sie beide Seiten einheitlich, wie es die Beispiele mit UTF-8-Strings tun.

Aufrufe hängen unter Last. Ein Single-Thread-Antwortender bearbeitet eine Anfrage nach der anderen. Lange Operationen sollten mit einem „accepted“-Status antworten und das Endergebnis später publizieren.

RPC auf ConnectMQ betreiben

ConnectMQ unterstützt Response Topic, Correlation Data und Message Expiry aus MQTT 5.0, sodass der obige Code unverändert gegen Ihren eigenen Broker läuft. Ersetzen Sie den Host durch Ihren Broker und ergänzen Sie username: "YOUR_USERNAME", password: "YOUR_PASSWORD" und clientId: "YOUR_CLIENT_ID" aus dem Portal. Ein verwalteter MQTT-Broker nimmt Ihnen den Betrieb ab, damit Sie sich auf die Request-Vereinbarung konzentrieren können. Für das große Ganze beginnen Sie mit dem MQTT-Tutorial.

Starten Sie mit ConnectMQ

Erstellen Sie unter app.connectmq.com einen kostenlosen Broker im Serverless-Plan, ganz ohne Kreditkarte. Lassen Sie den Antwortenden auf einem Gerät oder Laptop laufen, rufen Sie ihn aus einem Skript auf und beobachten Sie Anfragen und Antworten im Live-Topic-Explorer des ConnectMQ-Portals.

Häufig gestellte Fragen

Ordnet der Broker Anfragen und Antworten einander zu?

Nein. Der Broker leitet zwei unabhängige Nachrichten weiter. Der Anfragende ordnet sie anhand der von ihm gewählten Correlation Data zu.

Ist Correlation Data erforderlich?

Nicht zwingend, aber ohne sie kann ein Client mit mehreren offenen Anfragen nicht erkennen, welche Antwort zu welchem Aufruf gehört. Setzen Sie sie immer.

Kann ich QoS 0 für Request/Response verwenden?

Ja, verlorene Anfragen zeigen sich dann aber nur als Timeouts. QoS 1 mit idempotenten Handlern ist die übliche Wahl.

Ersetzt MQTT-Request/Response HTTP-APIs?

Für Gerätebefehle und Abfragen hinter NAT ist es oft die bessere Wahl. Für große Dateiübertragungen oder öffentliche APIs ist HTTP meist einfacher.

Das MQTT-Request/Response-Muster mit Response Topic und Correlation Data