MQTT-Daten in MongoDB: Gerätenachrichten mit Node.js und Time-Series-Collections speichern

MQTT-Daten in MongoDB zu speichern liegt nahe, wenn Ihre Geräte JSON senden: Dokumente werden so eingefügt, wie sie ankommen, und Time-Series-Collections halten den Speicherbedarf klein und Abfragen schnell. In diesem Tutorial bauen Sie mit MQTT.js einen Node.js-Subscriber, der Gerätenachrichten von einem Cloud-Broker gebündelt in eine MongoDB-Time-Series-Collection schreibt, mit automatischem Ablauf und fertigen Aggregationsabfragen.

Wie die Pipeline zusammenpasst

devices --MQTT--> ConnectMQ broker --MQTT--> Node.js subscriber --insertMany--> MongoDB time-series collection

Die Aufgabe des Brokers ist das Weiterleiten von Nachrichten; er schreibt nicht in Datenbanken. Der Subscriber ist ein gewöhnlicher MQTT-Client, den Sie neben Ihrer Datenbank betreiben, was das Design einfach und portabel hält. Auf der linken Seite funktioniert jeder verwaltete MQTT-Broker; dieses Tutorial verwendet ConnectMQ.

Warum Time-Series-Collections

Eine normale Collection speichert jeden Messwert als eigenes Dokument mit eigenem Overhead. Eine Time-Series-Collection gruppiert Messwerte intern nach einem Metadatenfeld und einem Zeitfenster, sodass IoT-Daten deutlich weniger Speicherplatz belegen und Zeitbereichsabfragen weniger scannen müssen. Zusätzlich erhalten Sie expireAfterSeconds zum automatischen Löschen alter Daten. Die Kompromisse: Dokumente sind praktisch nur anhängbar (Updates und Löschungen sind eingeschränkt), und eindeutige Indizes werden nicht unterstützt, sodass eine Deduplizierung vor dem Einfügen erfolgen muss, falls Sie sie brauchen.

Schritt 1: MongoDB starten

docker run -d --name mongo -p 27017:27017 \
  -v mongo-data:/data/db \
  mongo:7

Schritt 2: Die Time-Series-Collection anlegen

Gerätenachrichten kommen auf <prefix>/<site>/<deviceId>/telemetry mit einem Payload wie diesem an:

{"ts": "2026-10-16T09:30:00.000Z", "temperature": 23.4, "humidity": 41, "fw": "1.4.2"}

Die Collection speichert jede Nachricht als ein Dokument: ein Zeitfeld ts, ein Objekt meta mit den Werten, die die Serie identifizieren (Standort, Gerät), und die Messwerte. Folgen Ihre Topics einem Unified Namespace (UNS) nach der ISA-95-Hierarchie, etwa acme/werk-muenchen/montage/linie-1/sensor-01/telemetry, gehören auch Bereich und Linie in meta, damit sich Werte pro Halle oder Linie aggregieren lassen. Legen Sie die Collection einmalig mit einem Setup-Skript an.

setup.mjs:

import { MongoClient } from "mongodb";

const uri = process.env.MONGO_URI ?? "mongodb://localhost:27017";
const client = new MongoClient(uri);
await client.connect();
const db = client.db("iot");

try {
  await db.createCollection("telemetry", {
    timeseries: { timeField: "ts", metaField: "meta", granularity: "seconds" },
    expireAfterSeconds: 60 * 60 * 24 * 90, // 90 Tage aufbewahren
  });
  console.log("created time-series collection iot.telemetry");
} catch (err) {
  if (err.codeName === "NamespaceExists") console.log("collection already exists");
  else throw err;
}

await db.collection("telemetry").createIndex({ "meta.deviceId": 1, ts: -1 });
await client.close();

npm init -y
npm install mqtt mongodb
node setup.mjs

Wählen Sie granularity passend dazu, wie oft ein einzelnes Gerät meldet: seconds für Intervalle unter einer Minute, minutes für Messwerte alle paar Minuten, hours für spärliche Daten. Ein gutes metaField enthält Werte, die sich für eine Serie selten ändern; legen Sie niemals die Messwerte selbst in meta ab.

Schritt 3: Der Subscriber für MQTT-Daten in MongoDB

