MQTT Fan-Out Testi: Tek Publisher, Binlerce Subscriber

Bir fan-out senaryosunda tek bir mesaj çok sayıda alıcıya gider: tüm filoya gönderilen bir yapılandırma değişikliği, 10.000 dashboard'da gösterilen bir fiyat güncellemesi, bir "yeni firmware var" duyurusu. Publisher tek bir PUBLISH gönderir; broker her eşleşen abonelik için bir tane gönderir. Bir fan-out testi, son subscriber'ın mesajı alması için geçen süreyi ve herkesin mesajı gerçekten alıp almadığını ölçer.

Bu rehber 1 → 10.000 bir benchmark kurar, teslim yayılımını ölçen bir sonda ekler ve ağ faturası sizi şaşırtmasın diye bant genişliği hesabını yapar. Genel hız testleri için MQTT throughput testi, ters desen için fan-in testi yazılarına bakın.

Fan-out neden ayrı bir testtir

Publish throughput testleri genellikle her publisher'ı bir subscriber ile eşleştirir. Fan-out ise başka bir şeyi zorlar: broker'ın bir mesajı binlerce çıkış kuyruğuna kopyalama ve bunları binlerce socket'e boşaltma yeteneğini.

Cevaplanması gereken soru "saniyede kaç mesaj" değil, gerçek filo boyutunuzda "her subscriber'ın mesajı alması ne kadar sürüyor" sorusudur, yani teslim yayılımı.

Önce bant genişliği hesabı

Başlamadan önce hesabı yapın: bir fan-out testi, broker zorlanmadan çok önce bir ağ bağlantısını doyurabilir. Hattaki bir PUBLISH kabaca sabit başlık + 2 bayt topic uzunluğu + topic + paket tanımlayıcı (QoS > 0) + özellikler + payload'dan oluşur. Ayrıntılar için MQTT bant genişliği ek yükü yazısına bakın.

Örnek: testmqtt/fanout/broadcast (25 bayt) üzerinde QoS 0 ile 500 baytlık payload ≈ PUBLISH başına 530 bayt.

Buna TCP/IP ve TLS ek yükünü de ekleyin. Son satır, birçok tekil yük üretici makinenin tek başına alabileceğinden fazladır; bu yüzden fan-out testleri genellikle birkaç makineye dağıtılır. Paket boyutu hesaplayıcı topic'iniz ve payload'ınız için paket başına tam boyutu verir.

Adım 1: subscriber'ları açın

emqtt-bench binlerce subscriber bağlantısını açık tutabilir. File descriptor ve ephemeral port sınırlarından uzak durmak için 10.000 bağlantıyı dört makineye (veya farklı host'lardaki container'lara) bölün:

ulimit -n 65535
docker run --rm -it --ulimit nofile=65535:65535 emqx/emqtt-bench sub \
  -h broker.connectmq.com -p 1883 \
  -c 2500 -i 5 \
  -t testmqtt/fanout/broadcast -q 0 \
  -u "$MQTT_USER" -P "$MQTT_PASS"

-i 5 her 5 ms'de bir bağlantı açar (makine başına saniyede 200); yani dört makinenin herkesi bağlaması 12 saniyeden biraz fazla sürer. Bağlantıları daha hızlı açmak başlı başına bir testtir (bağlantı stres testi yazısına bakın) ve fan-out ölçümleriyle karıştırılmamalıdır. Publish etmeden önce dört makinenin de tam sayıyı raporlamasını bekleyin.

Adım 2: publish edin ve sondalayın

emqtt-bench'in subscriber çıktısı toplam alım hızını gösterir, ancak en yavaş subscriber'ın ne kadar geciktiğini göstermez. Bu paho-mqtt 2.x betiği tek bir süreçte 50 sonda subscriber ekler, aynı broadcast topic'ine zaman damgalı mesajlar publish eder ve sondalar genelinde gecikme ile teslim yayılımını raporlar. Publisher ve sondalar aynı saati paylaştığından değerler doğrudan karşılaştırılabilir:

import json, threading, time, uuid
from collections import defaultdict
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
TOPIC = "testmqtt/fanout/broadcast"
PROBES, MESSAGES, RATE, QOS = 50, 60, 1, 0
lat = defaultdict(list)          # seq -> sondalar genelinde gecikmeler (ms)
lock = threading.Lock()

def on_message(client, userdata, msg):
    try:
        m = json.loads(msg.payload)
    except ValueError:
        return                   # sonda disi trafigi yok say
    with lock:
        lat[m["seq"]].append((time.perf_counter() - m["t"]) * 1000)

def new_client(name):
    return mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                       client_id=f"{name}-{uuid.uuid4().hex[:6]}")

probes = []
for n in range(PROBES):
    c = new_client(f"fanout-probe-{n}")
    c.on_message = on_message
    c.connect(HOST, PORT)
    c.subscribe(TOPIC, qos=QOS)
    c.loop_start()
    probes.append(c)

