MQTT vers MongoDB : enregistrer les messages d'appareils avec Node.js et les collections time-series

Enregistrer MQTT vers MongoDB est un choix naturel lorsque vos appareils envoient du JSON : les documents entrent tels qu'ils arrivent, et les collections time-series gardent un stockage compact et des requêtes rapides. Dans ce tutoriel, vous allez construire un subscriber Node.js avec MQTT.js qui insère par lots les messages des appareils provenant d'un broker cloud dans une collection time-series MongoDB, avec expiration automatique et requêtes d'agrégation prêtes à l'emploi.

Comment s'articule le pipeline

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

Le rôle du broker est de router les messages ; il n'écrit pas dans les bases de données. Le subscriber est un client MQTT ordinaire que vous exécutez à côté de votre base, ce qui garde une conception simple et portable. N'importe quel broker MQTT managé convient côté gauche ; ce tutoriel utilise ConnectMQ.

Pourquoi les collections time-series

Une collection classique stocke chaque relevé dans son propre document, avec son propre surcoût. Une collection time-series regroupe en interne les relevés par champ de métadonnées et par fenêtre temporelle : les données IoT occupent donc beaucoup moins d'espace disque et les requêtes par plage temporelle parcourent moins de données. Vous bénéficiez aussi de expireAfterSeconds pour la suppression automatique des anciennes données. Les compromis : les documents sont de fait en ajout seul (mises à jour et suppressions limitées), et les index uniques ne sont pas pris en charge ; la déduplication doit donc se faire avant l'insertion si vous en avez besoin.

Étape 1 : démarrer MongoDB

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

Étape 2 : créer la collection time-series

Les messages des appareils arrivent sur <prefix>/<site>/<deviceId>/telemetry avec un payload comme celui-ci :

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

La collection stocke chaque message sous forme d'un document : un champ temporel ts, un objet meta contenant les valeurs qui identifient la série (site, appareil), et les mesures. Créez-la une fois avec un script d'initialisation.

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, // conserver 90 jours
  });
  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

Choisissez la granularity selon la fréquence de remontée d'un appareil : seconds pour des intervalles inférieurs à la minute, minutes pour des relevés toutes les quelques minutes, hours pour des données clairsemées. Un bon metaField contient des valeurs qui changent rarement pour une série ; ne placez jamais les mesures elles-mêmes dans meta.

Étape 3 : le subscriber MQTT vers MongoDB

Le subscriber garde le callback MQTT léger : il valide le message, met en forme un document et le place dans un tampon en mémoire. Un timer vide le tampon avec insertMany toutes les secondes, ou immédiatement dès que 500 documents sont en attente.

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; // noms de champs dangereux
    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,                                   // session persistante
  properties: { sessionExpiryInterval: 1800 },    // le broker la conserve 30 min
  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);

Notes de conception

• Regroupement avec insertMany et ordered: false. Un aller-retour par lot au lieu d'un par message, et un document invalide n'arrête pas le reste du lot.
• Un tampon borné. Si MongoDB est injoignable, les lots en échec sont remis en file, mais le tampon est plafonné pour que la mémoire ne croisse pas sans limite.
• Une session persistante. clean: false avec une expiration de session et un client ID fixe indique au broker de conserver en file les messages QoS 1 pendant le redémarrage du subscriber. Le serveur MQTT public gratuit conserve les sessions jusqu'à 30 minutes. Utilisez un MQTT_CLIENT_ID stable, sinon vous obtenez une session neuve et vide à chaque démarrage. Plus d'informations dans les sessions persistantes MQTT.
• Assainissement des champs. Les clés commençant par $ ou contenant des points sont ignorées, et seules les valeurs primitives sont stockées : un appareil malveillant ou bogué ne peut donc pas injecter de structures étranges dans vos documents.
• Horodatages des appareils. Utiliser le ts de l'appareil place correctement dans le temps les relevés mis en tampon hors ligne ; receivedAt enregistre le moment de réception par le serveur, ce qui aide à diagnostiquer les problèmes d'horloge.

Mettre en forme les documents : un par message ou un par métrique ?

Le subscriber stocke un document par message MQTT, avec chaque mesure en champ de premier niveau. Cette forme correspond à la façon dont les appareils publient, garde les valeurs liées ensemble et rend triviales des requêtes comme « température et humidité au même instant ». Elle fonctionne bien lorsque chaque type d'appareil envoie un ensemble stable de champs.

