Node-RED Dinamik MQTT Abonelik: Topic'leri Çalışma Anında Değiştirin

Node-RED dinamik MQTT abonelik modu, mqtt in node'una tek bir topic'i sabit kodlamak yerine flow'un çalışırken hangi topic'leri dinleyeceğine karar vermesini sağlar. Bu eğitimde dinamik bir subscriber yapılandıracak, msg.action ile topic ekleyip kaldıracak, aktif abonelikleri listeleyecek, bunları yeniden başlatmadan sonra geri yükleyecek ve tüm yapıyı basit bir cihaz listesinden yöneteceksiniz.

Dinamik Aboneliklere Ne Zaman İhtiyacınız Olur?

Normal bir mqtt in node'u deploy anında sabit bir topic filtresine subscribe olur. Bu, özellikle site/+/telemetry gibi bir wildcard zaten her cihazla eşleştiğinde çoğu flow'u karşılar. Dinamik abonelikler şu durumlarda ek bağlantı zahmetine değer:

• Bir operatör bir panodan hangi makineyi izleyeceğini seçer ve yalnızca o makinenin verisinin akmasını istersiniz.
• Cihaz listesi bir veritabanında veya yapılandırma topic'inde durur ve yeniden deploy etmeden değişir.
• Bir wildcard ihtiyacınızdan çok daha fazla trafik çekerdi; örneğin yoğun bir sahada #.
• Sorun giderme sırasında geçici olarak bir debug topic'ini izleyip sonra bırakmak istersiniz.
• Bir broker'ın abonelik sınırına yaklaşıyorsunuz ve seçici olmanız gerekiyor.

Son madde paylaşımlı altyapıda önemlidir. Örneğin broker.connectmq.com adresindeki ücretsiz herkese açık MQTT broker istemci başına 10 aboneliğe izin verir; bu yüzden talep üzerine subscribe olmak bütçe içinde kalmanın pratik bir yoludur.

mqtt in Node'unu Dinamik Mod İçin Yapılandırın

Dinamik abonelikler Node-RED 2.2 ile geldi ve tüm güncel sürümlerde bulunur. Etkinleştirmek için:

1. Çalışma alanına bir mqtt in node'u sürükleyin ve açın.
2. Bir broker config node'u seçin veya oluşturun. Test için sunucu broker.connectmq.com, port 1883 ve nr-dyn-91c4 gibi benzersiz, boş olmayan bir client ID kullanın.
3. Action ayarını "Subscribe to single topic" yerine Dynamic subscription yapın.
4. Output ayarını her zamanki gibi "auto-detect" veya "a parsed JSON object" olarak belirleyin.
5. Deploy edin.

Node'un artık bir giriş portu vardır ve durumu "connected" gösterir, ancak henüz hiçbir şeye subscribe değildir. Bundan sonraki her şey bu girişe gönderdiğiniz mesajlarla kontrol edilir.

Broker config node'unda bir Connect automatically seçeneği de göreceksiniz. İleride ele alınan connect ve disconnect eylemleriyle bağlantının kendisini de kontrol etmek istemiyorsanız işaretli bırakın.

Node-RED Dinamik MQTT Abonelik Eylemleri

Node, msg.action ve msg.topic değerlerine tepki verir:

Nesne biçimi, broker node'u protokol sürümü 5'i kullandığında nl (no local), rap (retain as published) ve rh (retain handling) gibi MQTT 5 abonelik seçeneklerini de kabul eder.

Dinamik olarak eklenen topic'lerden alınan mesajlar, msg.topic gerçek topic'e ayarlanmış şekilde normal bir abonelikle aynı çıkıştan gelir; böylece flow'unuzun geri kalanı değişmez.

Herkese Açık Broker'da Çalışan Bir Örnek

Üç makineyi simüle eden bir publisher başlatın. Topic'lerinizin benzersiz olması için 91c4 yerine kendi rastgele dizenizi yazın:

for i in $(seq 1 300); do
  for m in cnc1 cnc2 press1; do
    mosquitto_pub -h broker.connectmq.com -p 1883 -i demo-dyn-pub-91c4 \
      -t "connectmq/demo/91c4/line1/$m/telemetry" \
      -m "{\"machine\":\"$m\",\"load\":$((RANDOM % 100))}"
  done
  sleep 3
done

Şimdi aşağıdaki flow'u içe aktarın. cnc1'e subscribe olan, press1'i ekleyen, abonelikleri listeleyen ve her şeyden unsubscribe olan inject node'ları içerir.

