MQTT.js Node.js Eğitimi: Publish, Subscribe, QoS ve TLS

MQTT.js, JavaScript için standart MQTT client'ıdır ve MQTT.js'i Node.js'te kullanmak IoT backend'leri, köprüler ve test araçları geliştirmenin en hızlı yoludur. Bu eğitimde MQTT.js v5 ile bir Node.js publisher ve subscriber oluşturacak, promise tabanlı API'sini kullanacak, QoS seviyelerini seçecek, bir Last Will ekleyecek, TLS'e geçecek ve uzun süre çalışan bir servis için yeniden bağlanmayı yapılandıracaksınız. Tüm örnekler ücretsiz public broker broker.connectmq.com üzerinde çalışır; bu yüzden yalnızca Node.js kurulu olması yeterlidir.

Ön Koşullar

• Node.js 18 veya üzeri (node --version)
• npm, pnpm veya yarn
• İsteğe bağlı: trafiği ikinci bir terminalden izlemek için mosquitto_sub

Bir proje oluşturun ve kütüphaneyi kurun:

mkdir mqtt-node-demo && cd mqtt-node-demo
npm init -y
npm pkg set type=module
npm install mqtt

"type": "module" ayarı import sözdizimini ve üst düzey await kullanmanızı sağlar. MQTT.js v5 TypeScript ile yazılmıştır ve kendi tip tanımlarını içerir; böylece editörler ek paket gerektirmeden otomatik tamamlama sunar.

Bağlantı URL'leri ve Portlar

MQTT.js taşıma katmanını URL şemasından seçer. ConnectMQ broker'larında portlar şunlardır:

Node.js'te 8883 portunda mqtts:// tercih edin. WebSocket, ham TCP soketi açamayan tarayıcılar için önemlidir; bu tarafı WebSocket üzerinden MQTT yazısı ele alır.

İlk MQTT.js Subscriber'ınız

subscriber.js olarak kaydedin:

import mqtt from "mqtt";

const PREFIX = "connectmq/demo/node-a71c";
const clientId = `node-sub-${Math.random().toString(16).slice(2, 10)}`;

const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId,
  protocolVersion: 5,
  clean: true,
  keepalive: 30,
  reconnectPeriod: 2000,   // yeniden bağlanma denemeleri arasında 2 sn bekle
  connectTimeout: 10000,   // tek bir denemeden 10 sn sonra vazgeç
});

client.on("connect", async (connack) => {
  console.log(`Connected as ${clientId}, session present: ${connack.sessionPresent}`);
  const granted = await client.subscribeAsync(`${PREFIX}/+/telemetry`, { qos: 1 });
  console.log("Subscribed:", granted.map((g) => `${g.topic} (qos ${g.qos})`).join(", "));
});

client.on("message", (topic, payload, packet) => {
  let data;
  try {
    data = JSON.parse(payload.toString());
  } catch {
    data = payload.toString();
  }
  const device = topic.split("/").at(-2);
  console.log(`[${device}] qos=${packet.qos} retain=${packet.retain}`, data);
});

client.on("reconnect", () => console.log("Reconnecting..."));
client.on("offline", () => console.log("Offline"));
client.on("error", (err) => console.error("MQTT error:", err.message));

process.on("SIGINT", async () => {
  await client.endAsync();
  process.exit(0);
});

connect handler'ı içinde abone olmak bilinçli bir tercihtir. clean: true ile broker bağlantı koptuğunda abonelikleri unutur; bu nedenle her bağlantıda yeniden abone olmak, yeniden bağlandıktan sonra mesaj almaya devam etmenizi garanti eder.

QoS ve Last Will ile Yayın Yapmak

publisher.js olarak kaydedin:

import mqtt from "mqtt";

const PREFIX = "connectmq/demo/node-a71c";
const DEVICE = "sensor-01";
const clientId = `node-pub-${Math.random().toString(16).slice(2, 10)}`;

const client = await mqtt.connectAsync("mqtts://broker.connectmq.com:8883", {
  clientId,
  protocolVersion: 5,
  will: {
    topic: `${PREFIX}/${DEVICE}/status`,
    payload: Buffer.from("offline"),
    qos: 1,
    retain: true,
  },
});

console.log("Connected");
await client.publishAsync(`${PREFIX}/${DEVICE}/status`, "online", { qos: 1, retain: true });

