MQTT.js Node.js Tutorial: Publish, Subscribe, QoS und TLS

MQTT.js ist der Standard-MQTT-Client für JavaScript, und MQTT.js in Node.js ist der schnellste Weg zu IoT-Backends, Bridges und Testwerkzeugen. In diesem Tutorial erstellen Sie mit MQTT.js v5 einen Publisher und einen Subscriber in Node.js, nutzen die Promise-basierte API, wählen QoS-Stufen, ergänzen einen Last Will, wechseln auf TLS und konfigurieren Reconnects für einen dauerhaft laufenden Dienst. Alle Beispiele laufen gegen den kostenlosen öffentlichen MQTT-Broker broker.connectmq.com – Sie brauchen also nur Node.js. Die gezeigten Muster eignen sich auch für industrielle Backends, etwa einen Dienst, der Maschinendaten aus dem Werk an ein MES oder ERP weiterreicht.

Voraussetzungen

• Node.js 18 oder neuer (node --version)
• npm, pnpm oder yarn
• Optional: mosquitto_sub, um den Datenverkehr in einem zweiten Terminal zu beobachten

Legen Sie ein Projekt an und installieren Sie die Bibliothek:

mkdir mqtt-node-demo && cd mqtt-node-demo
npm init -y
npm pkg set type=module
npm install mqtt

Mit "type": "module" können Sie import-Syntax und Top-Level-await verwenden. MQTT.js v5 ist in TypeScript geschrieben und bringt eigene Typdefinitionen mit, sodass Editoren ohne zusätzliche Pakete Autovervollständigung bieten.

Verbindungs-URLs und Ports

MQTT.js wählt den Transport anhand des URL-Schemas. Auf ConnectMQ-Brokern gelten diese Ports:

In Node.js sollten Sie mqtts:// auf 8883 bevorzugen. WebSocket ist für Browser relevant, die keine rohen TCP-Sockets öffnen können; diese Seite behandelt der Artikel MQTT over WebSocket.

Ihr erster MQTT.js-Subscriber

Speichern Sie als subscriber.js:

import mqtt from "mqtt";

const PREFIX = "connectmq/demo/node-a71c";
const clientId = `node-sub-${Math.random().toString(16).slice(2, 10)}`;

const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId,
  protocolVersion: 5,
  clean: true,
  keepalive: 30,
  reconnectPeriod: 2000,   // 2 s zwischen Reconnect-Versuchen warten
  connectTimeout: 10000,   // einzelnen Versuch nach 10 s abbrechen
});

client.on("connect", async (connack) => {
  console.log(`Connected as ${clientId}, session present: ${connack.sessionPresent}`);
  const granted = await client.subscribeAsync(`${PREFIX}/+/telemetry`, { qos: 1 });
  console.log("Subscribed:", granted.map((g) => `${g.topic} (qos ${g.qos})`).join(", "));
});

client.on("message", (topic, payload, packet) => {
  let data;
  try {
    data = JSON.parse(payload.toString());
  } catch {
    data = payload.toString();
  }
  const device = topic.split("/").at(-2);
  console.log(`[${device}] qos=${packet.qos} retain=${packet.retain}`, data);
});

client.on("reconnect", () => console.log("Reconnecting..."));
client.on("offline", () => console.log("Offline"));
client.on("error", (err) => console.error("MQTT error:", err.message));

process.on("SIGINT", async () => {
  await client.endAsync();
  process.exit(0);
});

Das Abonnieren im connect-Handler ist Absicht. Mit clean: true vergisst der Broker die Subscriptions, wenn die Verbindung abbricht. Wer bei jedem Connect neu abonniert, empfängt also auch nach einem Reconnect zuverlässig Nachrichten.

Publizieren mit QoS und Last Will

Speichern Sie als publisher.js:

import mqtt from "mqtt";

const PREFIX = "connectmq/demo/node-a71c";
const DEVICE = "sensor-01";
const clientId = `node-pub-${Math.random().toString(16).slice(2, 10)}`;

const client = await mqtt.connectAsync("mqtts://broker.connectmq.com:8883", {
  clientId,
  protocolVersion: 5,
  will: {
    topic: `${PREFIX}/${DEVICE}/status`,
    payload: Buffer.from("offline"),
    qos: 1,
    retain: true,
  },
});

console.log("Connected");
await client.publishAsync(`${PREFIX}/${DEVICE}/status`, "online", { qos: 1, retain: true });

