MQTT Paylaşımlı Abonelikler: Mesaj Tüketicileri Arasında Yük Dengeleme

MQTT paylaşımlı abonelikleri (shared subscriptions), birkaç istemcinin bir grup olarak abone olmasını sağlar; böylece bir topic'teki her mesaj grubun tüm üyelerine değil, yalnızca bir üyesine teslim edilir. Bu yazının sonunda $share/group/topic sözdizimini, worker'lar gelip giderken teslimatın nasıl işlediğini, backend tüketicilerini bununla nasıl ölçekleyeceğinizi ve kullandığınız broker'ın bunu destekleyip desteklemediğini nasıl kontrol edeceğinizi anlamış olacaksınız.

Normal Abonelikler Tüketicileri Neden Ölçekleyemez?

Standart publish/subscribe modelinde, filtresi eşleşen her subscriber her mesajın kendi kopyasını alır. Dashboard'lar için tam olarak istediğiniz budur: her ekran her ölçümü gösterir. Backend işleme için ise tam olarak istemediğiniz şey budur. Telemetriyi bir veritabanına yazan bir servisin üç örneğini çalıştırırsanız, üç normal abonelik her ölçümün üç kez yazılması anlamına gelir.

MQTT 5'ten önce geçici çözümler hantaldı: darboğaza dönüşen tek bir tüketici, topic'leri örnekler arasında elle bölmek veya MQTT'yi ayrı bir kuyruk sistemine köprülemek. Paylaşımlı abonelikler bu sorunu MQTT broker'ın kendi içinde çözer.

MQTT Paylaşımlı Abonelikleri Nasıl Çalışır?

Paylaşımlı bir abonelik, MQTT 5'te standartlaştırılmış özel bir filtre formatı kullanır:

$share/<ShareName>/<TopicFilter>

• $share, aboneliği paylaşımlı olarak işaretleyen sabit önektir.
• <ShareName>, tüketici grubunun adıdır. /, + veya # içermemelidir.
• <TopicFilter>, wildcard'lar dahil normal bir filtredir.

Aynı $share/<ShareName>/<TopicFilter> filtresine abone olan her istemci aynı gruba katılır. Eşleşen her mesaj için broker grubun bir üyesini seçer ve mesajı ona teslim eder. Farklı paylaşım adları birbirinden bağımsız gruplardır ve düz topic'e normal şekilde abone olanlar kendi kopyalarını almaya devam eder.

Spesifikasyon üye seçimini broker'a bırakır. Round-robin, rastgele ve "en az meşgul" stratejilerinin hepsi yaygındır; bu nedenle belirli bir sıraya veya bir cihazdan gelen mesajların her zaman aynı worker'a ulaşacağına güvenmeyin.

Bilmeniz Gereken Teslimat Semantiği

1. Sıralama topic bazında değil, üye bazındadır. Her worker kendi mesajlarını sırayla görür, ancak aynı sensörden gelen ardışık iki ölçüm farklı worker'lara gidip paralel olarak işlenebilir. Sıra önemliyse bir sıra numarası veya zaman damgası ekleyin ve bunu veritabanınızda çözün.
2. QoS hâlâ geçerlidir. QoS 1'de bir worker bir mesajı onaylamadan bağlantıyı keserse, broker'lar genellikle mesajı yeniden bağlandığında aynı oturuma tekrar teslim eder veya başka bir üyeye yönlendirir; davranış broker'a göre değişir. İşleyicileri idempotent yapın.
3. Retained mesajlar abonelik anında gönderilmez. Spesifikasyon, paylaşımlı bir abonelik oluşturulduğunda retained mesajların ona teslim edilmediğini söyler; çünkü hangi üyenin bunları alması gerektiği belirsiz olurdu.
4. No Local yasaktır. Paylaşımlı bir abonelikte No Local abonelik seçeneğini ayarlamak protokol hatasıdır.
5. Oturumlar üye bazındadır. Her worker'ın kendi client ID'si ve oturumu vardır. Bir worker'daki kalıcı oturum, o worker çevrimdışıyken aboneliğini canlı tutar. Bunun kuyruklamayı nasıl etkilediğini görmek için MQTT kalıcı oturumlar yazısına bakın.

