MQTT-Nachrichten an Webhooks und REST-APIs weiterleiten

Um MQTT-Nachrichten an einen Webhook oder eine REST-API weiterzuleiten, betreibst du einen kleinen Consumer, der die relevanten Topics abonniert und jede Nachricht in einen HTTP-POST verwandelt. Das Abonnieren ist der einfache Teil. Produktionsreif wird ein Forwarder erst durch das, was passiert, wenn die HTTP-Seite Probleme macht: Timeouts, 503 Service Unavailable, 429 Too Many Requests, Deploys, die eine Minute lang Verbindungen kappen. Dieses Tutorial baut einen Node.js-Forwarder mit MQTT.js 5, der mit exponentiellem Backoff und Jitter wiederholt, Retry-After beachtet, Requests signiert, Idempotency Keys sendet und nicht zustellbare Nachrichten auf einem Dead-Letter-Topic parkt.

Die Architektur

devices ──MQTT──▶ broker ──▶ forwarder ──HTTPS POST──▶ webhook / REST API
                                 │
                                 └──▶ testmqtt/webhook/dlq (failed deliveries)

Geräte publishen Events auf testmqtt/webhook/<device>/events. Der Forwarder abonniert mit QoS 1, postet jedes Event als JSON und entscheidet anhand des HTTP-Ergebnisses, ob es erledigt ist, wiederholt oder auf Dead-Letter gelegt wird. So müssen Geräte nichts über HTTP wissen und die empfangende API nichts über MQTT. Ist der Empfänger eine Serverless-Funktion, gilt dasselbe Muster; die Variante findest du unter MQTT mit AWS Lambda.

Retry-Regeln, die funktionieren

Nutze exponentiellen Backoff mit Full Jitter: Warte eine zufällige Zeit zwischen 0 und min(cap, base × 2^attempt). Ohne Jitter wiederholen alle Nachrichten, die während eines Ausfalls gescheitert sind, im selben Moment und legen die API genau dann wieder lahm, wenn sie sich erholt.

Der Forwarder

Installiere MQTT.js (npm install mqtt). Node 18+ bringt fetch und AbortSignal.timeout mit, eine HTTP-Bibliothek ist also nicht nötig:

import mqtt from "mqtt";
import crypto from "node:crypto";

const WEBHOOK_URL = process.env.WEBHOOK_URL ?? "http://localhost:3000/hook";
const SECRET = process.env.WEBHOOK_SECRET ?? "change-me";
const MAX_ATTEMPTS = 6;
const CONCURRENCY = 8;
const DLQ_TOPIC = "testmqtt/webhook/dlq";

const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId: "webhook-forwarder-1",
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  protocolVersion: 5,
});

client.on("connect", () => {
  console.log("verbunden");
  client.subscribe("testmqtt/webhook/+/events", { qos: 1 });
});
client.on("error", (err) => console.error("MQTT-Fehler:", err.message));

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const retryable = (status) => status === 408 || status === 429 || status >= 500;

async function deliver(job) {
  const body = JSON.stringify({
    topic: job.topic,
    payload: job.payload.toString("utf8"),
    receivedAt: new Date().toISOString(),
  });
  const signature = crypto.createHmac("sha256", SECRET).update(body).digest("hex");

  for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
    let retryAfterMs = 0;
    try {
      const res = await fetch(WEBHOOK_URL, {
        method: "POST",
        headers: {
          "content-type": "application/json",
          "idempotency-key": job.id,
          "x-signature-sha256": signature,
        },
        body,
        signal: AbortSignal.timeout(10_000),
      });
      await res.body?.cancel();                    // Verbindung freigeben
      if (res.ok) return { ok: true };
      if (!retryable(res.status)) return { ok: false, reason: `HTTP ${res.status}` };
      retryAfterMs = (Number(res.headers.get("retry-after")) || 0) * 1000;
      console.warn(`Versuch ${attempt}: HTTP ${res.status}`);
    } catch (err) {
      console.warn(`Versuch ${attempt}: ${err.name} ${err.message}`);
    }
    if (attempt < MAX_ATTEMPTS) {
      const backoff = Math.random() * Math.min(30_000, 500 * 2 ** attempt);
      await sleep(Math.max(backoff, retryAfterMs));
    }
  }
  return { ok: false, reason: "Retries ausgeschöpft" };
}

const queue = [];
let active = 0;

function drain() {
  while (active < CONCURRENCY && queue.length > 0) {
    const job = queue.shift();
    active++;
    deliver(job)
      .then((result) => {
        if (!result.ok) {
          console.error(`Dead-Letter für ${job.id}: ${result.reason}`);
          client.publish(DLQ_TOPIC, JSON.stringify({
            topic: job.topic,
            payload: job.payload.toString("base64"),
            id: job.id,
            reason: result.reason,
          }), { qos: 1 });
        }
      })
      .finally(() => { active--; drain(); });
  }
}

client.on("message", (topic, payload) => {
  // stabiler Key: dieselbe erneut zugestellte Nachricht bekommt denselben Key
  const id = crypto.createHash("sha256").update(topic).update(payload).digest("hex").slice(0, 32);
  queue.push({ topic, payload, id });
  drain();
});

Was die einzelnen Teile tun:

