MQTT Paylaşımlı Abonelikler ve Yük Dengeleme Nasıl Test Edilir

Paylaşımlı abonelik ($share/<grup>/<topic>), birden çok subscriber'ı bir worker havuzuna dönüştürür: topic'teki her mesaj tüm üyelere değil, grubun yalnızca bir üyesine teslim edilir. Test etmek için aynı grupta birkaç worker başlatın, numaralı bir mesaj dizisi yayınlayın ve üç şeyi kontrol edin: her mesaj gruba tam olarak bir kez ulaştı mı, yük dağıldı mı ve akışın ortasında bir worker çöktüğünde bir şey kayboldu mu.

Bu rehber hızlı bir CLI kontrolü, çökme simülasyonlu bir Python testi ve doğrulamaya değer uç durumları gösteriyor. Kavramlar MQTT paylaşımlı abonelikler yazısında anlatılıyor. Burada yalnızca test ediyoruz.

Doğru çalışan bir broker ne yapar

Dağıtım stratejisi broker'a bağlıdır. Mosquitto round robin kullanır. Diğer broker'lar rastgele, sticky veya hash tabanlı, çoğu zaman yapılandırılabilir stratejiler sunar. Belirli bir stratejiye bağımlı değilseniz testiniz kesin bir deseni değil, tolerans içinde adil dağılımı kontrol etmelidir.

Adım 0: broker'ın desteklediğini doğrulayın

Bir MQTT 5 broker'ı desteği CONNACK içindeki Shared Subscription Available özelliğiyle bildirir. Özellik yoksa destek var kabul edilir. Değeri 0 ise broker $share aboneliklerini reddeder veya yok sayar ve SUBACK 0x9E Shared Subscriptions not supported döndürür. Her durumda asıl kanıt aşağıdaki dağılım testidir. Reason code aracı aldığınız SUBACK'i çözer.

Grup adı tek bir topic seviyesidir: /, + veya # içeremez. Geri kalanı normal bir topic filtresidir, bu yüzden gruptan sonra wildcard'lar çalışır: $share/workers/testmqtt/share/+/jobs. Broker'ı suçlamadan önce filtrelerinizi topic matcher ile kontrol edin.

Mosquitto CLI ile hızlı kontrol

Aynı grubun üyeleriyle iki terminal, normal bir subscriber ile de üçüncü bir terminal açın:

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

mosquitto_sub -h broker.connectmq.com -V mqttv5 -q 1 \
  -t 'testmqtt/share/9b1e/jobs' -v

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

Terminal 1 ve 2 on işi aralarında paylaşır (round robin ile tek numaralar bir tarafta, çift numaralar diğer tarafta). Terminal 3 on işin hepsini görür. Topic'i her zaman tek tırnak içine alın: tırnaksız ya da çift tırnak içinde yazarsanız shell $share ifadesini değişken olarak genişletir.

Otomatik dağılım ve failover testi

Bu paho-mqtt 2.x betiği tek bir grupta üç worker başlatır, numaralı 100 QoS 1 mesaj yayınlar ve worker başına sayıları, boşlukları ve kopyaları raporlar. --failover ile worker-0'ı yolun yarısında çökertir: worker okumayı bırakır, böylece son mesajları onaylanmadan kalır; ardından TCP bağlantısı gerçek bir çökmede olduğu gibi DISCONNECT gönderilmeden koparılır.

import sys, time, uuid, collections
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
RUN = uuid.uuid4().hex[:8]
TOPIC = f"testmqtt/share/{RUN}/jobs"
GROUP = f"workers{RUN}"
WORKERS, MESSAGES, RATE = 3, 100, 10     # RATE in messages per second
FAILOVER = "--failover" in sys.argv          # crash worker-0 halfway through

got = collections.defaultdict(list)          # worker name -> message ids received

def make_worker(name):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{name}-{RUN}", protocol=mqtt.MQTTv5)

    def on_connect(cl, userdata, flags, rc, props):
        if getattr(props, "SharedSubscriptionAvailable", 1) == 0:
            print(f"{name}: CONNACK says shared subscriptions are not available")
        cl.subscribe(f"$share/{GROUP}/{TOPIC}", qos=1)

    def on_subscribe(cl, userdata, mid, reason_codes, props):
        print(f"{name}: SUBACK {reason_codes[0]}")

    c.on_connect, c.on_subscribe = on_connect, on_subscribe
    c.on_message = lambda cl, userdata, msg: got[name].append(int(msg.payload))
    c.connect(HOST, PORT)
    c.loop_start()
    return c

workers = {f"worker-{i}": make_worker(f"worker-{i}") for i in range(WORKERS)}
time.sleep(2)                                 # let all subscriptions settle

pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"share-pub-{RUN}", protocol=mqtt.MQTTv5)
pub.connect(HOST, PORT)
pub.loop_start()
for i in range(MESSAGES):
    if FAILOVER and i == MESSAGES // 2:
        crashed = workers.pop("worker-0")
        crashed.loop_stop()                   # stop reading: unacked messages stay in flight
        crashed.socket().close()              # drop TCP without DISCONNECT, like a crash
        print(f"crashed worker-0 after message {i - 1}")
    pub.publish(TOPIC, str(i), qos=1)
    time.sleep(1 / RATE)
time.sleep(5)