let count = 0;
const timer = setInterval(async () => {
  const reading = {
    temperature: +(20 + Math.random() * 5).toFixed(2),
    humidity: +(40 + Math.random() * 10).toFixed(1),
    ts: Date.now(),
  };
  try {
    await client.publishAsync(`${PREFIX}/${DEVICE}/telemetry`, JSON.stringify(reading), {
      qos: 1,
      properties: { messageExpiryInterval: 60, contentType: "application/json" },
    });
    console.log("Published", reading);
  } catch (err) {
    console.error("Publish failed:", err.message);
  }
  if (++count === 20) {
    clearInterval(timer);
    await client.publishAsync(`${PREFIX}/${DEVICE}/status`, "offline", { qos: 1, retain: true });
    await client.endAsync();
  }
}, 2000);

İkisini de çalıştırın:

node subscriber.js
# ikinci bir terminalde
node publisher.js

Burada birkaç şey oluyor:

• connectAsync, broker CONNACK gönderdiğinde çözümlenir; bu da betiklerde pratiktir. Uzun süre çalışan servislerde olay tabanlı connect yine de faydalıdır, çünkü her yeniden bağlantıdan sonra tekrar tetiklenir.
• QoS 1, broker'ın her yayını onayladığı anlamına gelir ve publishAsync ancak bu onaydan sonra çözümlenir. Ara sıra kaybın sorun olmadığı yüksek hızlı telemetri için QoS 0 daha ucuzdur. Ödünleşimler MQTT QoS seviyeleri yazısında ele alınır.
• Last Will, bağlantı sırasında broker tarafından saklanır ve client temiz bir bağlantı kesme olmadan kaybolursa yayınlanır; böylece diğer servisler sensörün çevrimdışı olduğunu öğrenir. Onu retained yapmak, yeni abonelerin güncel durumu hemen görmesini sağlar.
• MQTT 5 özellikleri, örneğin messageExpiryInterval, broker'a mesajı 60 saniye içinde teslim edemezse atmasını söyler; bu da çevrimdışı aboneler için eski ölçümlerin birikmesini önler.

Ücretsiz public MQTT broker'ın public alanı herkes tarafından okunabilir; bu yüzden önekteki node-a71c kısmını kendi rastgele dizenizle değiştirin ve oraya asla gerçek veri göndermeyin.

Önemli Yeniden Bağlanma Seçenekleri

MQTT.js otomatik olarak yeniden bağlanır. Davranışını şu seçenekler belirler:

Kısa süreli çevrimdışı kalırken mesaj kaçırmaması gereken bir servis için sabit bir client ID, clean: false, QoS 1 abonelikleri ve bir oturum sona erme süresini birleştirin. Broker bu durumda oturum için mesajları kuyruğa alır ve client geri döndüğünde teslim eder.

Uzun Süre Çalışan Bir Node.js Servisini Yapılandırmak

Yukarıdaki betikler bilinçli olarak kısadır. Aylarca çalışan bir servis biraz daha fazla yapı gerektirir ve bunun büyük kısmı MQTT'nin kendisinden çok yaşam döngüsüyle ilgilidir:

• Süreç başına bir client oluşturun ve onu bir modül üzerinden paylaşın. Her istek veya iş için yeni bir bağlantı açmak kaynak israfıdır ve bağlantı sınırlarına hızla takılır.
• connect olayını "hazır" olarak kabul edin. Broker gerektiren işleri client'ın connected özelliğine bağlayın ve bu değer false olduğunda ne yapılacağına karar verin: tamponlamak, atmak veya çağırana hata döndürmek.
• Kapatma sinyallerini ele alın. SIGTERM geldiğinde (Docker, Kubernetes ve systemd tarafından gönderilir) kullanıyorsanız son bir durum yayınlayın, iletimdeki QoS 1 mesajlarının boşaltılması için endAsync() çağırın ve ancak ondan sonra çıkın.
• Her mesajı değil, durum geçişlerini loglayın. Bir kesintiyi ayıklarken ihtiyacınız olan şey connect, reconnect, offline, close ve error olaylarıdır. Mesaj başına loglar bunları boğar.
• Handler'ları hızlı tutun. message olayı Node'un event loop'unda çalışır. Veritabanı yazma gibi ağır işler dikkatle await edilmeli veya bir kuyruğa itilmelidir; böylece yavaş bir alt sistem diğer tüm mesajları geciktirmez.

Bu alışkanlıklar herhangi bir seçenek değerinden daha önemlidir ve MQTT.js üzerine kurulan köprüler, worker'lar ve test araçları için eşit derecede geçerlidir.

Private Bir Broker ile Kimlik Doğrulama

Üretim için broker'ınızdan aldığınız kimlik bilgilerini kullanın. ConnectMQ ile portalda bir broker oluşturur ve bir kullanıcı adı, parola ve client ID alırsınız. Bunları ortam değişkenlerinden yükleyin:

import mqtt from "mqtt";