• Timeout für jeden Request. AbortSignal.timeout(10_000) verhindert, dass ein hängender Endpoint alle Worker blockiert.
• Begrenzte Parallelität. Acht Requests gleichzeitig; passe das an das an, was die API verträgt. Ohne Limit wird aus einem MQTT-Burst ein HTTP-Burst.
• HMAC-Signatur. Der Empfänger berechnet HMAC-SHA256(secret, body) neu und lehnt Requests ab, die nicht passen: übliche Webhook-Praxis.
• Idempotency Key. Aus Topic und Payload abgeleitet, sodass eine erneut zugestellte Nachricht denselben Key trägt. Das setzt voraus, dass Payloads einen Timestamp oder eine Event-ID enthalten; können zwei tatsächlich verschiedene Events byte-identisch sein, pack eine ID in den Payload und nutze diese.
• Dead-Letter-Topic. Nicht zustellbare Nachrichten landen auf testmqtt/webhook/dlq, statt zu verschwinden. Abonniere es aus einem Alerting-Job oder spiel es nach dem Fix der API erneut ab.

Zustellgarantien und ihre Grenzen

Standardmäßig bestätigt MQTT.js eine QoS-1-Nachricht, sobald dein message-Handler zurückkehrt. Die In-Memory-Queue oben geht also verloren, wenn der Prozess abstürzt. Für viele Webhooks im Benachrichtigungsstil ist das in Ordnung. Wenn nicht, hast du zwei Optionen.

Erst nach der Zustellung bestätigen. Statt message-Handler und In-Memory-Queue überschreibst du handleMessage(packet, callback) von MQTT.js; das PUBACK wird erst gesendet, wenn du callback aufrufst. Nachrichten werden dann pro Client nacheinander verarbeitet, was zusätzlich die Reihenfolge erhält:

client.handleMessage = (packet, callback) => {
  const id = crypto.createHash("sha256").update(packet.topic).update(packet.payload).digest("hex").slice(0, 32);
  deliver({ topic: packet.topic, payload: packet.payload, id })
    .then((result) => { if (!result.ok) console.error("dead-letter:", id, result.reason); })
    .finally(() => callback());   // PUBACK erst jetzt
};

Kombiniere das mit clean: false (MQTT 3.1.1) oder einem Session Expiry Interval (MQTT 5), damit der Broker Nachrichten puffert, solange der Forwarder weg ist. Dafür muss der Broker persistente Sessions halten: Nutze deine eigene private Broker-Instanz, denn der geteilte Test-Endpoint hält keine Offline-Sessions. Siehe MQTT Persistent Sessions.

Eine dauerhafte Queue dazwischenschalten. Schreib die Nachrichten in Redis Streams, SQS oder eine Datenbanktabelle und lass HTTP-Worker sie abarbeiten. Mehr bewegliche Teile, aber Retries überleben Neustarts.

So oder so ist das Ergebnis at-least-once, die empfangende API muss also Duplikate verkraften. Doppelte MQTT-Nachrichten behandeln beschreibt Muster zur Deduplizierung.

Horizontal skalieren

Ein einzelner Forwarder schafft viele Webhooks, weil die Arbeit hauptsächlich aus Warten auf HTTP besteht. Brauchst du mehr, starte mehrere Instanzen mit einer Shared Subscription wie $share/webhook/testmqtt/webhook/+/events. Der Broker verteilt die Nachrichten auf die Instanzen, und fällt eine aus, übernehmen die anderen ihren Anteil. Gib jeder Instanz eine eindeutige Client-ID.

Die Retries testen

Starte einen Empfänger, der 40 % der Requests mit 503 beantwortet und Duplikate loggt:

import http from "node:http";

const seen = new Set();
http.createServer((req, res) => {
  let body = "";
  req.on("data", (chunk) => (body += chunk));
  req.on("end", () => {
    if (Math.random() < 0.4) {
      res.writeHead(503, { "retry-after": "1" });
      return res.end();
    }
    const key = req.headers["idempotency-key"];
    console.log(seen.has(key) ? "DUPLIKAT" : "angenommen", key, body);
    seen.add(key);
    res.writeHead(204).end();
  });
}).listen(3000, () => console.log("Empfänger auf :3000"));

Starte node receiver.mjs, dann node forwarder.mjs, und publishe eine Reihe von Events:

for i in $(seq 1 20); do
  mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
    -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
    -t testmqtt/webhook/door-1/events -m "{\"event\":\"open\",\"n\":$i,\"ts\":$(date +%s)}"
done

Jedes Event sollte irgendwann als angenommen erscheinen, mit Warnungen Versuch N: HTTP 503 im Forwarder-Log. Teste dann die anderen Pfade:

• Lass den Empfänger 400 zurückgeben: Die Nachrichten müssen sofort, ohne Retries, auf dem Dead-Letter-Topic landen. Beobachte das mit mosquitto_sub -t testmqtt/webhook/dlq -v oder im Online-MQTT-Client.
• Stoppe den Empfänger für 30 Sekunden, während du publishst: Die Retries sollten sich zeitlich strecken und erfolgreich sein, sobald er zurück ist.

Events und Dead-Letter-Topic im Browser beobachten

Ein Webhook-Forwarder besteht aus ein paar Dutzend Zeilen, und fast alle drehen sich um Fehler. Klassifiziere Antworten, nutze Backoff mit Jitter, sende Idempotency Keys, verwirf nie still eine Nachricht und führe den Test mit dem fehlerhaften Empfänger aus, bevor du ihm Produktionstraffic anvertraust. Gehört dir die API, lohnt es sich auch, die HTTP-Seite mit Express zu bauen, direkt neben dem Forwarder.

Häufig gestellte Fragen