Apache NiFi ile MQTT: ConsumeMQTT ve PublishMQTT Yapılandırması

Apache NiFi, MQTT'ye iki standart processor ile bağlanır: ConsumeMQTT bir topic filtresine abone olup gelen mesajları FlowFile'a dönüştürür, PublishMQTT ise FlowFile içeriğini bir topic'e gönderir. Minimum yapılandırma; tcp://broker.connectmq.com:1883 gibi bir Broker URI, bir Topic Filter veya Topic, bir QoS ve ConsumeMQTT'de bir Max Queue Size'dır. Geri kalan her şey (TLS, kimlik bilgileri, session'lar, record'lar, cluster), akışın production'da güvenilir olup olmayacağını belirler.

Bu eğitim önemli her özelliği tek tek ele alıyor, ardından JSON telemetrisini tüketen, SQL ile filtreleyen ve alarmları MQTT'ye geri publish eden bir akışı kurup test ediyor.

Test için NiFi'yi başlatın

NiFi 2.x, Java 21 üzerinde çalışır; arayüze en hızlı ulaşma yolu resmi Docker imajıdır:

docker run --name nifi -p 8443:8443 \
  -e SINGLE_USER_CREDENTIALS_USERNAME=admin \
  -e SINGLE_USER_CREDENTIALS_PASSWORD=change-me-to-12-chars-min \
  -d apache/nifi:latest

https://localhost:8443/nifi adresini açın (self-signed sertifikayı kabul edin) ve giriş yapın. Kalıcı kullanacağınız her şey için belirli bir imaj tag'i sabitleyin. Parola en az 12 karakter olmalıdır, aksi halde container başlamaz.

ConsumeMQTT özellikleri

Canvas'a bir processor sürükleyin, ConsumeMQTT arayın ve özelliklerini açın. Her akışta ayarlayacaklarınız şunlardır:

Processor her mesajı Message relationship'ine yazar ve şu attribute'ları ekler: mqtt.broker, mqtt.topic, mqtt.qos, mqtt.isDuplicate ve mqtt.isRetained. Record reader yoksa her MQTT mesajı bir FlowFile olur ve FlowFile içeriği ham payload'dur.

Record'larla toplu işleme

Mesaj başına bir FlowFile, saniyede birkaç mesaj için uygundur, ötesinde israftır. Record Reader'ı bir JsonTreeReader, Record Writer'ı bir JsonRecordSetWriter yapın; ConsumeMQTT son çalışmadan bu yana gelen tüm mesajları tek bir FlowFile'a record olarak yazar. Add attributes as fields true iken her record'a ayrıca _topic, _qos, _isDuplicate ve _isRetained alanları eklenir; böylece birçok cihazın mesajı aynı FlowFile'ı paylaştığında topic bilgisini kaybetmezsiniz. Reader'ın parse edemediği payload'lar parse.failure relationship'ine gider; hatalı cihazların görünür olması için bunu bir funnel'a veya log'a yönlendirin. Tipik nedenler için JSON parse hataları yazısına bakın.

TLS

URI'yi ssl://broker.connectmq.com:8883 olarak değiştirin (yalnızca HTTPS'e izin veren proxy'lerin arkasında wss://broker.connectmq.com:8084/mqtt) ve SSL Context Service olarak bir StandardSSLContextService ayarlayın. Herkesçe güvenilen sertifikaya sahip bir broker için truststore, JVM'in cacerts dosyasını gösterebilir; özel bir CA için CA'yı ayrı bir truststore'a aktarın. Handshake başarısız olursa bir openssl s_client testiyle karşılaştırın; komutlar MQTT TLS bağlantısını test etme yazısında.

PublishMQTT özellikleri

PublishMQTT bağlantı özelliklerini (Broker URI, sürüm, Client ID, kimlik bilgileri, SSL) paylaşır ve şunları ekler:

Record reader yoksa FlowFile içeriğinin tamamı tek bir payload'dur. Boyutu broker'ınızın maksimum paket boyutuyla karşılaştırın; tek bir MQTT mesajı olarak gönderilen 5 MB'lık birleştirilmiş bir FlowFile sık rastlanan bir kazadır. Paket boyutu hesaplayıcı ek yükü tahmin etmenize yardımcı olur.

Başarısız publish'ler failure'a gider. Broker'daki kısa bir kesintinin veri kaybettirmemesi için bu relationship'i retry veya penalty ile PublishMQTT'ye geri bağlayın.

Test edilmiş bir akış kurun

Hedef: testmqtt/nifi/<device>/telemetry üzerinden telemetri tüketmek, hepsini saklamak ve sıcaklık 30'u aştığında bir alarm publish etmek.

ConsumeMQTT ──Message──▶ QueryRecord ──hot──▶ PublishMQTT (testmqtt/nifi/alerts)
     │                        └──original──▶ PutDatabaseRecord / PublishKafka / PutFile
     └──parse.failure──▶ LogAttribute

