MQTT Shared Subscriptions: $share ile Yük Dengeleme

Normal MQTT’de filtresi bir topic’le eşleşen her subscriber, her mesajın kendi kopyasını alır. Bu fan-out için mükemmeldir, ancak birden fazla backend worker’ının bir mesaj akışını aralarında paylaşmasını istediğinizde kullanışsızdır. MQTT 5’te standartlaştırılan shared subscription’lar (paylaşımlı abonelikler) bunu çözer: subscriber’lar adlandırılmış bir gruba katılır ve broker her mesajı bu grubun yalnızca bir üyesine teslim eder.

$share sözdizimi

Paylaşımlı abonelik, özel bir topic filtresine sahip normal bir subscribe isteğidir:

$share/<ShareName>/<TopicFilter>

• $share — aboneliği paylaşımlı olarak işaretleyen sabit bir önek.
• ShareName — grup adı. En az bir karakter olmalı ve /, + veya # içermemelidir.
• TopicFilter — wildcard’lar dahil sıradan bir filtre. Bkz. MQTT topic’leri ve joker karakterler.

Örneğin $share/ingest/factory/+/telemetry, subscriber’ı her factory/<line>/telemetry topic’i için ingest grubuna koyar. Publisher’lar özel bir şey yapmaz: yine factory/line-3/telemetry topic’ine publish ederler ve $share önekini hiç görmezler. Bir filtrenin hangi topic’lerle eşleştiğini kontrol etmek için grup adından sonraki kısmı MQTT topic eşleştiriciye yapıştırın.

Mesajlar nasıl dağıtılır?

Teslimat kurallarını en kolay bir örnekle görebilirsiniz. Şu aboneliklerin var olduğunu varsayalım:

Yani her grup bir kopya alır ve grubun içinde bir üye onu işler. Üyenin nasıl seçileceği spesifikasyonda tanımlanmamıştır. Broker’lar round robin, rastgele, sticky (üye ayrılana kadar hep aynı üyeye gönderme) veya client ID ya da topic üzerinden hash gibi stratejiler sunar — örneğin EMQX bunu yapılandırmanıza izin verir. Her üye kendi abonelik QoS’unu korur ve mesajlar her zamanki gibi üye bazında onaylanır.

Broker desteği

Paylaşımlı abonelikler MQTT 5 standardının parçasıdır ve broker desteğini CONNACK içindeki Shared Subscription Available özelliğiyle bildirir. Desteklemeyen bir broker, subscribe isteğini 0x9E (Shared Subscriptions not supported) reason code’u ile reddeder — bkz. MQTT reason code başvuru kaynağı.

• Mosquitto, 1.6 sürümünden beri MQTT 5 ve 3.1.1 client’ları için $share’i destekler.
• EMQX, $share’in yanı sıra tek bir varsayılan grupta paylaşımlı abonelik gibi davranan eski bir kısayol olan $queue/<filter>’i de destekler.
• HiveMQ, her iki protokol sürümü için $share’i destekler.

Yönetilen ve bulut broker’lar farklılık gösterir; bu yüzden ona güvenmeden önce dokümantasyonu kontrol edin. MQTT broker karşılaştırmamız başlıca seçenekleri ele alıyor.

Örnek: bir akışı paylaşan iki worker

Üç terminal açın. İlk ikisi aynı gruba katılır; üçüncüsü on mesaj publish eder:

mosquitto_sub -V mqttv5 -h broker.hivemq.com -q 1 \
  -t '$share/workers/testmqtt/share-demo/jobs' -v

for i in $(seq 1 10); do
  mosquitto_pub -V mqttv5 -h broker.hivemq.com -q 1 \
    -t 'testmqtt/share-demo/jobs' -m "job $i"
done

Her iş tam olarak bir worker terminalinde görünür. Filtrenin etrafındaki tek tırnaklara dikkat edin: çoğu shell’de çift tırnak içindeki $share bir değişken olarak genişletilir. Aynısı kodda da çalışır:

import mqtt from 'mqtt';

const client = mqtt.connect('wss://broker.hivemq.com:8884/mqtt', { protocolVersion: 5 });

client.on('connect', () => {
  client.subscribe('$share/workers/testmqtt/share-demo/jobs', { qos: 1 });
});