Örnek: Python Worker'larını Ölçeklemek

Aşağıdaki worker db-writers adlı bir gruba katılır ve telemetriyi işler. Yükün nasıl dağıldığını izlemek için onu farklı worker numaralarıyla iki veya üç terminalde çalıştırın.

import sys
import json
import paho.mqtt.client as mqtt

WORKER = sys.argv[1] if len(sys.argv) > 1 else "1"
FILTER = "$share/db-writers/connectmq/demo/s7h3/factory/+/telemetry"

def on_connect(client, userdata, flags, reason_code, properties):
    print(f"Worker {WORKER} connected: {reason_code}")
    client.subscribe(FILTER, qos=1)

def on_subscribe(client, userdata, mid, reason_code_list, properties):
    for rc in reason_code_list:
        if rc.is_failure:
            print(f"Subscribe rejected: {rc}")  # ör. Shared Subscriptions not supported
        else:
            print(f"Joined group, granted {rc}")

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    print(f"Worker {WORKER} got {msg.topic}: {data}")
    # burada veritabanınıza yazın; idempotent tutun

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"db-writer-s7h3-{WORKER}",
                     protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_subscribe = on_subscribe
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

İki worker başlatın:

python worker.py 1
python worker.py 2

Ardından bir dizi test ölçümü publish edin:

for i in $(seq 1 10); do
  mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -i sh-pub-s7h3 -q 1 \
    -t "connectmq/demo/s7h3/factory/line$((i % 3))/telemetry" \
    -m "{\"seq\": $i, \"temp\": 70}"
done

Broker paylaşımlı abonelikleri destekliyorsa, her worker on mesajın tamamını yazdırmak yerine on mesaj worker'lar arasında bölünür. s7h3 değerini kendi rastgele dizenizle değiştirin; herkese açık MQTT test broker'ı herkes tarafından okunabilir, bu yüzden yalnızca sahte veri publish edin.

Aynı abonelik MQTT.js'te şöyle görünür:

const mqtt = require("mqtt");
const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId: "db-writer-node-s7h3-1",
  protocolVersion: 5,
});
client.on("connect", () => {
  client.subscribe("$share/db-writers/connectmq/demo/s7h3/factory/+/telemetry",
    { qos: 1 }, (err, granted) => console.log(err || granted));
});
client.on("message", (topic, payload) => console.log(topic, payload.toString()));

Paylaşımlı Abonelik Desteği Nasıl Doğrulanır?

Destek broker'lar ve planlar arasında farklılık gösterir; bu yüzden varsaymak yerine kontrol edin:

1. SUBACK'i okuyun. MQTT 5'te paylaşımlı abonelikleri desteklemeyen bir broker 0x9E (Shared Subscriptions not supported) reason code'unu döndürür. Yukarıdaki on_subscribe callback'i bunu yazdırır.
2. CONNACK'i kontrol edin. MQTT 5 broker'ları CONNACK içinde Shared Subscription Available = 0 bildirebilir. Özellik yoksa destek olduğu varsayılır.
3. Bölme testini çalıştırın. İki üye, on mesaj. Her iki üye de on mesajın tamamını yazdırıyorsa broker $share/... ifadesini bir grup olarak değil, sıradan bir topic filtresi olarak ele almıştır.
4. Planı kontrol edin. Yönetilen servisler bu özelliği çoğu zaman bir kademeye bağlar. ConnectMQ'da paylaşımlı abonelikler Serverless (Paid) planı ve üzerinde dahildir.
5. MQTT 3.1.1 ile ayrıca test edin. Bazı broker'lar 3.1.1 istemcilerinden gelen $share ifadesini bir uzantı olarak kabul eder; diğerleri MQTT 5 gerektirir. Broker'ınız bunu belgelemedikçe buna güvenmeyin.

Paylaşımlı Aboneliklerin Diğer Yaklaşımlarla Karşılaştırılması

