MQTT para MongoDB: Salve Mensagens de Dispositivos com Node.js e Coleções Time-Series

Salvar dados de MQTT para MongoDB é uma escolha natural quando seus dispositivos enviam JSON: os documentos entram como chegam, e as coleções time-series mantêm o armazenamento compacto e as consultas rápidas. Neste tutorial você vai construir um subscriber Node.js com MQTT.js que grava em lotes as mensagens dos dispositivos, vindas de um broker na nuvem, em uma coleção time-series do MongoDB, com expiração automática e consultas de agregação prontas para usar.

Como o Pipeline se Encaixa

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

A função do broker é rotear mensagens; ele não grava em bancos de dados. O subscriber é um cliente MQTT comum que você executa perto do seu banco de dados, o que mantém o design simples e portável. Qualquer broker MQTT gerenciado funciona do lado esquerdo; este tutorial usa o ConnectMQ.

Por que Coleções Time-Series

Uma coleção comum armazena cada leitura como um documento próprio, com seu próprio overhead. Uma coleção time-series agrupa internamente as leituras por um campo de metadados e uma janela de tempo, de modo que os dados de IoT ocupam muito menos espaço em disco e as consultas por intervalo de tempo varrem menos dados. Você também ganha expireAfterSeconds para excluir dados antigos automaticamente. As contrapartidas: os documentos são, na prática, somente de inserção (atualizações e exclusões são limitadas), e índices únicos não são suportados, então a deduplicação precisa acontecer antes da inserção, se você precisar dela.

Passo 1: Inicie o MongoDB

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

Passo 2: Crie a Coleção Time-Series

As mensagens dos dispositivos chegam em <prefix>/<site>/<deviceId>/telemetry com um payload como:

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

A coleção armazena cada mensagem como um documento: um campo de tempo ts, um objeto meta com os valores que identificam a série (local, dispositivo) e as medições. Crie-a uma única vez com um script de configuração.

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, // mantém 90 dias
  });
  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

Escolha a granularity de acordo com a frequência com que um único dispositivo reporta: seconds para intervalos menores que um minuto, minutes para leituras a cada poucos minutos, hours para dados esparsos. Um bom metaField contém valores que raramente mudam em uma série; nunca coloque as próprias medições em meta.

Passo 3: O Subscriber de MQTT para MongoDB

O subscriber mantém o callback MQTT leve: ele valida a mensagem, monta um documento e o coloca em um buffer em memória. Um timer esvazia o buffer com insertMany a cada segundo, ou imediatamente quando 500 documentos estão aguardando.

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; // nomes de campo inseguros
    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,                                   // sessão persistente
  properties: { sessionExpiryInterval: 1800 },    // o broker a mantém por 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);

Notas de Design

• Lotes com insertMany e ordered: false. Uma ida e volta por lote em vez de uma por mensagem, e um único documento inválido não interrompe o restante do lote.
• Um buffer limitado. Se o MongoDB estiver inacessível, os lotes que falharam voltam para a fila, mas o buffer tem um teto, para que a memória não cresça sem limite.
• Uma sessão persistente. clean: false com expiração de sessão e um client ID fixo diz ao broker para enfileirar mensagens QoS 1 enquanto o subscriber reinicia. O servidor MQTT público mantém sessões por até 30 minutos. Use um MQTT_CLIENT_ID estável, ou você terá uma sessão nova e vazia a cada inicialização. Saiba mais em sessões persistentes MQTT.
• Sanitização de campos. Chaves que começam com $ ou contêm pontos são ignoradas, e apenas valores primitivos são armazenados, para que um dispositivo malicioso ou com bug não consiga injetar estruturas estranhas nos seus documentos.
• Timestamps do dispositivo. Usar o ts do dispositivo mantém as leituras armazenadas offline no ponto certo do tempo; receivedAt registra quando o servidor as recebeu, o que ajuda a depurar problemas de relógio.

Formato dos Documentos: Um por Mensagem ou Um por Métrica?

O subscriber armazena um documento por mensagem MQTT, com cada medição como um campo de nível superior. Esse formato corresponde à forma como os dispositivos publicam, mantém valores relacionados juntos e torna triviais consultas como "temperatura e umidade no mesmo instante". Funciona bem quando cada tipo de dispositivo envia um conjunto estável de campos.