const client = mqtt.connect(`mqtts://${process.env.MQTT_HOST}:8883`, {
  clientId: process.env.MQTT_CLIENT_ID,   // portaldaki YOUR_CLIENT_ID
  username: process.env.MQTT_USERNAME,    // YOUR_USERNAME
  password: process.env.MQTT_PASSWORD,    // YOUR_PASSWORD
  protocolVersion: 5,
  clean: false,
  properties: { sessionExpiryInterval: 3600 },
});

client.on("connect", () => console.log("Connected to private broker"));
client.on("error", (err) => console.error(err.message));

ConnectMQ, Node.js'in kutudan çıktığı haliyle güvendiği bir Let's Encrypt sertifikası kullanır; bu yüzden bir CA dosyası vermeniz gerekmez. Üretimde asla rejectUnauthorized: false ayarlamayın; bu, sertifika kontrollerini devre dışı bırakır ve TLS'i anlamsız hale getirir.

Node.js'te MQTT.js Sorun Giderme

1. Bağlanıyor, sonra birkaç saniyede bir bağlantı kopuyor. Başka bir client aynı client ID'yi kullanıyor. Çalışan her sürecin kendi ID'si olmalıdır.
2. ERR_REQUIRE_ESM veya Cannot use import statement. package.json içinde "type": "module" ayarlayın veya CommonJS dosyalarında const mqtt = require("mqtt") kullanın.
3. Ağ koptuktan sonra mesaj gelmiyor. clean: true kullanırken connect handler'ı dışında yalnızca bir kez abone olmuşsunuz.
4. connack timeout. Bir güvenlik duvarı 8883 portunu engelliyor. mosquitto_sub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs -i test-1 -t test ile test edin veya 8084 portunda wss:// deneyin.
5. Abonelik QoS 128 veya 0x80 ya da üzeri bir reason code ile verildi. Broker aboneliği reddetti; bu genellikle geçersiz bir filtre veya public broker'daki abonelik sınırından (client başına 10) kaynaklanır.
6. Çevrimdışıyken bellek büyüyor. Giden mesajlar yeniden bağlanana kadar kuyrukta tutulur. Kendi tamponunuza sınır koyun veya önemsiz telemetri için queueQoSZero: false ile QoS 0 kullanın.

Sonraki Adımlar

Buradan aynı client'ı bir frontend'e taşıyabilirsiniz: React ile MQTT, MQTT.js'in tarayıcı derlemesi üzerinde hook tabanlı bir pano oluşturur; TypeScript ile tip güvenli MQTT ise bu kodun üzerine tipli topic'ler ve doğrulanmış payload'lar ekler. Diğer çalışma ortamları için MQTT client kütüphaneleri rehberi başlıca seçenekleri karşılaştırır. Seçeneklerin tam listesi MQTT.js README dosyasındadır.

ConnectMQ ile Başlayın

Node.js servislerinizi dakikalar içinde public test broker'dan private bir broker'a taşıyın. app.connectmq.com adresinden kaydolun, kredi kartı gerektirmeyen ücretsiz Serverless planını seçin, bir broker oluşturun ve üretilen kimlik bilgilerini ortamınıza ekleyin. Yönetilen bir MQTT broker ile siz uygulama kodunuza odaklanırken TLS uç noktalarını ve broker altyapısını ConnectMQ işletir.

Sık Sorulan Sorular

MQTT.js v5 eski kodla uyumlu mu?

Büyük ölçüde evet. Callback API'si hâlâ çalışır ve v5, connectAsync, publishAsync, subscribeAsync ve endAsync gibi promise tabanlı metotlar ekler. Başlıca değişiklikler TypeScript ile yeniden yazım ve daha yeni bir minimum Node.js sürümüdür.

MQTT.js ile MQTT 3.1.1 mi yoksa MQTT 5 mi kullanmalıyım?

Broker'ınız destekliyorsa (ConnectMQ'nun desteklediği gibi) protocolVersion: 5 kullanın. Publish ve subscribe kodu aynı kalırken reason code'lar, mesaj sona erme süresi, oturum sona erme süresi ve user property'ler elde edersiniz.

Bir Node.js servisi kaç bağlantı açmalı?

Genellikle bir. Tek bir MQTT.js client'ı aynı anda birçok topic'e yayın yapabilir ve abone olabilir; her ek bağlantı kendi benzersiz client ID'sini gerektirir ve planınızın bağlantı sınırına dahil edilir.

MQTT.js'i Bun veya Deno ile kullanabilir miyim?

MQTT.js, Node.js ve tarayıcıları hedefler. Node.js uyumluluğu iyi olan çalışma ortamları onu çoğu zaman çalıştırabilir, ancak ona güvenmeden önce TLS ve yeniden bağlanma davranışını dikkatle test edin.