Der Subscriber hält den MQTT-Callback schlank: Er validiert die Nachricht, formt ein Dokument und legt es in einem Puffer im Arbeitsspeicher ab. Ein Timer schreibt den Puffer jede Sekunde per insertMany weg, oder sofort, sobald 500 Dokumente warten.

subscriber.mjs:

import mqtt from "mqtt";
import { MongoClient } from "mongodb";
import { randomBytes } from "node:crypto";

const MQTT_URL = process.env.MQTT_URL ?? "mqtts://broker.connectmq.com:8883";
const PREFIX = process.env.MQTT_TOPIC_PREFIX ?? "connectmq/demo/mongo-9d42a7";
const CLIENT_ID = process.env.MQTT_CLIENT_ID ?? `mongo-sink-${randomBytes(4).toString("hex")}`;
const MONGO_URI = process.env.MONGO_URI ?? "mongodb://localhost:27017";

const BATCH_SIZE = 500;
const FLUSH_MS = 1000;
const MAX_BUFFER = 50_000;

const mongo = new MongoClient(MONGO_URI);
await mongo.connect();
const telemetry = mongo.db("iot").collection("telemetry");

let buffer = [];
let flushing = false;

function toDocument(topic, payload) {
  // <prefix>/<site>/<deviceId>/telemetry
  const parts = topic.split("/");
  if (parts.length < 3 || parts.at(-1) !== "telemetry") return null;
  const deviceId = parts.at(-2);
  const site = parts.at(-3);

  let data;
  try {
    data = JSON.parse(payload.toString());
  } catch {
    return null;
  }
  if (typeof data !== "object" || data === null || Array.isArray(data)) return null;

  const { ts, ...fields } = data;
  const when = ts ? new Date(ts) : new Date();
  if (Number.isNaN(when.getTime())) return null;

  const doc = { ts: when, meta: { site, deviceId }, receivedAt: new Date() };
  for (const [key, value] of Object.entries(fields)) {
    if (key.startsWith("$") || key.includes(".")) continue; // unsichere Feldnamen
    if (["number", "string", "boolean"].includes(typeof value)) doc[key] = value;
  }
  return doc;
}

async function flush() {
  if (flushing || buffer.length === 0) return;
  flushing = true;
  const batch = buffer;
  buffer = [];
  try {
    await telemetry.insertMany(batch, { ordered: false });
    console.log(`inserted ${batch.length} documents`);
  } catch (err) {
    console.error("insert failed, re-queueing batch:", err.message);
    buffer = batch.concat(buffer).slice(-MAX_BUFFER);
  } finally {
    flushing = false;
  }
}

const timer = setInterval(flush, FLUSH_MS);

const client = mqtt.connect(MQTT_URL, {
  clientId: CLIENT_ID,
  username: process.env.MQTT_USERNAME,
  password: process.env.MQTT_PASSWORD,
  protocolVersion: 5,
  clean: false,                                   // persistente Session
  properties: { sessionExpiryInterval: 1800 },    // Broker hält sie 30 Minuten
  keepalive: 30,
  reconnectPeriod: 2000,
});

client.on("connect", (connack) => {
  console.log(`connected as ${CLIENT_ID}, session present: ${connack.sessionPresent}`);
  client.subscribe(`${PREFIX}/+/+/telemetry`, { qos: 1 }, (err) => {
    if (err) console.error("subscribe failed:", err.message);
  });
});

client.on("message", (topic, payload) => {
  const doc = toDocument(topic, payload);
  if (!doc) return;
  if (buffer.length >= MAX_BUFFER) {
    console.error("buffer full, dropping message");
    return;
  }
  buffer.push(doc);
  if (buffer.length >= BATCH_SIZE) flush();
});

client.on("error", (err) => console.error("mqtt error:", err.message));
client.on("reconnect", () => console.log("reconnecting..."));

async function shutdown() {
  console.log("shutting down");
  clearInterval(timer);
  await client.endAsync();
  await flush();
  await mongo.close();
  process.exit(0);
}
process.on("SIGINT", shutdown);
process.on("SIGTERM", shutdown);

Hinweise zum Design