1. Yukarıdaki özelliklerle ConsumeMQTT; Record Reader JsonTreeReader, Record Writer JsonRecordSetWriter, Add attributes as fields true.
2. Aynı reader/writer ile ve hot adlı dinamik bir özellikle QueryRecord:

SELECT _topic, device, temperature, ts
FROM FLOWFILE
WHERE temperature > 30

1. hot relationship'inde PublishMQTT; Topic testmqtt/nifi/alerts, QoS 1, Record Reader/Writer ile — böylece her sıcak ölçüm ayrı bir mesaj olarak publish edilir.
2. original relationship'ini depolamanıza gönderin: PutDatabaseRecord ile PostgreSQL, PublishKafka ile Kafka veya test ederken basitçe PutFile.

Varsayılan olarak QueryRecord, hiçbir şey eşleşmediğinde de hot'a boş bir FlowFile gönderir. Yalnızca en az bir sıcak ölçüm içeren grupların PublishMQTT'ye ulaşması için Include Zero Record FlowFiles'ı false yapın.

Uçtan uca test edin

Tüm processor'ları başlatın, ardından alarm topic'ine bir subscriber açın:

mosquitto_sub -h broker.connectmq.com -p 1883 -t 'testmqtt/nifi/alerts' -v

Bir normal ve bir sıcak ölçüm publish edin:

mosquitto_pub -h broker.connectmq.com -p 1883 -q 1 \
  -t testmqtt/nifi/press-7/telemetry \
  -m '{"device":"press-7","temperature":24.1,"ts":1760700000}'

mosquitto_pub -h broker.connectmq.com -p 1883 -q 1 \
  -t testmqtt/nifi/press-7/telemetry \
  -m '{"device":"press-7","temperature":33.8,"ts":1760700005}'

Terminal 1, "temperature":33.8 ve "_topic":"testmqtt/nifi/press-7/telemetry" içeren tam olarak bir alarm yazdırmalıdır. NiFi'de ConsumeMQTT'den sonraki kuyruğa sağ tıklayıp List queue'yu seçin (veya Data Provenance'ı açın); iki FlowFile'ı ve mqtt.* attribute'larını görürsünüz. Sonra aynı topic'e not json publish edin ve akışı bozmak yerine parse.failure'a düştüğünü doğrulayın. Tarayıcıyı tercih ediyorsanız aynısını online MQTT istemcisiyle yapabilirsiniz.

NiFi test topic'lerini canlı izleyin

ConsumeMQTT'yi cluster'da çalıştırmak

Bir NiFi cluster'ında her node her processor'ı çalıştırır. ConsumeMQTT için bu, her node'un abone olduğu ve her node'un her mesajı aldığı anlamına gelir. İki desenden birini seçin:

• Yalnızca primary node. Processor'ın Scheduling sekmesinde Execution'ı Primary node yapın. Basit ve sıralıdır, ancak tüm MQTT işini tek node yapar ve primary değiştiğinde kısa bir boşluk oluşur.
• Shared subscription. Bir Group ID (örneğin nifi-ingest) ayarlayın ve her node'un benzersiz bir ID alması için Client ID'yi boş bırakın. Broker mesajları node'lar arasında dağıtır; bkz. MQTT shared subscription'lar. Bu durumda sıralama global değil, yalnızca node bazında korunur.

Tüm node'lara asla aynı sabit Client ID'yi vermeyin: broker client ID başına bir bağlantıya izin verir, dolayısıyla node'lar session'ı döngü halinde birbirlerinden alır.

Session'lar, QoS ve yeniden başlatmalar

Clean Session ve QoS 0 ile NiFi dururken publish edilen her şey kaybolur. Kaybetmeyi göze alamayacağınız veri alımı için QoS 1, tüketici başına sabit bir Client ID, Resume Session ve MQTT 5'te bakım pencerelerini kapsayacak uzunlukta bir Session Expiry Interval kullanın. Broker kesinti boyunca mesajları kuyruğa alır ve NiFi yeniden bağlandığında teslim eder. Bu, kalıcı session tutan bir broker gerektirir; paylaşımlı test endpoint'i broker.connectmq.com offline session saklamaz, bu yüzden bu davranışı kendi özel broker instance'ınızda doğrulayın (bkz. kalıcı session'lar).

Sık karşılaşılan hatalar

Sayısal reason code'ları MQTT reason code aracında arayın.

NiFi'nin iki MQTT processor'ı basittir, ancak production akışları ayrıntılara bağlıdır: TLS ve parametreleştirilmiş kimlik bilgileri, mesaj başına bir FlowFile yerine record'lar, net bir cluster stratejisi ve ne kadar veri kaybını göze alabileceğinize uygun bir session yapılandırması. Gerçek cihazları emanet etmeden önce her akışı ikinci bir MQTT istemcisiyle test edin.

Sıkça sorulan sorular