İşi bölmenin başka yolları da vardır ve hangisinin ne zaman uygun olduğunu bilmek faydalıdır. Elle topic bölümleme (worker 1 line1'e, worker 2 line2'ye abone olur) bölüm bazında katı sıralama sağlar, ancak her worker veya hat eklediğinizde yeniden yapılandırma gerektirir. Dahili bir kuyruğu besleyen tek bir tüketici sıralamayı basit tutar, ancak bu tek süreci bir darboğaz ve tek hata noktası haline getirir. Paylaşımlı abonelikler worker'lar arasında koordinasyon gerektirmez ve yalnızca daha fazla örnek başlatarak ölçeklenir; bunun bedeli, worker'lar arasında topic bazında sıralamanın kaybedilmesidir. Çoğu telemetri alımı ve alarm işleme için doğru takas budur.

Kalıplar ve Tuzaklar

• Her amaç için bir grup. Her bağımsız iş için bir paylaşım adı kullanın (db-writers, alerting, exporter). Böylece her iş, worker'ları genelinde her mesajı tam olarak bir kez alır.
• Otomatik ölçekleme. Bir worker eklemek, aynı filtreyle başka bir istemciyi bağlamaktan ibarettir; birini çıkarmak ise bağlantıyı kesmektir. MQTT client ID en iyi uygulamaları yazısında açıklandığı gibi her örneğe benzersiz bir client ID verin.
• Akış kontrolü. Yavaş bir örneğe büyük bir onaylanmamış mesaj yükü verilmemesi için paylaşımlı abonelikleri her worker'da küçük bir MQTT 5 Receive Maximum değeriyle birleştirin.
• Wildcard'lar sorun değil. $share/group/factory/# bütün bir dalı dağıtır. Eşleşme kuralları MQTT wildcard'ları yazısında açıklanan olağan kurallardır.
• Fan-out için kullanmayın. Her tüketicinin her mesajı görmesi gerekiyorsa normal abonelikler kullanın.

MQTT'de genel olarak yeni misiniz? MQTT eğitimi abonelikleri, QoS'u ve oturumları en baştan ele alıyor.

Tüketicileri ConnectMQ'da Ölçekleyin

ConnectMQ, MQTT 3.1.1 ve 5.0'ı destekleyen tamamen yönetilen bir MQTT servisidir; paylaşımlı abonelikler QoS 2, 1 MB mesajlar ve 1.000'e kadar bağlantıyla birlikte Serverless (Paid) planından itibaren kullanılabilir. Yalıtılmış bir kümeye ihtiyaç duyduğunuzda Dedicated ve BYOC seçenekleri de mevcuttur. Kredi kartı gerektirmeden kaydolun ve ücretsiz Serverless planıyla başlayın, backend'iniz tüketici gruplarına ihtiyaç duyduğunda planınızı yükseltin. Worker'larınızı YOUR_USERNAME, YOUR_PASSWORD ve YOUR_CLIENT_ID ile bağlayın.

Sık Sorulan Sorular

MQTT paylaşımlı aboneliğinin sözdizimi nedir?

$share/<ShareName>/<TopicFilter> kullanın; örneğin $share/workers/sensors/+/temperature. Aynı paylaşım adını ve filtreyi kullanan tüm istemciler tek bir grup oluşturur.

Paylaşımlı abonelikler MQTT 3.1.1 ile çalışır mı?

MQTT 5'te tanımlanmışlardır. Bazı broker'lar $share sözdizimini 3.1.1 istemcilerinden bir uzantı olarak kabul eder, ancak buna güvenmeden önce broker'ınızın belgelerinden doğrulamalısınız.

Retained mesajlar paylaşımlı aboneliklere teslim edilir mi?

Hayır. Paylaşımlı bir abonelik oluşturulduğunda retained mesajlar ona gönderilmez. Bir worker'ın retained duruma ihtiyacı varsa bunu ayrı bir normal abonelikle okuyun.

Broker bir mesajı hangi subscriber'ın alacağını nasıl seçer?

MQTT spesifikasyonu bunu broker'a bırakır. Round-robin ve rastgele seçim yaygındır; bu nedenle belirli bir mesajı belirli bir üyenin alacağına asla güvenmeyin.