Guarda mensajes MQTT en MongoDB con Node.js (colecciones time series)

Para guardar mensajes MQTT en MongoDB, ejecuta un consumidor Node.js que se suscriba con MQTT.js, convierta cada payload JSON en un documento y escriba lotes con insertMany en una colección time series. MongoDB agrupa entonces las lecturas en buckets comprimidos por dispositivo y puede borrar los datos antiguos automáticamente.

Este tutorial construye ese consumidor paso a paso: creación de la colección, un escritor por lotes con un buffer acotado, una sesión MQTT persistente para los reinicios, y las consultas y pruebas que demuestran que cada mensaje llegó.

Por qué una colección time series

La telemetría MQTT tiene una forma fija: un timestamp, algo que identifica el origen y unas pocas mediciones. Las colecciones time series de MongoDB (MongoDB 5.0+) están diseñadas justo para eso.

Las opciones time series que eliges al crearla:

• timeField: el Date de la lectura (ts).
• metaField: lo que identifica la serie (meta.deviceId, meta.topic). Las lecturas con el mismo valor meta se almacenan juntas, así que mantenlo estable por dispositivo.
• granularity: "seconds" para lecturas cada pocos segundos; "minutes" u "hours" para dispositivos más lentos.
• expireAfterSeconds: borrado automático de los datos antiguos.

Ejecuta MongoDB y crea la colección

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

El consumidor crea la colección en el primer arranque, así que no necesitas un paso de migración aparte.

El consumidor

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: crear la colección time series una sola vez
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, // conservar 30 días
  });
}
const telemetry = db.collection("telemetry");

// 2. Escritor por lotes
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}`); // datos inválidos: no reintentar
    } 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: sesión persistente, 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(); // proteger la memoria, descartar lo más antiguo
  if (buffer.length >= BATCH_SIZE) flush();
});

// 4. Apagado ordenado: parar MQTT y luego escribir lo pendiente
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);

Qué hace cada parte:

• insertMany con ordered: false envía una petición por lote y sigue insertando el resto de documentos si uno es rechazado. Un insertOne por mensaje es la razón más habitual por la que estos consumidores se quedan atrás.
• El campo de tiempo se convierte en Date. Los dispositivos suelen enviar milisegundos desde epoch; un número en bruto en ts hace fallar el insert.
• La gestión de errores separa los datos inválidos de las caídas. Un MongoBulkWriteError significa que MongoDB rechazó documentos: reintentar no servirá. Los errores de red o de selección de servidor vuelven a encolar el lote para el siguiente flush.
• MAX_BUFFER limita la memoria si MongoDB está caído mucho tiempo. Descartar los datos más antiguos es una decisión deliberada; para no perder nada, apóyate en la sesión del broker descrita más abajo en lugar de en un buffer mayor.

Si MQTT.js es nuevo para ti, la guía de MQTT con JavaScript cubre las opciones de conexión y las reconexiones, y el generador de código produce un cliente inicial con los ajustes de tu broker.

Reinicios, sesiones y duplicados

clean: false con sessionExpiryInterval y un clientId fijo hace que el broker conserve tu suscripción y encole los mensajes QoS 1 mientras el consumidor está desconectado, por ejemplo durante un despliegue. Al reconectar, el log muestra session present: true y llegan los mensajes encolados. Cada instancia necesita su propio client ID; genera uno con el generador de client ID. Más contexto en sesiones persistentes.

Los endpoints públicos de prueba compartidos como broker.connectmq.com no guardan sesiones offline, así que verifica el comportamiento de reinicio en tu propia instancia privada.

Dos límites que conviene conocer:

• MQTT.js confirma un mensaje QoS 1 en cuanto se ha disparado el evento message, antes de que se escriba el lote. Si el proceso se cae, los documentos en el buffer se pierden. Si necesitas confirmar después de escribir, MQTT.js te permite sobrescribir client.handleMessage(packet, callback), que envía el PUBACK solo cuando llamas a callback; pero procesa los mensajes de uno en uno, así que cambia throughput por seguridad.
• Las colecciones time series no admiten índices únicos, así que no puedes deduplicar las reentregas QoS 1 con una clave única. Si te importa almacenar cada mensaje exactamente una vez, escribe en una colección normal con un índice único sobre { "meta.deviceId": 1, seq: 1 } e ignora los errores de clave duplicada.

Prueba el pipeline

1. Arranca MongoDB y ejecuta node mqtt-to-mongo.mjs.
2. Suscríbete a testmqtt/mongo/# en el cliente MQTT online para ver el tráfico y publica unas cuantas lecturas:

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. Consulta MongoDB:

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 } }
])

Abre la shell con docker exec -it mongo mongosh. Los recuentos por dispositivo deberían sumar 30. Después prueba los modos de fallo: detén MongoDB con docker stop mongo, publica diez mensajes, vuelve a arrancarlo y comprueba que el escritor los volvió a encolar y los insertó.

Observa el topic de prueba de MongoDB en tu navegador

Solución de problemas

Para ejecutar varios escritores, suscríbete a $share/mongo-writers/testmqtt/mongo/+/telemetry con un client ID distinto por instancia; el broker reparte el flujo entre ellos (consulta suscripciones compartidas). Las versiones recientes de MongoDB crean automáticamente un índice sobre los campos meta y de tiempo de las nuevas colecciones time series; en versiones antiguas añade tú mismo db.telemetry.createIndex({ "meta.deviceId": 1, ts: -1 }). Si tu equipo prefiere SQL para analítica, compara este enfoque con MQTT a PostgreSQL/TimescaleDB.

MongoDB es un hogar natural para telemetría JSON si respetas tres reglas: un Date real en el campo de tiempo, un meta estable por dispositivo y escrituras por lotes. Añade una sesión persistente en tu propio broker y una prueba de reinicio, y el pipeline sobrevivirá a los despliegues sin que nadie lo note.

Preguntas frecuentes