Stocker des messages MQTT dans MongoDB avec Node.js (collections time series)

Pour stocker des messages MQTT dans MongoDB, faites tourner un consommateur Node.js qui s'abonne avec MQTT.js, convertit chaque payload JSON en document et écrit des lots avec insertMany dans une collection time series. MongoDB regroupe alors les mesures dans des buckets compressés par appareil et peut supprimer automatiquement les anciennes données.

Ce tutoriel construit ce consommateur pas à pas : création de la collection, une écriture par lots avec un buffer borné, une session MQTT persistante pour les redémarrages, ainsi que les requêtes et tests qui prouvent que chaque message est bien arrivé.

Pourquoi une collection time series

La télémétrie MQTT a une forme fixe : un horodatage, de quoi identifier la source et quelques mesures. Les collections time series de MongoDB (MongoDB 5.0+) sont conçues exactement pour cela.

Les options time series choisies à la création :

• timeField : le Date de la mesure (ts).
• metaField : ce qui identifie la série (meta.deviceId, meta.topic). Les mesures ayant la même valeur meta sont stockées ensemble : gardez-la stable par appareil.
• granularity : "seconds" pour des mesures toutes les quelques secondes ; "minutes" ou "hours" pour des appareils plus lents.
• expireAfterSeconds : suppression automatique des anciennes données.

Lancer MongoDB et créer la collection

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

Le consommateur crée la collection au premier démarrage : pas besoin d'étape de migration séparée.

Le consommateur

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 : créer la collection time series une seule fois
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, // conserver 30 jours
  });
}
const telemetry = db.collection("telemetry");

// 2. Écriture par lots
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}`); // données invalides : ne pas réessayer
    } 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 : session persistante, 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(); // protéger la mémoire, supprimer le plus ancien
  if (buffer.length >= BATCH_SIZE) flush();
});

// 4. Arrêt propre : stopper MQTT, puis écrire le reste
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);

Rôle de chaque partie :

• insertMany avec ordered: false envoie une requête par lot et continue d'insérer les autres documents si l'un est rejeté. Un insertOne par message est la raison la plus courante pour laquelle ces consommateurs prennent du retard.
• Le champ de temps est converti en Date. Les appareils envoient généralement des millisecondes depuis l'epoch ; un nombre brut dans ts fait échouer l'insertion.
• La gestion d'erreurs distingue données invalides et pannes. Un MongoBulkWriteError signifie que MongoDB a rejeté des documents : réessayer n'y changera rien. Les erreurs réseau ou de sélection de serveur remettent le lot en file pour le prochain flush.
• MAX_BUFFER plafonne la mémoire si MongoDB est indisponible longtemps. Supprimer les données les plus anciennes est un choix délibéré ; pour zéro perte, comptez sur la session du broker décrite plus bas plutôt que sur un buffer plus grand.

Si MQTT.js est nouveau pour vous, le guide MQTT en JavaScript couvre les options de connexion et les reconnexions, et le générateur de code produit un client de départ pour les paramètres de votre broker.

Redémarrages, sessions et doublons

clean: false avec sessionExpiryInterval et un clientId fixe permet au broker de conserver votre abonnement et de mettre en file les messages QoS 1 pendant que le consommateur est hors ligne, par exemple pendant un déploiement. À la reconnexion, le log affiche session present: true et les messages en attente arrivent. Chaque instance a besoin de son propre client ID ; générez-en un avec le générateur de client ID. Plus de contexte dans sessions persistantes.

Les endpoints publics de test partagés comme broker.connectmq.com ne conservent pas de session hors ligne : vérifiez le comportement au redémarrage sur votre propre instance privée.

Deux limites à connaître :

• MQTT.js acquitte un message QoS 1 dès que l'événement message a été émis, avant l'écriture du lot. Si le processus plante, les documents du buffer sont perdus. Si vous devez acquitter après l'écriture, MQTT.js vous permet de redéfinir client.handleMessage(packet, callback), qui n'envoie le PUBACK qu'à l'appel de callback ; mais les messages sont alors traités un par un, ce qui échange du débit contre de la sûreté.
• Les collections time series ne prennent pas en charge les index uniques : impossible de dédupliquer les relivraisons QoS 1 avec une clé unique. Si un stockage strictement unique compte, écrivez dans une collection classique avec un index unique sur { "meta.deviceId": 1, seq: 1 } et ignorez les erreurs de clé dupliquée.

Tester le pipeline

1. Démarrez MongoDB et lancez node mqtt-to-mongo.mjs.
2. Abonnez-vous à testmqtt/mongo/# dans le client MQTT en ligne pour suivre le trafic, puis publiez quelques mesures :

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

Ouvrez le shell avec docker exec -it mongo mongosh. Le total par appareil doit atteindre 30. Testez ensuite les cas de panne : arrêtez MongoDB avec docker stop mongo, publiez dix messages, redémarrez-le et vérifiez que le writer les a remis en file puis insérés.

Suivre le topic de test MongoDB dans le navigateur

Dépannage

Pour faire tourner plusieurs writers, abonnez-vous à $share/mongo-writers/testmqtt/mongo/+/telemetry avec un client ID distinct par instance ; le broker répartit le flux entre eux (voir abonnements partagés). Les versions récentes de MongoDB créent automatiquement un index sur les champs meta et temps des nouvelles collections time series ; sur les versions plus anciennes, ajoutez vous-même db.telemetry.createIndex({ "meta.deviceId": 1, ts: -1 }). Si votre équipe préfère SQL pour l'analyse, comparez cette approche avec MQTT vers PostgreSQL/TimescaleDB.

MongoDB est un foyer naturel pour la télémétrie JSON, à condition de respecter trois règles : un vrai Date dans le champ de temps, un meta stable par appareil et des écritures par lots. Ajoutez une session persistante sur votre propre broker et un test de redémarrage, et le pipeline survivra aux déploiements sans que personne ne le remarque.

Questions fréquentes