Armazene mensagens MQTT no MongoDB com Node.js (coleções time series)

Para armazenar mensagens MQTT no MongoDB, rode um consumidor Node.js que assina com MQTT.js, transforma cada payload JSON em um documento e grava lotes com insertMany em uma coleção time series. O MongoDB então agrupa as leituras em buckets comprimidos por dispositivo e pode apagar dados antigos automaticamente.

Este tutorial constrói esse consumidor passo a passo: criação da coleção, um gravador em lote com buffer limitado, uma sessão MQTT persistente para reinícios, e as consultas e testes que provam que cada mensagem chegou.

Por que uma coleção time series

A telemetria MQTT tem um formato fixo: um timestamp, algo que identifica a origem e algumas medições. As coleções time series do MongoDB (MongoDB 5.0+) foram feitas exatamente para isso.

As opções time series definidas na criação:

• timeField: o Date da leitura (ts).
• metaField: o que identifica a série (meta.deviceId, meta.topic). Leituras com o mesmo valor de meta são armazenadas juntas, então mantenha-o estável por dispositivo.
• granularity: "seconds" para leituras a cada poucos segundos; "minutes" ou "hours" para dispositivos mais lentos.
• expireAfterSeconds: exclusão automática de dados antigos.

Rode o MongoDB e crie a coleção

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

O consumidor cria a coleção na primeira inicialização, então você não precisa de uma etapa de migração separada.

O 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: criar a coleção time series uma única 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, // manter 30 dias
  });
}
const telemetry = db.collection("telemetry");

// 2. Gravador em lote
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}`); // dados inválidos: não repetir
    } 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: sessão 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 a memória, descartar o mais antigo
  if (buffer.length >= BATCH_SIZE) flush();
});

// 4. Desligamento limpo: parar o MQTT e depois gravar o que sobrou
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);

O que cada parte faz:

• insertMany com ordered: false envia uma requisição por lote e continua inserindo os demais documentos se um for rejeitado. Um insertOne por mensagem é o motivo mais comum para esses consumidores ficarem para trás.
• O campo de tempo é convertido em Date. Dispositivos costumam enviar milissegundos desde o epoch; um número bruto em ts faz o insert falhar.
• O tratamento de erros separa dados inválidos de quedas. Um MongoBulkWriteError significa que o MongoDB rejeitou documentos: repetir não ajuda. Erros de rede ou de seleção de servidor recolocam o lote na fila para o próximo flush.
• MAX_BUFFER limita a memória se o MongoDB ficar fora do ar por muito tempo. Descartar os dados mais antigos é uma escolha deliberada; para perda zero, conte com a sessão do broker descrita abaixo em vez de um buffer maior.

Se o MQTT.js é novo para você, o guia de MQTT com JavaScript cobre opções de conexão e reconexões, e o gerador de código gera um cliente inicial com as configurações do seu broker.

Reinícios, sessões e duplicatas

clean: false com sessionExpiryInterval e um clientId fixo faz o broker manter sua assinatura e enfileirar mensagens QoS 1 enquanto o consumidor está offline, por exemplo durante um deploy. Na reconexão, o log mostra session present: true e as mensagens enfileiradas chegam. Cada instância precisa do seu próprio client ID; gere um com o gerador de client ID. Mais contexto em sessões persistentes.

Endpoints públicos de teste compartilhados como broker.connectmq.com não guardam sessões offline, então verifique o comportamento de reinício na sua própria instância privada.

Dois limites que você deve conhecer:

• O MQTT.js confirma uma mensagem QoS 1 assim que o evento message é disparado, antes de o lote ser gravado. Se o processo cair, os documentos no buffer se perdem. Se você precisa confirmar depois de gravar, o MQTT.js permite sobrescrever client.handleMessage(packet, callback), que só envia o PUBACK quando você chama callback; mas ele processa as mensagens uma de cada vez, trocando throughput por segurança.
• Coleções time series não suportam índices únicos, então não dá para deduplicar reentregas QoS 1 com uma chave única. Se armazenar cada mensagem exatamente uma vez for importante, grave em uma coleção normal com índice único em { "meta.deviceId": 1, seq: 1 } e ignore os erros de chave duplicada.

Teste o pipeline

1. Suba o MongoDB e rode node mqtt-to-mongo.mjs.
2. Assine testmqtt/mongo/# no cliente MQTT online para acompanhar o tráfego e publique algumas leituras:

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

Abra o shell com docker exec -it mongo mongosh. As contagens por dispositivo devem somar 30. Depois teste os modos de falha: pare o MongoDB com docker stop mongo, publique dez mensagens, inicie-o de novo e verifique se o gravador recolocou na fila e inseriu essas mensagens.

Acompanhe o topic de teste do MongoDB no navegador

Solução de problemas

Para rodar vários gravadores, assine $share/mongo-writers/testmqtt/mongo/+/telemetry com um client ID diferente por instância; o broker divide o fluxo entre eles (veja assinaturas compartilhadas). Versões recentes do MongoDB criam automaticamente um índice nos campos meta e de tempo das novas coleções time series; em versões antigas, adicione você mesmo db.telemetry.createIndex({ "meta.deviceId": 1, ts: -1 }). Se sua equipe prefere SQL para análises, compare esta abordagem com MQTT para PostgreSQL/TimescaleDB.

O MongoDB é um lar natural para telemetria em JSON, desde que você respeite três regras: um Date de verdade no campo de tempo, um meta estável por dispositivo e gravações em lote. Adicione uma sessão persistente no seu próprio broker e um teste de reinício, e o pipeline vai sobreviver aos deploys sem ninguém perceber.

Perguntas frequentes