MQTT Verisini MongoDB'ye Kaydetme: Node.js ve Time-Series Koleksiyonlarıyla Cihaz Mesajları

Cihazlarınız JSON gönderiyorsa MQTT verisini MongoDB'ye kaydetmek doğal bir seçimdir: dokümanlar geldikleri gibi kaydedilir, time-series koleksiyonları da depolamayı kompakt ve sorguları hızlı tutar. Bu eğitimde MQTT.js ile, bir bulut broker'dan gelen cihaz mesajlarını otomatik süre dolumu ve hazır aggregation sorgularıyla bir MongoDB time-series koleksiyonuna toplu halde yazan bir Node.js subscriber geliştireceksiniz.

Pipeline Nasıl Bir Araya Gelir?

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

Broker'ın işi mesajları yönlendirmektir; veritabanlarına yazmaz. Subscriber, veritabanınızın yanında çalıştırdığınız sıradan bir MQTT client'tır; bu da tasarımı basit ve taşınabilir tutar. Sol tarafta herhangi bir tam yönetilen MQTT servisi çalışır; bu eğitim ConnectMQ kullanır.

Neden Time-Series Koleksiyonları?

Normal bir koleksiyon her okumayı kendi ek yüküyle ayrı bir doküman olarak saklar. Bir time-series koleksiyonu ise okumaları dahili olarak bir metadata alanına ve zaman penceresine göre gruplar; böylece IoT verisi çok daha az disk alanı kaplar ve zaman aralığı sorguları daha az veri tarar. Ayrıca eski verinin otomatik silinmesi için expireAfterSeconds elde edersiniz. Ödünleşimler: dokümanlar fiilen yalnızca eklemeye açıktır (güncellemeler ve silmeler sınırlıdır) ve benzersiz indeksler desteklenmez; bu yüzden tekilleştirmeye ihtiyacınız varsa bunu ekleme öncesinde yapmanız gerekir.

Adım 1: MongoDB'yi Başlatın

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

Adım 2: Time-Series Koleksiyonunu Oluşturun

Cihaz mesajları <prefix>/<site>/<deviceId>/telemetry topic'ine şuna benzer bir payload ile gelir:

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

Koleksiyon her mesajı tek bir doküman olarak saklar: bir ts zaman alanı, seriyi tanımlayan değerleri (saha, cihaz) içeren bir meta nesnesi ve ölçümler. Koleksiyonu bir kurulum betiğiyle bir kez oluşturun.

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, // 90 gün sakla
  });
  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

granularity değerini tek bir cihazın ne sıklıkla rapor verdiğine göre seçin: dakikanın altındaki aralıklar için seconds, birkaç dakikada bir gelen okumalar için minutes, seyrek veri için hours. İyi bir metaField, bir seri için nadiren değişen değerleri tutar; ölçümlerin kendisini asla meta içine koymayın.

Adım 3: MQTT'den MongoDB'ye Subscriber

Subscriber, MQTT callback'ini hafif tutar: mesajı doğrular, bir doküman oluşturur ve bunu bellekteki bir tampona ekler. Bir zamanlayıcı, tamponu her saniyede bir veya 500 doküman beklediği anda hemen insertMany ile boşaltır.

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; // güvensiz alan adları
    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,                                   // kalıcı oturum
  properties: { sessionExpiryInterval: 1800 },    // broker 30 dk saklar
  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);

Tasarım Notları

• insertMany ve ordered: false ile toplu yazma. Mesaj başına değil toplu iş başına bir gidiş-dönüş; tek bir hatalı doküman da toplu işin geri kalanını durdurmaz.
• Sınırlı bir tampon. MongoDB'ye ulaşılamazsa başarısız toplu işler yeniden kuyruğa alınır, ancak tampon sınırlandırılmıştır; böylece bellek sınırsızca büyüyemez.
• Kalıcı bir oturum. Bir oturum süresi ve sabit bir client ID ile birlikte clean: false, broker'a subscriber yeniden başlarken QoS 1 mesajlarını kuyrukta tutmasını söyler. Public broker oturumları en fazla 30 dakika saklar. Sabit bir MQTT_CLIENT_ID kullanın; aksi halde her başlatmada yeni ve boş bir oturum alırsınız. Ayrıntılar MQTT kalıcı oturumları yazısındadır.
• Alan temizleme. $ ile başlayan veya nokta içeren anahtarlar atlanır ve yalnızca ilkel değerler saklanır; böylece kötü niyetli veya hatalı bir cihaz dokümanlarınıza tuhaf yapılar enjekte edemez.
• Cihaz zaman damgaları. Cihazın ts değerini kullanmak, çevrimdışıyken tamponlanmış okumaları zaman içinde doğru yerde tutar; receivedAt ise sunucunun bunları ne zaman aldığını kaydeder, bu da saat sorunlarını ayıklamaya yardımcı olur.

Dokümanları Şekillendirme: Mesaj Başına mı, Metrik Başına mı?

Subscriber, her ölçümü üst düzey bir alan olarak içeren, MQTT mesajı başına bir doküman saklar. Bu şekil cihazların publish etme biçimiyle örtüşür, ilişkili değerleri bir arada tutar ve "aynı andaki sıcaklık ve nem" gibi sorguları önemsiz hale getirir. Her cihaz türü sabit bir alan seti gönderdiğinde iyi çalışır.

