MQTT Mesajlarını Node.js ile MongoDB'ye Kaydedin (Time Series Koleksiyonları)

MQTT mesajlarını MongoDB'de saklamak için MQTT.js ile abone olan, her JSON payload'ı bir belgeye dönüştüren ve insertMany ile toplu olarak bir time series koleksiyonuna yazan bir Node.js tüketicisi çalıştırın. MongoDB ölçümleri cihaz başına sıkıştırılmış bucket'larda gruplar ve eski verileri otomatik olarak silebilir.

Bu eğitimde tüketiciyi adım adım oluşturacağız: koleksiyon kurulumu, sınırlı arabellekli bir toplu yazıcı, yeniden başlatmalar için kalıcı bir MQTT oturumu ve her mesajın ulaştığını kanıtlayan sorgular ve testler.

Neden time series koleksiyonu

MQTT telemetrisinin sabit bir yapısı vardır: bir zaman damgası, kaynağı tanımlayan bir şey ve birkaç ölçüm. MongoDB time series koleksiyonları (MongoDB 5.0+) tam olarak bunun için tasarlanmıştır.

Oluştururken seçtiğiniz time series seçenekleri:

• timeField: ölçümün Date değeri (ts).
• metaField: seriyi tanımlayan bilgi (meta.deviceId, meta.topic). Aynı meta değerine sahip ölçümler birlikte saklanır; bu yüzden cihaz başına sabit tutun.
• granularity: birkaç saniyede bir gelen ölçümler için "seconds"; daha yavaş cihazlar için "minutes" veya "hours".
• expireAfterSeconds: eski verilerin otomatik silinmesi.

MongoDB'yi çalıştırın ve koleksiyonu oluşturun

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

Tüketici koleksiyonu ilk açılışta oluşturur; ayrı bir migration adımına gerek yoktur.

Tüketici

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: time series koleksiyonunu bir kez oluştur
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, // 30 gün sakla
  });
}
const telemetry = db.collection("telemetry");

// 2. Toplu yazıcı
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}`); // hatalı veri: tekrar deneme
    } 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: kalıcı oturum, 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(); // belleği koru, en eskiyi at
  if (buffer.length >= BATCH_SIZE) flush();
});

// 4. Düzgün kapanış: önce MQTT'yi durdur, sonra kalanı yaz
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);

Her parçanın görevi:

• ordered: false ile insertMany toplu iş başına tek bir istek gönderir ve bir belge reddedilse bile kalanları eklemeye devam eder. Mesaj başına bir insertOne, bu tür tüketicilerin geride kalmasının en yaygın nedenidir.
• Zaman alanı Date türüne dönüştürülür. Cihazlar genellikle epoch'tan bu yana milisaniye gönderir; ts içindeki ham bir sayı insert'in başarısız olmasına yol açar.
• Hata yönetimi hatalı veriyi kesintiden ayırır. MongoBulkWriteError, MongoDB'nin belgeleri reddettiği anlamına gelir; tekrar denemek işe yaramaz. Ağ veya sunucu seçimi hataları ise toplu işi bir sonraki flush için yeniden kuyruğa alır.
• MAX_BUFFER, MongoDB uzun süre kapalı kalırsa bellek kullanımını sınırlar. En eski veriyi atmak bilinçli bir tercihtir; sıfır kayıp için daha büyük bir arabellek yerine aşağıda anlatılan broker oturumuna güvenin.

MQTT.js'de yeniyseniz JavaScript MQTT rehberi bağlantı seçeneklerini ve yeniden bağlanmayı anlatır; kod üreteci de broker ayarlarınız için başlangıç istemcisi oluşturur.

Yeniden başlatmalar, oturumlar ve tekrarlar

sessionExpiryInterval ve sabit bir clientId ile birlikte clean: false, broker'ın aboneliğinizi korumasını ve tüketici çevrimdışıyken (örneğin bir deploy sırasında) QoS 1 mesajlarını kuyruğa almasını sağlar. Yeniden bağlanıldığında logda session present: true görünür ve kuyruktaki mesajlar gelir. Her instance kendi client ID'sine ihtiyaç duyar; client ID üreteci ile bir tane oluşturun. Arka plan için kalıcı oturumlar yazısına bakın.

broker.connectmq.com gibi paylaşımlı herkese açık test endpoint'leri çevrimdışı oturum tutmaz; yeniden başlatma davranışını kendi özel instance'ınızda doğrulayın.

Bilmeniz gereken iki sınır:

• MQTT.js, bir QoS 1 mesajını message olayı tetiklenir tetiklenmez, yani toplu iş yazılmadan önce onaylar. Süreç çökerse arabellekteki belgeler kaybolur. Yazdıktan sonra onaylamanız gerekiyorsa MQTT.js'de client.handleMessage(packet, callback) metodunu ezebilirsiniz; PUBACK yalnızca callback çağrıldığında gönderilir. Ancak mesajlar tek tek işlenir; yani güvenlik için throughput'tan vazgeçersiniz.
• Time series koleksiyonları benzersiz indeksleri desteklemez; bu nedenle QoS 1 tekrar teslimlerini benzersiz bir anahtarla ayıklayamazsınız. Her mesajın tam olarak bir kez saklanması önemliyse { "meta.deviceId": 1, seq: 1 } üzerinde benzersiz indeksi olan normal bir koleksiyona yazın ve duplicate key hatalarını yok sayın.

Pipeline'ı test edin

1. MongoDB'yi başlatın ve node mqtt-to-mongo.mjs komutunu çalıştırın.
2. Trafiği izlemek için online MQTT istemcisinde testmqtt/mongo/# topic'ine abone olun, ardından birkaç ölçüm yayınlayın:

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. MongoDB'yi sorgulayın:

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

Shell'i docker exec -it mongo mongosh ile açın. Cihaz başına sayıların toplamı 30 olmalıdır. Ardından hata senaryolarını test edin: MongoDB'yi docker stop mongo ile durdurun, on mesaj yayınlayın, yeniden başlatın ve yazıcının bunları yeniden kuyruğa alıp eklediğini kontrol edin.

MongoDB test topic'ini tarayıcıda izleyin

Sorun giderme

Birden fazla yazıcı çalıştırmak için her instance'a ayrı bir client ID vererek $share/mongo-writers/testmqtt/mongo/+/telemetry topic'ine abone olun; broker akışı aralarında böler (bkz. paylaşımlı abonelikler). Güncel MongoDB sürümleri, yeni time series koleksiyonlarında meta ve zaman alanları üzerinde otomatik olarak bir indeks oluşturur; eski sürümlerde db.telemetry.createIndex({ "meta.deviceId": 1, ts: -1 }) indeksini kendiniz ekleyin. Ekibiniz analiz için SQL'i tercih ediyorsa bu yaklaşımı MQTT'den PostgreSQL/TimescaleDB'ye ile karşılaştırın.

Üç kurala uyduğunuz sürece MongoDB, JSON telemetrisi için doğal bir yuvadır: zaman alanında gerçek bir Date, cihaz başına sabit bir meta ve toplu yazma. Kendi broker'ınızda kalıcı bir oturum ve bir yeniden başlatma testi ekleyin; pipeline deploy'ları kimse fark etmeden atlatır.

Sıkça sorulan sorular