[
  {"id":"tabd","type":"tab","label":"Dynamic subscribe"},
  {"id":"brkd","type":"mqtt-broker","name":"ConnectMQ public","broker":"broker.connectmq.com","port":"1883","clientid":"nr-dyn-91c4","autoConnect":true,"usetls":false,"protocolVersion":"4","keepalive":"60","cleansession":true},
  {"id":"i1","type":"inject","z":"tabd","name":"watch cnc1","props":[{"p":"action","v":"subscribe","vt":"str"},{"p":"topic","v":"connectmq/demo/91c4/line1/cnc1/telemetry","vt":"str"}],"repeat":"","once":false,"x":130,"y":60,"wires":[["dyn"]]},
  {"id":"i2","type":"inject","z":"tabd","name":"add press1 (QoS 1)","props":[{"p":"action","v":"subscribe","vt":"str"},{"p":"topic","v":"{\"topic\":\"connectmq/demo/91c4/line1/press1/telemetry\",\"qos\":1}","vt":"json"}],"repeat":"","once":false,"x":150,"y":110,"wires":[["dyn"]]},
  {"id":"i3","type":"inject","z":"tabd","name":"list","props":[{"p":"action","v":"getSubscriptions","vt":"str"}],"repeat":"","once":false,"x":110,"y":160,"wires":[["dyn"]]},
  {"id":"i4","type":"inject","z":"tabd","name":"unsubscribe all","props":[{"p":"action","v":"unsubscribe","vt":"str"},{"p":"topic","v":"true","vt":"bool"}],"repeat":"","once":false,"x":140,"y":210,"wires":[["dyn"]]},
  {"id":"dyn","type":"mqtt in","z":"tabd","name":"dynamic","topic":"","qos":"0","datatype":"auto-detect","broker":"brkd","inputs":1,"x":380,"y":130,"wires":[["dbgd"]]},
  {"id":"dbgd","type":"debug","z":"tabd","name":"received","active":true,"complete":"true","x":560,"y":130,"wires":[]}
]

"watch cnc1" düğmesine tıklayın; debug kenar çubuğunda yalnızca cnc1 verisi görünür. "add press1" düğmesine tıklayın; her iki makine de görünür. "list" aktif abonelikleri çıkarır, "unsubscribe all" ise yeniden deploy etmeden veri akışını durdurur.

Yeniden Başlatmadan Sonra Abonelikleri Geri Yükleyin

Çoğu kişinin gözden kaçırdığı kısım şudur: dinamik abonelikler bellekte yaşar. Node'u yeniden deploy ettiğinizde, Node-RED'i yeniden başlattığınızda veya broker bağlantısı kopup clean session ile yeniden bağlandığında node hiçbir aboneliği olmadan geri gelebilir. Bunu en baştan planlayın.

Güvenilir bir yöntem, istenen listeyi kalıcı context'te tutmak ve başlangıçta yeniden uygulamaktır:

// Function node "apply subscriptions"
// Giriş: msg.payload = makine ID'leri dizisi, örn. ["cnc1", "press1"]
const prefix = "connectmq/demo/91c4/line1/";
const wanted = Array.isArray(msg.payload) ? msg.payload : [];

global.set("watchedMachines", wanted, "file");   // dosya tabanlı bir context deposu gerektirir

// Tek bir çıkıştan sırayla iki mesaj
return [[
    { action: "unsubscribe", topic: true },
    { action: "subscribe", topic: wanted.map(m => ({ topic: prefix + m + "/telemetry", qos: 1 })) }
]];

Çift köşeli parantezler her iki mesajı tek çıkıştan sırayla gönderir; böylece unsubscribe her zaman yeni subscribe'dan önce ulaşır. Ardından birkaç saniye sonra bir kez tetiklenecek şekilde ayarlanmış, payload'ı global özellik #:(file)::watchedMachines (Node-RED'in adlandırılmış bir context deposundan okumak için kullandığı söz dizimi) olan bir inject node ekleyin ve onu aynı function'a bağlayın. Dosya tabanlı depolamanın kurulumu Node-RED context rehberimizde açıklanmıştır.

Yeniden subscribe olmadan önce her şeyi temizlemek node'un durumunu öngörülebilir tutar: ne sakladıysanız tam olarak onu elde edersiniz.

Abonelikleri Bir Panodan veya Config Topic'inden Yönetmek

Giriş yalnızca bir mesaj olduğu için onu her şey kontrol edebilir:

• Pano açılır menüsü. Bir Dashboard 2.0 açılır menüsü seçilen makineyi çıkarır; bir Change node msg.action = "subscribe" ve topic'i oluşturur. Önce önceki seçim için bir unsubscribe gönderin.
• Yapılandırma topic'i. Makinelerin retained bir JSON listesini connectmq/demo/91c4/config/watchlist gibi bir topic'e publish edin. Normal bir mqtt in node'u bunu okur ve yukarıdaki function'ı besler; böylece retained mesajı değiştirmek onu dinleyen her Node-RED örneğini yeniden yapılandırır.
• HTTP uç noktası. Bir http in node'u başka bir sistemden bir liste kabul edip aynı şekilde uygulayabilir.