Alternatif, metrik başına bir dokümandır: { ts, meta: { deviceId, metric }, value }. Cihazların çok farklı ölçümler raporladığı filolara uyar, çünkü her doküman aynı yapıya sahiptir ve meta.metric üzerinde genel olarak indeksleme ve aggregation yapabilirsiniz. Bedeli daha fazla doküman ve birlikte gelen değerleri yeniden gruplama ihtiyacıdır.

Her koleksiyon için tek bir şekil seçin ve ona bağlı kalın. Tek bir time-series koleksiyonunda şekilleri karıştırmak aggregation'ları zorlaştırır ve depolama avantajlarını zayıflatır. Emin değilseniz mesaj başına bir dokümanla başlayın; sonradan bölmek, birleştirmekten daha kolaydır.

Adım 4: Simüle Edilmiş Cihazlarla Test Edin

Subscriber'ı çalıştırın ve ücretsiz public MQTT broker'ına test okumaları publish edin. Public alan herkes tarafından görülebilir; bu yüzden yapay veriyle sınırlı kalın:

node subscriber.mjs

# başka bir terminalde
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

Ardından veriye mongosh içinde bakın:

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

Adım 5: Veriyi Sorgulayın

Son 24 saatte tek bir cihaz için dakikalık ortalamalar:

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

Filo genel bakış sayfası için kullanışlı olan, cihaz başına en son okuma:

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

Uzun vadeli panolar için saatlik özetleri $merge ile ayrı bir koleksiyona yazan zamanlanmış bir iş kurun ve ham koleksiyonun 90 gün sonra süresinin dolmasına izin verin.

Üretim Kontrol Listesi

1. ConnectMQ portalında özel bir broker oluşturun ve MQTT_USERNAME, MQTT_PASSWORD ve MQTT_CLIENT_ID değerlerini üretilen değerlere ayarlayın; 8883 portunda mqtts:// kullanmaya devam edin.
2. Subscriber'a kendi kimlik bilgilerini verin. ConnectMQ client ID'leri verilen kimlik bilgilerine bağlar ve aynı ID'ye sahip iki süreç birbirinin bağlantısını keser.
3. Subscriber'ı bir süreç yöneticisi altında çalıştırın (systemd, PM2, yeniden başlatma politikalı Docker).
4. Yalnızca iot.telemetry üzerinde ekleme yetkisi olan bir MongoDB kullanıcısı kullanın.
5. QoS 1'in yinelenen mesajlar teslim edebileceğini kabul edin; bunlar önemliyse tekilleştirmeyi subscriber'da yapın (örneğin mesaj ID'lerinden oluşan kısa ömürlü bir bellek içi önbellek), çünkü time-series koleksiyonları benzersizliği zorunlu kılamaz.
6. Yükü birkaç subscriber arasında paylaştırmak için ücretli Serverless planları ve üzerinde bulunan MQTT 5 shared subscription'larını ($share/mongo/<prefix>/+/+/telemetry) kullanın.
7. Ekleme hızını ve tampon boyutunu günlüklerde veya metriklerde izleyin.

Sorun Giderme

Continuous aggregate'lere sahip SQL tabanlı bir alternatif için MQTT verisini TimescaleDB'ye kaydetme yazısına bakın; daha hafif ihtiyaçlar için Python ile MQTT verisini SQLite'a kaydetme veya MQTT'den Google Sheets'e yeterli olabilir. MQTT client kütüphaneleri rehberi, MQTT.js'i diğer client'larla birlikte ele alır.

ConnectMQ ile Başlayın

MongoDB pipeline'ınızı birkaç dakika içinde özel bir broker'a bağlayın. app.connectmq.com adresinde kredi kartı gerekmeden ücretsiz bir Serverless broker için kaydolun, üretilen kimlik bilgilerini ortam değişkenleri olarak dışa aktarın ve payload'ları veritabanınıza ulaşmadan önce portalın canlı topic explorer'ı ile kontrol edin.

Sık Sorulan Sorular

ConnectMQ, MQTT mesajlarını MongoDB'de otomatik olarak saklayabilir mi?

Hayır, ConnectMQ bir MQTT broker'dır ve yerleşik veritabanı entegrasyonları yoktur. MQTT client olarak bağlanıp MongoDB'ye yazan, bu eğitimdeki Node.js servisi veya bir Node-RED akışı gibi bir subscriber çalıştırın.

MQTT verisi için time-series koleksiyonu mu yoksa normal koleksiyon mu kullanmalıyım?

Yalnızca eklenen sensör okumaları için time-series koleksiyonu kullanın: bunları kompakt şekilde saklar ve otomatik süre dolumunu destekler. Cihaz yapılandırması veya cihaz başına en son durum gibi yerinde güncellediğiniz veriler için normal bir koleksiyon kullanın.

MongoDB kapalıyken MQTT mesajlarını kaybetmekten nasıl kaçınırım?

Subscriber'da sınırlı bir kuyrukla tamponlayın ve başarısız eklemeleri yeniden deneyin; ayrıca broker'ın subscriber yeniden başlarken mesajları tutması için QoS 1 ile kalıcı bir MQTT oturumu kullanın.

MongoDB time-series koleksiyonunda metaField ne işe yarar?

Her seriyi tanımlar; örneğin saha ve cihaz ID'si. MongoDB aynı metadata'ya sahip okumaları dahili olarak bir arada gruplar; ölçümlerin asla metaField içine konmaması gerekmesinin nedeni budur.