let count = 0;
const timer = setInterval(async () => {
  const reading = {
    temperature: +(20 + Math.random() * 5).toFixed(2),
    humidity: +(40 + Math.random() * 10).toFixed(1),
    ts: Date.now(),
  };
  try {
    await client.publishAsync(`${PREFIX}/${DEVICE}/telemetry`, JSON.stringify(reading), {
      qos: 1,
      properties: { messageExpiryInterval: 60, contentType: "application/json" },
    });
    console.log("Published", reading);
  } catch (err) {
    console.error("Publish failed:", err.message);
  }
  if (++count === 20) {
    clearInterval(timer);
    await client.publishAsync(`${PREFIX}/${DEVICE}/status`, "offline", { qos: 1, retain: true });
    await client.endAsync();
  }
}, 2000);

Starten Sie beide:

node subscriber.js
# in einem zweiten Terminal
node publisher.js

Dabei passiert Folgendes:

• connectAsync wird aufgelöst, sobald der Broker CONNACK sendet – praktisch in Skripten. In dauerhaft laufenden Diensten bleibt das eventbasierte connect nützlich, weil es nach jedem Reconnect erneut ausgelöst wird.
• QoS 1 bedeutet, dass der Broker jeden Publish bestätigt, und publishAsync wird erst nach dieser Bestätigung aufgelöst. Für hochfrequente Telemetrie, bei der ein gelegentlicher Verlust in Ordnung ist, ist QoS 0 günstiger. Die Abwägungen behandelt MQTT-QoS-Stufen erklärt.
• Der Last Will wird beim Verbindungsaufbau vom Broker gespeichert und publiziert, wenn der Client ohne sauberes Disconnect verschwindet – so erfahren andere Dienste, dass der Sensor offline ist. Weil er retained ist, sehen neue Subscriber den aktuellen Status sofort.
• MQTT-5-Properties wie messageExpiryInterval weisen den Broker an, die Nachricht zu verwerfen, wenn sie nicht innerhalb von 60 Sekunden zugestellt werden kann. So stauen sich keine veralteten Messwerte für Offline-Subscriber.

Der kostenlose öffentliche MQTT-Broker ist für Tests und Prototypen gedacht und für alle lesbar. Ersetzen Sie daher den Teil node-a71c im Präfix durch eine eigene Zufallszeichenfolge und senden Sie dort niemals echte Daten.

Reconnect-Optionen, die zählen

MQTT.js verbindet sich automatisch neu. Diese Optionen bestimmen das Verhalten:

Für einen Dienst, der bei kurzen Offline-Phasen keine Nachrichten verpassen darf, kombinieren Sie eine feste Client ID, clean: false, QoS-1-Subscriptions und ein Session-Expiry-Intervall. Der Broker puffert dann Nachrichten für die Session und stellt sie zu, sobald der Client zurück ist.

Einen dauerhaft laufenden Node.js-Dienst strukturieren

Die obigen Skripte sind bewusst kurz. Ein Dienst, der monatelang läuft, braucht etwas mehr Struktur – und die betrifft vor allem den Lebenszyklus, weniger MQTT selbst:

• Erstellen Sie einen Client pro Prozess und teilen Sie ihn über ein Modul. Eine neue Verbindung pro Request oder Job verschwendet Ressourcen und stößt schnell an Verbindungslimits.
• Behandeln Sie connect als „bereit". Machen Sie Arbeit, die den Broker braucht, von der Eigenschaft connected des Clients abhängig, und legen Sie fest, was bei false geschieht: puffern, verwerfen oder dem Aufrufer einen Fehler zurückgeben.
• Behandeln Sie Shutdown-Signale. Bei SIGTERM (gesendet von Docker, Kubernetes und systemd) publizieren Sie, falls genutzt, einen abschließenden Status, rufen endAsync() auf, damit laufende QoS-1-Nachrichten noch übertragen werden, und beenden erst dann den Prozess.
• Loggen Sie Zustandswechsel, nicht jede Nachricht. Die Events connect, reconnect, offline, close und error brauchen Sie bei der Analyse eines Ausfalls. Logs pro Nachricht lassen sie untergehen.
• Halten Sie Handler schnell. Das message-Event läuft in der Event Loop von Node. Aufwendige Arbeit wie Datenbank-Schreibvorgänge sollten Sie sorgfältig awaiten oder in eine Queue verlagern, damit ein langsames nachgelagertes System nicht jede andere Nachricht verzögert.

Diese Gewohnheiten sind wichtiger als jeder einzelne Optionswert und gelten gleichermaßen für Bridges, Worker und Testwerkzeuge auf Basis von MQTT.js.

Authentifizierung bei einem privaten Broker

Für die Produktion verwenden Sie die Zugangsdaten Ihres Brokers. Mit ConnectMQ erstellen Sie im Portal einen Broker und erhalten Benutzername, Passwort und Client ID. Laden Sie diese aus Umgebungsvariablen:

import mqtt from "mqtt";