Retained yapılandırma topic'i özellikle şıktır: broker en son listeyi Node-RED subscribe olur olmaz ona iletir; bu da dosya depolamaya gerek kalmadan yeniden başlatma sorununu çözer. Bu listeyi daha sonra temizlemeniz gerekirse retained mesajları silme notumuza bakın.

Dinamik Broker Bağlantıları

Aynı mekanizma bağlantıları da kontrol eder. msg.action = "disconnect" göndermek broker bağlantısını kapatır, connect ise yeniden açar. connect ile birlikte, çalışma anında config node'unu geçersiz kılmak için broker, port, clientid, username ve password gibi özellikler içeren msg.broker da gönderebilirsiniz. Bu, bir test broker'ı ile üretim arasında geçiş yapmak için kullanışlıdır; ancak kimlik bilgilerini flow'ların dışında tutun: bunları inject node'larına yazmak yerine bir function node'unda ortam değişkenlerinden okuyun.

Tuzaklar Kontrol Listesi

1. Giriş kablosu yoksa abonelik de yok. Dinamik bir node bir mesaj alana kadar hiçbir şeye subscribe olmaz.
2. Yinelenen client ID'ler. Aynı client ID'ye sahip iki Node-RED örneği birbirini broker'dan düşürür; her örneğin kendi ID'sine ihtiyacı vardır.
3. Çakışan filtreler. Hem line1/# hem de line1/cnc1/telemetry topic'ine subscribe olmak aynı mesajı iki kez iletebilir. Filtreleri ayrık tutun.
4. Sınırlar. Broker'ın abonelik ve topic derinliği sınırlarına uyun; herkese açık broker topic'leri 8 seviyeyle sınırlar.
5. Kaybolan durum. Yukarıda gösterildiği gibi istenen listeyi saklayın ve başlangıçta yeniden uygulayın.
6. Yanlış topic için yanlış çıkış. Tüm dinamik topic'ler tek bir çıkışı paylaşır; farklı işlem gerektiriyorlarsa msg.topic üzerinde bir Switch node ile yönlendirin.

Bir şey hiç bağlanmıyorsa abonelik mantığında hata ayıklamadan önce Node-RED MQTT bağlanmıyor rehberini adım adım izleyin. Node'un giriş özelliklerinin tüm ayrıntıları resmi Node-RED MQTT node belgelerinde ve node'un yerleşik yardım panelinde yer alır.

Dinamik Abonelikleri ConnectMQ ile Kullanın

Özel bir broker'da büyümek için alanınız olur: ConnectMQ, MQTT 3.1.1 ve 5.0'ı destekleyen barındırılan bir MQTT broker'dır; bu yüzden nesne biçimindeki MQTT 5 abonelik seçenekleri beklendiği gibi çalışır. Portalı canlı bir topic gezgini içerir; bu da bir izleme listesi oluşturmadan önce cihazların gerçekte hangi topic'lere publish ettiğini doğrulamayı kolaylaştırır. app.connectmq.com adresinde kredi kartı gerektirmeden ücretsiz bir Serverless broker oluşturun, kimlik bilgilerini 8883 portunda TLS ile broker node'unuza yapıştırın; dinamik flow'unuz değişmeden çalışır. Barındırma seçenekleri hakkında arka plan bilgisi için bulutta MQTT broker çalıştırma yazısını okuyun veya bu node'larda yeniyseniz Node-RED MQTT eğitimiyle başlayın.

Sık Sorulan Sorular

Hangi Node-RED sürümü dinamik MQTT aboneliklerini destekler?

Node-RED 2.2 ve sonrası. mqtt in node'unuzda "Dynamic subscription" seçeneği yoksa Node-RED'i güncelleyin.

Dinamik abonelikler yeniden deploy işleminden sonra korunur mu?

Hayır. Bellekte tutulurlar; bu yüzden istediğiniz listeyi kalıcı context'te veya retained bir yapılandırma topic'inde saklayın ve flow başladığında yeniden uygulayın.

Dinamik aboneliklerde topic başına QoS ayarlayabilir miyim?

Evet. msg.action değeri subscribe iken msg.topic olarak { "topic": "a/b", "qos": 1 } gibi bir nesne veya bu tür nesnelerden oluşan bir dizi gönderin.

Node'un neye subscribe olduğunu nasıl görürüm?

msg.action değeri getSubscriptions olan bir mesaj gönderin. Node güncel abonelik listesini msg.payload içinde çıkarır.