L'alternative est un document par métrique : { ts, meta: { deviceId, metric }, value }. Elle convient aux flottes où les appareils remontent des mesures très différentes, car chaque document a la même structure et vous pouvez indexer et agréger génériquement sur meta.metric. Le coût : davantage de documents et la nécessité de regrouper les valeurs arrivées ensemble.

Choisissez une forme par collection et tenez-vous-y. Mélanger les formes dans une même collection time-series complique les agrégations et réduit les bénéfices de stockage. En cas de doute, commencez par un document par message ; il est plus facile de scinder plus tard que de fusionner.

Étape 4 : tester avec des appareils simulés

Lancez le subscriber et publiez des relevés de test sur le serveur MQTT public gratuit, réservé aux tests et aux prototypes. La zone publique est visible par tous : contentez-vous de données synthétiques :

node subscriber.mjs

# dans un autre 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

Examinez ensuite les données dans mongosh :

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

Étape 5 : interroger les données

Moyennes par minute pour un appareil sur les dernières 24 heures :

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

Dernier relevé par appareil, utile pour une page de vue d'ensemble de la flotte :

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

Pour les tableaux de bord à long terme, planifiez un job qui écrit des cumuls horaires dans une collection séparée avec $merge, et laissez la collection brute expirer au bout de 90 jours.

Liste de contrôle pour la production

1. Créez un broker privé dans le portail ConnectMQ et définissez MQTT_USERNAME, MQTT_PASSWORD et MQTT_CLIENT_ID avec les valeurs générées ; conservez mqtts:// sur le port 8883.
2. Donnez au subscriber ses propres identifiants. ConnectMQ lie les client IDs aux identifiants émis, et deux processus avec le même ID se déconnectent mutuellement.
3. Exécutez le subscriber sous un gestionnaire de processus (systemd, PM2, Docker avec politique de redémarrage).
4. Utilisez un utilisateur MongoDB disposant uniquement des droits d'insertion sur iot.telemetry.
5. Acceptez que le QoS 1 puisse livrer des doublons ; s'ils comptent, dédupliquez dans le subscriber (par exemple avec un petit cache en mémoire des ID de messages), car les collections time-series ne peuvent pas imposer l'unicité.
6. Pour répartir la charge entre plusieurs subscribers, utilisez les abonnements partagés MQTT 5 ($share/mongo/<prefix>/+/+/telemetry), disponibles à partir des plans Serverless payants.
7. Surveillez le débit d'insertion et la taille du tampon dans les logs ou les métriques.

Dépannage

Pour une alternative SQL avec agrégats continus, consultez MQTT vers TimescaleDB ; pour des besoins plus légers, enregistrer des données MQTT dans SQLite avec Python ou MQTT vers Google Sheets peuvent suffire. Le guide des bibliothèques clientes MQTT présente MQTT.js aux côtés d'autres clients.

Démarrer avec ConnectMQ

Connectez votre pipeline MongoDB à un broker privé en quelques minutes. Inscrivez-vous sur app.connectmq.com pour obtenir un broker Serverless gratuit, sans carte bancaire, exportez les identifiants générés comme variables d'environnement et utilisez l'explorateur de topics en direct du portail pour vérifier les payloads avant qu'ils n'atteignent votre base de données.

Questions fréquentes

ConnectMQ peut-il stocker automatiquement les messages MQTT dans MongoDB ?

Non, ConnectMQ est le broker MQTT et ne dispose d'aucune intégration native avec des bases de données. Exécutez un subscriber, comme le service Node.js de ce tutoriel ou un flow Node-RED, qui se connecte en tant que client MQTT et écrit dans MongoDB.

Faut-il une collection time-series ou une collection classique pour les données MQTT ?

Utilisez une collection time-series pour les relevés de capteurs en ajout seul : elle les stocke de manière compacte et prend en charge l'expiration automatique. Utilisez une collection classique pour les données que vous mettez à jour sur place, comme la configuration des appareils ou le dernier état de chaque appareil.

Comment éviter de perdre des messages MQTT quand MongoDB est indisponible ?

Mettez en tampon dans le subscriber avec une file bornée et réessayez les insertions en échec, et utilisez le QoS 1 avec une session MQTT persistante pour que le broker conserve les messages pendant le redémarrage du subscriber.

À quoi sert le metaField dans une collection time-series MongoDB ?

Il identifie chaque série, par exemple le site et l'ID de l'appareil. MongoDB regroupe en interne les relevés partageant les mêmes métadonnées, c'est pourquoi les mesures ne doivent jamais aller dans le metaField.