const client = mqtt.connect(`mqtts://${process.env.MQTT_HOST}:8883`, {
  clientId: process.env.MQTT_CLIENT_ID,   // YOUR_CLIENT_ID aus dem Portal
  username: process.env.MQTT_USERNAME,    // YOUR_USERNAME
  password: process.env.MQTT_PASSWORD,    // YOUR_PASSWORD
  protocolVersion: 5,
  clean: false,
  properties: { sessionExpiryInterval: 3600 },
});

client.on("connect", () => console.log("Connected to private broker"));
client.on("error", (err) => console.error(err.message));

ConnectMQ verwendet ein Let's-Encrypt-Zertifikat, dem Node.js von Haus aus vertraut, sodass Sie keine CA-Datei übergeben müssen. Setzen Sie in der Produktion niemals rejectUnauthorized: false; das deaktiviert die Zertifikatsprüfung und macht TLS sinnlos.

Fehlerbehebung bei MQTT.js in Node.js

1. Verbindet sich und trennt sich alle paar Sekunden. Ein anderer Client verwendet dieselbe Client ID. Jeder laufende Prozess braucht eine eigene ID.
2. ERR_REQUIRE_ESM oder Cannot use import statement. Setzen Sie "type": "module" in der package.json oder verwenden Sie in CommonJS-Dateien const mqtt = require("mqtt").
3. Keine Nachrichten nach einem Netzwerkabbruch. Sie haben bei clean: true nur einmal außerhalb des connect-Handlers abonniert.
4. connack timeout. Eine Firewall blockiert Port 8883. Testen Sie mit mosquitto_sub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs -i test-1 -t test oder probieren Sie wss:// auf 8084.
5. Subscription mit QoS 128 oder einem Reason Code ab 0x80 bestätigt. Der Broker hat die Subscription abgelehnt, oft wegen eines ungültigen Filters oder eines Subscription-Limits auf dem öffentlichen MQTT-Broker (10 pro Client).
6. Speicherverbrauch wächst im Offline-Zustand. Ausgehende Nachrichten werden bis zum Reconnect gepuffert. Begrenzen Sie Ihren eigenen Puffer oder verwenden Sie für verzichtbare Telemetrie QoS 0 mit queueQoSZero: false.

Nächste Schritte

Von hier aus können Sie denselben Client ins Frontend bringen: React mit MQTT baut ein Hooks-basiertes Dashboard auf dem Browser-Build von MQTT.js, und typsicheres MQTT in TypeScript ergänzt diesen Code um typisierte Topics und validierte Payloads. Für andere Laufzeitumgebungen vergleicht der Leitfaden zu MQTT-Client-Bibliotheken die wichtigsten Optionen. Die vollständige Optionsliste finden Sie in der MQTT.js README.

Starten Sie mit ConnectMQ

Bringen Sie Ihre Node.js-Dienste in wenigen Minuten vom öffentlichen MQTT-Broker auf einen privaten Broker. Registrieren Sie sich unter app.connectmq.com, wählen Sie den kostenlosen Serverless-Plan ohne Kreditkarte, erstellen Sie einen Broker und hinterlegen Sie die generierten Zugangsdaten in Ihrer Umgebung. Als verwalteter MQTT-Broker betreibt ConnectMQ die TLS-Endpunkte und die Broker-Infrastruktur, während Sie sich auf Ihren Anwendungscode konzentrieren.

Häufig gestellte Fragen

Ist MQTT.js v5 mit älterem Code kompatibel?

Größtenteils. Die Callback-API funktioniert weiterhin, und v5 ergänzt Promise-basierte Methoden wie connectAsync, publishAsync, subscribeAsync und endAsync. Die wichtigsten Änderungen sind die Neuimplementierung in TypeScript und eine höhere Mindestversion von Node.js.

Sollte ich mit MQTT.js MQTT 3.1.1 oder MQTT 5 verwenden?

Verwenden Sie protocolVersion: 5, wenn Ihr Broker es unterstützt, wie es bei ConnectMQ der Fall ist. Sie erhalten Reason Codes, Message Expiry, Session Expiry und User Properties, während der Code für Publish und Subscribe gleich bleibt.

Wie viele Verbindungen sollte ein Node.js-Dienst öffnen?

In der Regel eine. Ein einzelner MQTT.js-Client kann gleichzeitig auf viele Topics publizieren und sie abonnieren. Jede zusätzliche Verbindung braucht eine eigene, eindeutige Client ID und zählt zum Verbindungslimit Ihres Plans.

Kann ich MQTT.js mit Bun oder Deno verwenden?

MQTT.js zielt auf Node.js und Browser. Laufzeitumgebungen mit guter Node.js-Kompatibilität können es oft ausführen, testen Sie aber TLS- und Reconnect-Verhalten gründlich, bevor Sie sich darauf verlassen.