MQTT-Nachrichten in MongoDB speichern mit Node.js (Time-Series-Collections)

Um MQTT-Nachrichten in MongoDB zu speichern, betreibst du einen Node.js-Consumer, der mit MQTT.js abonniert, jeden JSON-Payload in ein Dokument umwandelt und Batches per insertMany in eine Time-Series-Collection schreibt. MongoDB fasst die Messwerte dann pro Gerät in komprimierten Buckets zusammen und kann alte Daten automatisch löschen.

Dieses Tutorial baut den Consumer Schritt für Schritt: Collection anlegen, ein Batch-Writer mit begrenztem Puffer, eine Persistent MQTT Session für Neustarts sowie die Abfragen und Tests, die belegen, dass jede Nachricht angekommen ist.

Warum eine Time-Series-Collection

MQTT-Telemetrie hat eine feste Form: ein Zeitstempel, etwas, das die Quelle identifiziert, und ein paar Messwerte. Genau dafür sind Time-Series-Collections in MongoDB (ab 5.0) gebaut.

Die Time-Series-Optionen, die du beim Anlegen festlegst:

• timeField – das Date des Messwerts (ts).
• metaField – was die Serie identifiziert (meta.deviceId, meta.topic). Messwerte mit gleichem Meta-Wert werden zusammen gespeichert, halte ihn also pro Gerät stabil.
• granularity – "seconds" für Messwerte alle paar Sekunden; "minutes" oder "hours" für langsamere Geräte.
• expireAfterSeconds – automatisches Löschen alter Daten.

MongoDB starten und Collection anlegen

docker run -d --name mongo -p 27017:27017 mongo:7
npm install mqtt@5 mongodb@6

Der Consumer legt die Collection beim ersten Start an, ein separater Migrationsschritt ist also nicht nötig.

Der Consumer

import mqtt from "mqtt";
import { MongoClient } from "mongodb";

const MONGO_URL = process.env.MONGO_URL ?? "mongodb://localhost:27017";
const TOPIC = "testmqtt/mongo/+/telemetry";
const BATCH_SIZE = 500;
const FLUSH_MS = 1000;
const MAX_BUFFER = 50_000;

// 1. MongoDB: Time-Series-Collection einmalig anlegen
const mongo = new MongoClient(MONGO_URL);
await mongo.connect();
const db = mongo.db("iot");
if (!(await db.listCollections({ name: "telemetry" }).hasNext())) {
  await db.createCollection("telemetry", {
    timeseries: { timeField: "ts", metaField: "meta", granularity: "seconds" },
    expireAfterSeconds: 60 * 60 * 24 * 30, // 30 Tage aufbewahren
  });
}
const telemetry = db.collection("telemetry");

// 2. Batch-Writer
let buffer = [];
let flushing = false;

async function flush() {
  if (flushing || buffer.length === 0) return;
  flushing = true;
  const docs = buffer;
  buffer = [];
  try {
    const res = await telemetry.insertMany(docs, { ordered: false });
    console.log(`inserted ${res.insertedCount} docs`);
  } catch (err) {
    if (err.name === "MongoBulkWriteError") {
      console.error(`some documents rejected: ${err.message}`); // fehlerhafte Daten: nicht wiederholen
    } else {
      console.error(`MongoDB unavailable, re-queueing ${docs.length} docs: ${err.message}`);
      buffer = docs.concat(buffer).slice(-MAX_BUFFER);
    }
  } finally {
    flushing = false;
  }
}
const timer = setInterval(flush, FLUSH_MS);

// 3. MQTT: Persistent Session, QoS 1
const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId: "mongo-writer-01",
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  protocolVersion: 5,
  clean: false,
  properties: { sessionExpiryInterval: 3600 },
  reconnectPeriod: 2000,
});

client.on("connect", async (connack) => {
  console.log(`connected, session present: ${connack.sessionPresent}`);
  await client.subscribeAsync(TOPIC, { qos: 1 });
});
client.on("error", (err) => console.error(`mqtt error: ${err.message}`));

client.on("message", (topic, payload) => {
  let data;
  try {
    data = JSON.parse(payload.toString());
  } catch {
    console.warn(`skipping non-JSON payload on ${topic}`);
    return;
  }
  if (typeof data !== "object" || data === null) return;

  const deviceId = topic.split("/")[2];
  const { ts, ...measurements } = data;
  buffer.push({
    ts: typeof ts === "number" ? new Date(ts) : new Date(),
    meta: { deviceId, topic },
    ...measurements,
  });
  if (buffer.length > MAX_BUFFER) buffer.shift(); // Speicher schützen, Ältestes verwerfen
  if (buffer.length >= BATCH_SIZE) flush();
});

// 4. Sauber herunterfahren: erst MQTT stoppen, dann den Rest schreiben
async function shutdown() {
  clearInterval(timer);
  await client.endAsync();
  while (flushing) await new Promise((r) => setTimeout(r, 50));
  await flush();
  await mongo.close();
  process.exit(0);
}
process.on("SIGINT", shutdown);
process.on("SIGTERM", shutdown);

Was die einzelnen Teile tun:

• insertMany mit ordered: false schickt eine Anfrage pro Batch und fügt die übrigen Dokumente weiter ein, wenn eines abgelehnt wird. Ein insertOne pro Nachricht ist der häufigste Grund, warum solche Consumer zurückfallen.
• Das Zeitfeld wird in ein Date umgewandelt. Geräte senden meist Millisekunden seit Epoch; eine rohe Zahl in ts lässt den Insert scheitern.
• Die Fehlerbehandlung trennt fehlerhafte Daten von Ausfällen. Ein MongoBulkWriteError bedeutet, dass MongoDB Dokumente abgelehnt hat – Wiederholen hilft nicht. Netzwerk- oder Server-Selection-Fehler stellen den Batch für den nächsten Flush wieder in die Queue.
• MAX_BUFFER begrenzt den Speicher, wenn MongoDB lange ausfällt. Die ältesten Daten zu verwerfen, ist eine bewusste Entscheidung; für null Verlust verlässt du dich auf die unten beschriebene Broker-Session statt auf einen größeren Puffer.

Ist MQTT.js neu für dich, behandelt der JavaScript-MQTT-Guide Verbindungsoptionen und Reconnects, und der Code-Generator erzeugt einen Start-Client für deine Broker-Einstellungen.

Neustarts, Sessions und Duplikate

clean: false mit sessionExpiryInterval und einer festen clientId sorgt dafür, dass der Broker dein Abonnement behält und QoS-1-Nachrichten puffert, während der Consumer offline ist – etwa während eines Deployments. Beim Reconnect zeigt das Log session present: true, und die gepufferten Nachrichten treffen ein. Jede Instanz braucht ihre eigene Client-ID; erzeuge eine mit dem Client-ID-Generator. Mehr Hintergrund in Persistent Sessions.

Geteilte öffentliche Test-Endpunkte wie broker.connectmq.com halten keine Offline-Sessions; prüfe das Neustart-Verhalten also auf deiner eigenen privaten Instanz.

Zwei Grenzen solltest du kennen:

• MQTT.js bestätigt eine QoS-1-Nachricht, sobald das message-Event ausgelöst wurde – bevor der Batch geschrieben ist. Stürzt der Prozess ab, sind gepufferte Dokumente verloren. Brauchst du Bestätigung erst nach dem Schreiben, kannst du in MQTT.js client.handleMessage(packet, callback) überschreiben; das PUBACK geht erst raus, wenn du callback aufrufst. Allerdings verarbeitet es Nachrichten einzeln, tauscht also Durchsatz gegen Sicherheit.
• Time-Series-Collections unterstützen keine Unique-Indizes, du kannst QoS-1-Wiederholungen also nicht per eindeutigem Schlüssel deduplizieren. Wenn exakt einmalige Speicherung zählt, schreibe in eine normale Collection mit Unique-Index auf { "meta.deviceId": 1, seq: 1 } und ignoriere Duplicate-Key-Fehler.

Die Pipeline testen

1. MongoDB starten und node mqtt-to-mongo.mjs ausführen.
2. Im Online-MQTT-Client testmqtt/mongo/# abonnieren, um den Traffic zu sehen, und ein paar Messwerte publizieren:

for i in $(seq 1 30); do
  mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
    -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 -t testmqtt/mongo/sensor-0$((i % 3))/telemetry \
    -m "{\"ts\":$(date +%s000),\"seq\":$i,\"temperature\":$((20 + i % 4)).2,\"battery\":$((100 - i))}"
done

1. MongoDB abfragen:

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

db.telemetry.aggregate([
  { $match: { ts: { $gte: new Date(Date.now() - 3600000) } } },
  { $group: {
      _id: { device: "$meta.deviceId", minute: { $dateTrunc: { date: "$ts", unit: "minute" } } },
      avgTemp: { $avg: "$temperature" },
      n: { $sum: 1 } } },
  { $sort: { "_id.minute": 1 } }
])

Die Shell öffnest du mit docker exec -it mongo mongosh. Die Anzahl pro Gerät sollte in Summe 30 ergeben. Teste danach Fehlerfälle: Stoppe MongoDB mit docker stop mongo, publiziere zehn Nachrichten, starte es wieder und prüfe, dass der Writer sie neu eingereiht und eingefügt hat.

MongoDB-Test-Topic im Browser beobachten

Fehlerbehebung

Für mehrere Writer abonnierst du $share/mongo-writers/testmqtt/mongo/+/telemetry mit einer eigenen Client-ID pro Instanz; der Broker teilt den Strom unter ihnen auf (siehe Shared Subscriptions). Aktuelle MongoDB-Versionen legen bei neuen Time-Series-Collections automatisch einen Index auf Meta- und Zeitfeld an; bei älteren Versionen ergänzt du db.telemetry.createIndex({ "meta.deviceId": 1, ts: -1 }) selbst. Bevorzugt dein Team SQL für Analysen, vergleiche diesen Ansatz mit MQTT in PostgreSQL/TimescaleDB.

MongoDB ist ein natürliches Zuhause für JSON-Telemetrie, solange du drei Regeln beachtest: ein echtes Date im Zeitfeld, ein stabiles meta pro Gerät und Schreibvorgänge in Batches. Ergänze eine Persistent Session auf deinem eigenen Broker und einen Neustart-Test, dann übersteht die Pipeline Deployments, ohne dass es jemand merkt.

Häufig gestellte Fragen