all_ids = [m for ids in got.values() for m in ids]
missing = sorted(set(range(MESSAGES)) - set(all_ids))
dupes = len(all_ids) - len(set(all_ids))
for name in sorted(got):
    print(f"{name}: {len(got[name]):4d} messages, in order: {got[name] == sorted(got[name])}")
print(f"delivered {len(set(all_ids))}/{MESSAGES}, missing {len(missing)}, duplicates {dupes}")
if missing:
    print("first missing ids:", missing[:10])
for c in workers.values():
    c.disconnect(); c.loop_stop()
pub.disconnect(); pub.loop_stop()

Round robin kullanan bir broker'a karşı çalıştırma şöyle görünür:

worker-0: SUBACK Granted QoS 1
worker-1: SUBACK Granted QoS 1
worker-2: SUBACK Granted QoS 1
crashed worker-0 after message 49
worker-0:   17 messages, in order: True
worker-1:   42 messages, in order: True
worker-2:   41 messages, in order: True
delivered 100/100, missing 0, duplicates 0

Nasıl yorumlanır:

• --failover olmadan sayılar yaklaşık eşit olmalıdır (round robin ile 33/33/34). Çok dengesiz bir dağılım sticky veya hash stratejisine ya da broker'ın atladığı yavaş bir worker'a işaret eder.
• --failover ile missing 0, broker'ın çöken üyenin işini yeniden yönlendirdiği anlamına gelir. Çökme anının hemen çevresinde eksik ID'ler varsa yoldaki mesajlar oturumla birlikte atılmıştır. Bazı broker'lar önce istemcinin yeniden bağlanmasını bekler, bu yüzden gerçek oturum ayarlarınızla test edin.
• Kopyalar QoS 1'de meşrudur: bir mesaj hem çöken worker'a teslim edilip hem de bir başkasına yeniden gönderilebilir. Paylaşımlı bir gruptaki worker'lar idempotent olmalıdır. Yinelenen MQTT mesajları yazısına bakın.
• Hiç çökme olmadan eksik mesajlar genellikle broker tarafında bir hız veya kuyruk limiti anlamına gelir. RATE değerini düşürüp tekrar çalıştırın. Paylaşımlı genel endpoint'ler özellikle hız sınırlamaya yatkındır.

Doğrulamaya değer uç durumlar

1. Sıralama. Her üye kendi mesajlarını sırayla görür ama grup genelinde global bir sıra yoktur. 41 numaralı mesaj, başka bir worker'da 40'tan önce işlenebilir. Cihaz bazında sıra önemliyse mümkün olan yerde topic hash stratejisi kullanın ya da topic'leri bölümlere ayırın. Mesaj sıralama testi bunun nasıl ölçüleceğini gösterir.
2. Retained mesajlar. Retained bir mesaj yayınlayın, ardından paylaşımlı bir subscriber başlatın: hiçbir şey almamalıdır. Başlangıç durumuna ihtiyaç duyan worker'lar bunu başka bir yoldan almalıdır.
3. Karışık QoS. Teslim edilen QoS, yayın QoS'u ile her üyeye verilen QoS'un küçüğüdür. QoS 0 ile abone olmuş bir üye, diğerleri QoS 1 kullansa bile çöktüğünde mesaj kaybeder.
4. Abonelik seçenekleri. Paylaşımlı abonelikte No Local bir protokol hatasıdır. Diğer seçenekler MQTT abonelik seçenekleri yazısında açıklanıyor.
5. Grubu ölçeklendirmek. Çalışma sırasında dördüncü bir worker ekleyin: hemen kendi payını almaya başlamalıdır. Birini çıkarın: diğerleri yükü üstlenir.
6. Kalıcı üyeler. Session Expiry 0'dan büyükse bağlantısı kesilen bir üye gruptaki yerini korur ve bazı broker'lar ona, çevrimdışıyken biriken mesajlar göndermeye devam eder. Worker'ların kalıcı olması gerekip gerekmediğine karar verin.

Throughput kazancını test etmek

Paylaşımlı abonelikler tüketicileri ölçeklemek için vardır. Ölçeklemenin gerçek olduğunu ölçün: betiği mesaj başına sabit bir işlem süresi bekleyen 1, 2 ve 4 worker ile çalıştırın ve bir mesaj patlamasını boşaltmanın toplam süresini karşılaştırın. Throughput, broker veya ağ sınır olana kadar neredeyse doğrusal artmalıdır. Doğal tamamlayıcısı bir fan-in testidir: tek bir grubu besleyen çok sayıda publisher.

Dağılımı canlı izleyin

Çevrimiçi MQTT istemcisini iki tarayıcı sekmesinde açın, ikisini de aynı $share/... filtresine abone edin ve üçüncü bir sekmeden yayın yapın. Paylaşımı gerçek zamanlı görürsünüz; betiği çalıştırmadan önce iyi bir sağlama olur.

Tarayıcıdan test işleri yayınlayın

Bir paylaşımlı abonelik testi birkaç dakika sürer ve üretimde önemli olan soruları yanıtlar: her iş işleniyor mu, yük gerçekten dağılıyor mu ve bir worker öldüğünde yarım kalan işe ne oluyor? Failover çalıştırmasını CI paketinizde tutun ve broker'ı ya da oturum ayarlarını her değiştirdiğinizde tekrarlayın.

Sıkça sorulan sorular