pub = new_client("fanout-pub")
pub.connect(HOST, PORT)
pub.loop_start()
time.sleep(3)                    # aboneliklerin oturmasini bekle

for seq in range(MESSAGES):
    body = {"seq": seq, "t": time.perf_counter(), "pad": "x" * 450}
    pub.publish(TOPIC, json.dumps(body), qos=QOS)
    time.sleep(1 / RATE)
time.sleep(5)

pct = lambda xs, q: xs[min(len(xs) - 1, int(len(xs) * q))]
all_lat = sorted(v for vals in lat.values() for v in vals)
spread = sorted(max(v) - min(v) for v in lat.values() if len(v) > 1)
print(f"delivered {len(all_lat)}/{PROBES * MESSAGES}")
if all_lat:
    print(f"latency p50 {pct(all_lat, .5):.1f}  p99 {pct(all_lat, .99):.1f}  max {all_lat[-1]:.1f} ms")
if spread:
    print(f"spread  p50 {pct(spread, .5):.1f}  max {spread[-1]:.1f} ms")

Betiği üç kez çalıştırın: bench subscriber'ı olmadan (taban çizgisi), 2.500 ile ve 10.000'in tamamıyla. Taban çizgisi ağın katkısını gösterir; oradan sonraki artış broker'ın fan-out maliyetidir. Ardından gecikmenin nerede artmaya başladığını bulmak için QOS = 1 ve daha yüksek bir RATE ile tekrarlayın.

Sonuçları okumak

• QoS 0'da sondalarda teslim edilen < beklenen ise broker bazı bağlantılar için mesaj düşürmüştür; genellikle çıkış kuyrukları dolu olduğu için. QoS 1 ile tekrar deneyin ve broker loglarını inceleyin.
• p50 sabit kalırken p99 gecikmesinin subscriber sayısıyla artması bir noktaya kadar normaldir: broker socket'lere sırayla yazar. Önemli olan p99'un gereksiniminizin içinde kalıp kalmadığıdır.
• Yüksek publish hızlarında yayılımın keskin artması, broker'ın (veya ağının) bir sonraki mesaj gelmeden tüm kopyaları boşaltamadığı anlamına gelir. Sürdürülebilir fan-out hızını aştınız.
• Bench makineleri beklenenden düşük alım hızı raporluyorsa önce onların CPU ve ağını kontrol edin. Saniyede 100.000 mesaj almak bir yük üreticisi için gerçek bir iştir.
• Birkaç subscriber geride kalıp bağlantısı kesiliyorsa yavaş tüketiciler yazısına bakın. Üretimde tek bir yavaş cihaz herkesi geciktirmemelidir.

Fan-out için topic tasarımı

Test genellikle bir tasarım tartışmasıyla biter. Bazı desenler iyi sonuç verir:

1. Broadcast'i yalnızca herkesin gerçekten ihtiyaç duyduğu yerde kullanın. Her cihazın alması gereken mesajlar için fleet/all/config; bunları küçük tutun.
2. Tek dev broadcast yerine grup topic'leri. fleet/region/eu/config veya fleet/model/x200/firmware mesaj başına fan-out'u azaltır ve kademeli dağıtıma izin verir.
3. Güncel durumu retain edin. Yapılandırmayı retain bayrağıyla publish edin; böylece sonradan yeniden bağlanan cihazlar son değeri hemen alır (retained mesajlar yazısına bakın). Geç gelen bir subscriber'ın bunu aldığını test edin.
4. QoS'u sonuca göre seçin. Değiştirilebilir güncellemeler için QoS 0; ulaşması şart olan komutlar için QoS 1, broker'ın subscriber başına takibi karşılanabilir kalacak şekilde boyutlandırılmış olarak.
5. Burada shared subscription kullanmayın. Bunlar yük dengeler; broadcast yapmaz.

Kısa bir kontrol listesi

1. Hedef filonuz ve publish hızınız için çıkış bant genişliğini hesaplayın.
2. Subscriber'ları yeterli sayıda yük makinesine dağıtın; file descriptor sınırlarını yükseltin.
3. Önce tüm subscriber'ları bağlayın, sonra publish edin; bağlantı testlerini ayrı tutun.
4. Gecikme ve yayılımı yalnızca toplam alım hızlarıyla değil, sondalarla ölçün.
5. QoS 0 ve QoS 1 ile, TLS'li ve TLS'siz tekrarlayın.
6. Broker'ın CPU, bellek ve çıkış ağı değerlerini istemci tarafı sayıların yanına kaydedin.

Fan-out, küçük bir mesajı büyük bir iş ve trafik patlamasına dönüştürür. Önce bant genişliği hesabını yapın, publish etmeden önce subscriber'ları bağlayın ve sonucu en yavaş subscriber'a göre değerlendirin; kullanıcılarınızın fark edeceği o olacaktır.

Sıkça sorulan sorular