A alternativa é um documento por métrica: { ts, meta: { deviceId, metric }, value }. Ela se adapta a frotas em que os dispositivos reportam medições muito diferentes, porque todo documento tem a mesma estrutura e você pode indexar e agregar por meta.metric de forma genérica. O custo é ter mais documentos e precisar reagrupar valores que chegaram juntos.

Escolha um formato por coleção e mantenha-o. Misturar formatos em uma única coleção time-series torna as agregações complicadas e enfraquece os benefícios de armazenamento. Se estiver em dúvida, comece com um documento por mensagem; é mais fácil dividir depois do que juntar.

Passo 4: Teste com Dispositivos Simulados

Execute o subscriber e publique leituras de teste no servidor MQTT público (gratuito). A área pública é visível para todos, então use apenas dados sintéticos:

node subscriber.mjs

# em outro 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

Depois, veja os dados no mongosh:

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

Passo 5: Consulte os Dados

Médias por minuto de um dispositivo nas últimas 24 horas:

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

Última leitura de cada dispositivo, útil para uma página de visão geral da frota:

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

Para dashboards de longo prazo, agende um job que grave agregados por hora em uma coleção separada com $merge, e deixe a coleção bruta expirar após 90 dias.

Checklist de Produção

1. Crie um broker privado no portal do ConnectMQ e defina MQTT_USERNAME, MQTT_PASSWORD e MQTT_CLIENT_ID com os valores gerados; mantenha mqtts:// na porta 8883.
2. Dê ao subscriber suas próprias credenciais. O ConnectMQ vincula client IDs às credenciais emitidas, e dois processos com o mesmo ID desconectam um ao outro.
3. Execute o subscriber sob um gerenciador de processos (systemd, PM2, Docker com política de reinício).
4. Use um usuário do MongoDB com permissão de inserção apenas em iot.telemetry.
5. Aceite que QoS 1 pode entregar duplicatas; se isso importar, faça a deduplicação no subscriber (por exemplo, um pequeno cache em memória de IDs de mensagem), porque coleções time-series não conseguem garantir unicidade.
6. Para dividir a carga entre vários subscribers, use shared subscriptions do MQTT 5 ($share/mongo/<prefix>/+/+/telemetry), disponíveis nos planos Serverless pagos e superiores.
7. Acompanhe a taxa de inserção e o tamanho do buffer em logs ou métricas.

Solução de Problemas

Para uma alternativa SQL com agregados contínuos, veja MQTT para TimescaleDB; para necessidades mais simples, salvar dados MQTT no SQLite com Python ou MQTT para Google Sheets podem ser suficientes. O guia de bibliotecas cliente MQTT aborda o MQTT.js junto com outros clientes.

Comece a Usar o ConnectMQ

Conecte seu pipeline do MongoDB a um broker privado em poucos minutos. Cadastre-se em app.connectmq.com para obter um broker Serverless gratuito, sem cartão de crédito, exporte as credenciais geradas como variáveis de ambiente e use o explorador de topics ao vivo do portal para conferir os payloads antes que cheguem ao seu banco de dados.

Perguntas frequentes

O ConnectMQ pode armazenar mensagens MQTT no MongoDB automaticamente?

Não. O ConnectMQ é o broker MQTT e não tem integrações nativas com bancos de dados. Execute um subscriber, como o serviço Node.js deste tutorial ou um fluxo do Node-RED, que se conecte como cliente MQTT e grave no MongoDB.

Devo usar uma coleção time-series ou uma coleção comum para dados MQTT?

Use uma coleção time-series para leituras de sensores que só recebem inserções: ela as armazena de forma compacta e suporta expiração automática. Use uma coleção comum para dados que você atualiza no lugar, como a configuração do dispositivo ou o estado mais recente de cada dispositivo.

Como evitar perder mensagens MQTT quando o MongoDB está fora do ar?

Faça buffer no subscriber com uma fila limitada e repita as inserções que falharem, e use QoS 1 com uma sessão MQTT persistente para que o broker guarde as mensagens enquanto o subscriber reinicia.

O que o metaField faz em uma coleção time-series do MongoDB?

Ele identifica cada série, por exemplo o local e o ID do dispositivo. O MongoDB agrupa internamente as leituras com os mesmos metadados, e é por isso que as medições nunca devem ir para o metaField.