client.on('message', (topic, payload) => {
  // topic is the real topic, without the $share prefix
  console.log(topic, payload.toString());
});

Publisher’ı tarayıcınızdan da çalıştırabilirsiniz: worker’larınız çalışırken aşağıdaki client’ı açın ve testmqtt/share-demo/jobs topic’ine publish edin.

Online client’ta test işleri publish edin

Kullanım senaryoları

• Backend tüketicilerinin yatay ölçeklenmesi. Bir veri alım (ingestion) servisinin N instance’ını çalıştırın ve ayrı bir mesaj kuyruğu olmadan telemetriyi broker’ın aralarında dağıtmasına izin verin.
• Yüksek erişilebilirlik. Bir tüketici çökerse kalan üyeler mesaj almaya devam eder.
• Veritabanlarına veya stream işlemcilerine köprü. Birden fazla yazıcı, mesajları bir zaman serisi veritabanına ekleme veya Kafka’ya iletme yükünü paylaşabilir.
• İş kuyrukları. “Bu resmi yeniden boyutlandır” gibi komutları onları alan herhangi bir worker işleyebilir.

Paylaşımlı abonelik kurulumu tasarlamak

Grup adlarını tüketici türüne göre seçin

Grupları instance’a göre değil, işe göre adlandırın: $share/ingest/..., $share/alerts/.... Ingest servisinin her instance’ı ingest’e, alarm servisinin her instance’ı alerts’a katılır. Böylece her servis bir bütün olarak her mesajı tam bir kez görürken instance’ları işi paylaşır. Yanlışlıkla her instance’a benzersiz bir grup adı verirseniz düz fan-out’a geri dönersiniz ve her instance her mesajı işler.

Worker’lar için oturumlar ve QoS

Worker’lar genellikle QoS 1 ile abone olur; böylece bir worker mesajı onaylamadan çökerse mesaj yeniden teslim edilir. Tüm üyeler çevrimdışıyken mesajların bir grup için kuyruğa alınıp alınmayacağı üyelerin oturumlarına bağlıdır: kalıcı bir oturumla broker kuyruğa almaya devam edebilir; clean session’larda ise gruptan kimse bağlı değilken publish edilen mesajlar o gruba hiç teslim edilmez.

Paylaşımlı abonelikler ve mesaj kuyrukları

Paylaşımlı abonelikler MQTT içinde basit rakip tüketiciler (competing consumers) sağlar. Bir offset’ten yeniden oynatma, dead-letter kuyrukları veya partition bazında sıralama garantileri gibi özel kuyruk sistemlerinin özelliklerini sunmazlar. Birçok IoT backend’i için bu denge uygundur; yeniden oynatmaya ihtiyacınız varsa akışı bir paylaşımlı abonelikten Kafka gibi bir log’a aktarın ve orada işleyin.

Tuzaklar ve sınırlamalar

• Global sıralama yoktur. MQTT tek bir subscriber için topic bazında sıralamayı garanti eder, ancak mesajlar birden fazla worker’a dağıtıldığında paralel işlenir. Cihaz bazında sıra önemliyse topic üzerinden hash yapan bir broker stratejisi kullanın veya topic’e göre kendiniz bölümleyin.
• Retained mesaj yoktur. MQTT 5 spesifikasyonuna göre paylaşımlı bir abonelik kurulduğunda retained mesajlar gönderilmez. Retained durum topic’leri gibi durum bilgileri için normal bir abonelik kullanın.
• No Local’a izin verilmez. Paylaşımlı bir abonelikte No Local seçeneğini set etmek MQTT 5’te bir protokol hatasıdır.
• Tam filtreyle unsubscribe olun. Gruptan ayrılmak için düz filtreden değil, tam olarak $share/group/filter dizesinden unsubscribe olun.
• ACL’ler. Broker’lar, erişim kurallarının $share/... dizesine mi yoksa alttaki filtreye mi uygulanacağı konusunda farklılık gösterir. Yetkilerinizi gerçek sözdizimiyle test edin.
• Kopyalar yine de mümkündür. QoS 1 “en az bir kez” demektir ve bir bağlantı kopmasından sonra başka bir üyeye yeniden teslimat kopyalar üretebilir. Handler’ları idempotent tasarlayın; bkz. MQTT QoS seviyeleri.

Sıkça sorulan sorular