• Bündeln mit insertMany und ordered: false. Ein Round Trip pro Batch statt pro Nachricht, und ein einzelnes fehlerhaftes Dokument stoppt den Rest des Batches nicht.
• Ein begrenzter Puffer. Ist MongoDB nicht erreichbar, werden fehlgeschlagene Batches erneut eingereiht, der Puffer ist aber gedeckelt, sodass der Speicher nicht unbegrenzt wachsen kann.
• Eine persistente Session. clean: false mit einem Session-Ablauf und einer festen Client ID weist den Broker an, QoS-1-Nachrichten zu puffern, während der Subscriber neu startet. Der öffentliche MQTT-Broker hält Sessions bis zu 30 Minuten. Verwenden Sie eine stabile MQTT_CLIENT_ID, sonst erhalten Sie bei jedem Start eine frische, leere Session. Mehr dazu unter MQTT Persistent Sessions.
• Bereinigung der Felder. Schlüssel, die mit $ beginnen oder Punkte enthalten, werden übersprungen, und nur primitive Werte werden gespeichert, damit ein bösartiges oder fehlerhaftes Gerät keine merkwürdigen Strukturen in Ihre Dokumente einschleusen kann.
• Gerätezeitstempel. Das ts des Geräts hält offline gepufferte Messwerte zeitlich an der richtigen Stelle; receivedAt hält fest, wann der Server sie erhalten hat, was beim Debuggen von Uhrenproblemen hilft.

Dokumente formen: eines pro Nachricht oder eines pro Metrik?

Der Subscriber speichert ein Dokument pro MQTT-Nachricht, mit jedem Messwert als Feld auf oberster Ebene. Diese Form entspricht der Art, wie Geräte publizieren, hält zusammengehörige Werte beieinander und macht Abfragen wie "Temperatur und Luftfeuchtigkeit im selben Moment" trivial. Sie funktioniert gut, wenn jeder Gerätetyp einen stabilen Satz an Feldern sendet.

Die Alternative ist ein Dokument pro Metrik: { ts, meta: { deviceId, metric }, value }. Sie passt zu Flotten, deren Geräte sehr unterschiedliche Messwerte melden, weil jedes Dokument dieselbe Struktur hat und Sie generisch auf meta.metric indizieren und aggregieren können. Der Preis sind mehr Dokumente und die Notwendigkeit, gemeinsam angekommene Werte wieder zusammenzuführen.

Wählen Sie pro Collection eine Form und bleiben Sie dabei. Gemischte Formen in einer einzigen Time-Series-Collection machen Aggregationen umständlich und schwächen die Speichervorteile. Wenn Sie unsicher sind, beginnen Sie mit einem Dokument pro Nachricht; später aufzuteilen ist einfacher als zusammenzuführen.

Schritt 4: Mit simulierten Geräten testen

Starten Sie den Subscriber und publizieren Sie Testwerte an den kostenlosen öffentlichen MQTT-Broker. Er ist für Tests und Prototypen gedacht und für alle sichtbar, bleiben Sie also bei synthetischen Daten statt echter Maschinendaten:

node subscriber.mjs

# in einem anderen Terminal
for i in $(seq 1 30); do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i mongo-sim-01 -q 1 \
    -t "connectmq/demo/mongo-9d42a7/plant1/sensor-01/telemetry" \
    -m "{\"temperature\": 2$((RANDOM % 10)).$((RANDOM % 10)), \"humidity\": 4$((RANDOM % 10))}"
  sleep 1
done

Sehen Sie sich die Daten anschließend in mongosh an:

use iot
db.telemetry.find().sort({ ts: -1 }).limit(3)

Schritt 5: Die Daten abfragen

Minutenmittelwerte für ein Gerät über die letzten 24 Stunden:

db.telemetry.aggregate([
  { $match: { "meta.deviceId": "sensor-01", ts: { $gte: new Date(Date.now() - 864e5) } } },
  { $group: {
      _id: { $dateTrunc: { date: "$ts", unit: "minute" } },
      avgTemp: { $avg: "$temperature" },
      maxTemp: { $max: "$temperature" },
      samples: { $sum: 1 }
  } },
  { $sort: { _id: 1 } }
])

Letzter Messwert pro Gerät, nützlich für eine Flottenübersicht:

db.telemetry.aggregate([
  { $sort: { "meta.deviceId": 1, ts: -1 } },
  { $group: { _id: "$meta.deviceId", last: { $first: "$$ROOT" } } }
])

Für langfristige Dashboards planen Sie einen Job, der stündliche Verdichtungen per $merge in eine separate Collection schreibt, und lassen die Rohdaten-Collection nach 90 Tagen ablaufen.

Checkliste für die Produktion

1. Erstellen Sie im ConnectMQ-Portal einen privaten Broker und setzen Sie MQTT_USERNAME, MQTT_PASSWORD und MQTT_CLIENT_ID auf die generierten Werte; bleiben Sie bei mqtts:// auf Port 8883.
2. Geben Sie dem Subscriber eigene Zugangsdaten. ConnectMQ bindet Client IDs an die ausgegebenen Zugangsdaten, und zwei Prozesse mit einer ID trennen sich gegenseitig.
3. Betreiben Sie den Subscriber unter einem Prozessmanager (systemd, PM2, Docker mit Restart-Policy).
4. Verwenden Sie einen MongoDB-Benutzer, der nur Einfügerechte auf iot.telemetry hat.
5. Akzeptieren Sie, dass QoS 1 Duplikate zustellen kann; wenn sie eine Rolle spielen, deduplizieren Sie im Subscriber (zum Beispiel mit einem kurzen In-Memory-Cache von Nachrichten-IDs), denn Time-Series-Collections können keine Eindeutigkeit erzwingen.
6. Um die Last auf mehrere Subscriber zu verteilen, nutzen Sie Shared Subscriptions aus MQTT 5 ($share/mongo/<prefix>/+/+/telemetry), verfügbar ab den kostenpflichtigen Serverless-Plänen.
7. Beobachten Sie Einfügerate und Puffergröße in Logs oder Metriken.

Fehlerbehebung

Eine SQL-Alternative mit Continuous Aggregates finden Sie unter MQTT-Daten in TimescaleDB; für kleinere Anforderungen reichen eventuell das Speichern von MQTT-Daten in SQLite mit Python oder MQTT-Daten in Google Sheets. Der Leitfaden zu MQTT-Client-Bibliotheken behandelt MQTT.js neben anderen Clients.

Starten Sie mit ConnectMQ

Verbinden Sie Ihre MongoDB-Pipeline in wenigen Minuten mit einem privaten Broker. Registrieren Sie sich unter app.connectmq.com für einen kostenlosen Serverless-Broker, ganz ohne Kreditkarte, exportieren Sie die generierten Zugangsdaten als Umgebungsvariablen und prüfen Sie Payloads im Live-Topic-Explorer des Portals, bevor sie Ihre Datenbank erreichen.

Häufig gestellte Fragen

Kann ConnectMQ MQTT-Nachrichten automatisch in MongoDB speichern?

Nein, ConnectMQ ist der MQTT-Broker und hat keine eingebauten Datenbankintegrationen. Betreiben Sie einen Subscriber, etwa den Node.js-Dienst aus diesem Tutorial oder einen Node-RED-Flow, der sich als MQTT-Client verbindet und in MongoDB schreibt.

Sollte ich für MQTT-Daten eine Time-Series-Collection oder eine normale Collection verwenden?

Verwenden Sie eine Time-Series-Collection für nur angehängte Sensorwerte: Sie speichert sie kompakt und unterstützt automatischen Ablauf. Verwenden Sie eine normale Collection für Daten, die Sie an Ort und Stelle aktualisieren, etwa Gerätekonfigurationen oder den letzten Zustand pro Gerät.

Wie vermeide ich den Verlust von MQTT-Nachrichten, wenn MongoDB ausfällt?

Puffern Sie im Subscriber mit einer begrenzten Queue, wiederholen Sie fehlgeschlagene Inserts und nutzen Sie QoS 1 mit einer persistenten MQTT-Session, damit der Broker Nachrichten vorhält, während der Subscriber neu startet.

Was bewirkt das metaField in einer MongoDB-Time-Series-Collection?

Es identifiziert jede Serie, zum Beispiel Standort und Geräte-ID. MongoDB gruppiert Messwerte mit denselben Metadaten intern zusammen, weshalb Messwerte niemals in das metaField gehören.