MQTT Fan-In Testi: Binlerce Cihaz, Tek Backend

Fan-in, IoT'nin gündelik biçimidir: binlerce cihazın her biri biraz telemetri publish eder ve tek bir backend hepsini saklamak veya işlemek için devices/+/telemetry topic'ine abone olur. Tek tek bakıldığında cihazlar sessizdir; birlikte ise her şeyi tek bir abonelik, tek bir TCP bağlantısı ve tek bir süreç üzerinden gönderirler. Bir fan-in testi, bu tüketicinin bir backlog oluşmadan önce kaç cihazı kaldırabildiğini ve kaldıramadığında nasıl ölçekleneceğini ortaya çıkarır.

Bu rehber emqtt-bench ile bir filo simüle eder, tüketici tarafında backlog ve gecikmeyi bir Python betiğiyle ölçer ve shared subscription'ların yükü nasıl dağıttığını gösterir. Ters desen fan-out testi yazısında ele alınır.

Fan-in darboğazları nerede saklanır

Broker genellikle fan-in girişini rahatça kaldırır; çok sayıda küçük mesajı tek bir subscriber'a yönlendirmek ucuzdur. Sorun daha ileridedir:

Son ikisi broker'ın tek bir oturum için koyduğu sınırlardır; örneğin Mosquitto'daki max_queued_messages ve max_inflight_messages. Tüketici geride kaldığında broker onun için mesajları kuyruğa alır; kuyruk dolduğunda broker'a bağlı olarak mesaj düşürmeye veya bağlantıyı kesmeye başlar. Farklı broker'ların nasıl davrandığı için yavaş tüketiciler yazısına bakın.

Adım 1: filoyu simüle edin

emqtt-bench tek bir makineden binlerce cihazı canlandırabilir. Topic yapısının üretimle aynı olması için her istemciye %i (istemci indeksi) ile kendi topic'ini verin. -I aralığı istemci başına ve milisaniye cinsindendir; yani -I 5000 ile 5.000 cihazın her biri 5 saniyede bir mesaj gönderir, toplamda 1.000 msg/s:

docker run --rm -it --ulimit nofile=65535:65535 emqx/emqtt-bench pub \
  -h broker.connectmq.com -p 1883 \
  -c 5000 -i 5 -I 5000 \
  -t 'testmqtt/fanin/%i/telemetry' -s 200 -q 1 \
  -u "$MQTT_USER" -P "$MQTT_PASS"

Yükü artırmak için bağlantı eklemek yerine -I değerini düşürün (5000 → 2000 → 1000 ms, 2.500 ve 5.000 msg/s verir): test ettiğiniz şey bağlantı sayısı değil, tüketicidir. Başlamadan önce wildcard'ınızın cihaz topic'leriyle gerçekten eşleştiğini topic matcher ile kontrol edin; filtredeki bir yazım hatası, hiçbir şey almayan bir tüketiciyle tamamen aynı görünür.

Adım 2: backlog'unu gösteren bir tüketici

Üretim tarzı bir tüketici on_message içinde asla yavaş iş yapmamalıdır: bu, ağ döngüsünü bloke eder; döngü socket'i okumayı bırakır ve backlog'u göremeyeceğiniz broker'a iter. Bu paho-mqtt 2.x tüketicisi her mesajı bir kuyruğa verir, mesaj başına simüle edilmiş bir maliyetle worker thread'lerinde işler ve her saniye alım hızını, işleme hızını ve backlog'u yazdırır:

import json, queue, sys, threading, time, uuid
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
FILTER = "testmqtt/fanin/+/telemetry"
SHARED = "--shared" in sys.argv
TOPIC = f"$share/ingest/{FILTER}" if SHARED else FILTER
WORK_MS = 0.5        # mesaj basina simule edilen maliyet (parse + DB insert)
WORKERS = 4

inbox = queue.Queue(maxsize=100_000)
counts = {"rx": 0, "done": 0}
lags = []

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(TOPIC, qos=1)

def on_message(client, userdata, msg):
    counts["rx"] += 1
    inbox.put(msg.payload)       # ag thread'inde asla yavas is yapmayin

def worker():
    while True:
        payload = inbox.get()
        if payload[:1] == b"{":  # zaman damgali sonda mesajlari
            try:
                lags.append((time.time() - json.loads(payload)["t"]) * 1000)
            except (ValueError, KeyError):
                pass
        time.sleep(WORK_MS / 1000)
        counts["done"] += 1

for _ in range(WORKERS):
    threading.Thread(target=worker, daemon=True).start()

c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                client_id=f"ingest-{uuid.uuid4().hex[:6]}", protocol=mqtt.MQTTv5)
c.on_connect, c.on_message = on_connect, on_message
c.connect(HOST, PORT)
c.loop_start()

while True:
    time.sleep(1)
    l = sorted(lags); lags.clear()
    p99 = f"{l[int(len(l) * 0.99) - 1]:.0f} ms" if l else "-"
    print(f"rx {counts['rx']:>6}/s  done {counts['done']:>6}/s  "
          f"backlog {inbox.qsize():>6}  probe lag p99 {p99}")
    counts.update(rx=0, done=0)

emqtt-bench payload'ları rastgele baytlardan oluşur, dolayısıyla zaman damgası taşıyamaz. Uçtan uca gecikmeyi ölçmek için, tüketiciyle aynı makinede (ikisi de aynı saati kullansın diye) aynı topic ağacına zaman damgalı JSON publish eden küçük bir sonda çalıştırın:

import json, time
import paho.mqtt.client as mqtt

p = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
p.connect("broker.connectmq.com", 1883)
p.loop_start()
while True:
    p.publish("testmqtt/fanin/probe/telemetry", json.dumps({"t": time.time()}), qos=1)
    time.sleep(0.1)

Sayıları okumak

Filoyu her basamakta çalıştırın ve tüketici çıktısını birkaç dakika izleyin:

• rx ≈ done, backlog sıfıra yakın, gecikme sabit: tüketici yetişiyor. Sonraki basamağa geçin.
• rx ≈ giriş ama done < rx, backlog büyüyor: ağ tarafı iyi, darboğaz işleme. Worker ekleyin, yazmaları toplu yapın veya hedefi hızlandırın.
• rx < giriş, backlog küçük: sınır tek bağlantı veya broker tarafındaki oturum (inflight penceresi, istemci başına kuyruk). Shared subscription tam burada işe yarar.
• Gecikme sürekli artıyor: henüz hiçbir şey kaybolmasa bile sürdürülebilir hızın üzerindesiniz; kuyruk yalnızca uzuyor.
• Tüketici yük altında kopuyor: broker istemci başına sınırına ulaştı ya da ağ thread'i bloke olduğu için keep-alive süresi doldu.

Adım 3: shared subscription ile ölçeklendirin

Sınır tek bir tüketici bağlantısıysa, aynı shared subscription grubunda birden fazla tüketici başlatın. Broker bu durumda her mesajı $share/ingest/... grubunun yalnızca bir üyesine teslim eder:

python ingest.py --shared   # terminal 1
python ingest.py --shared   # terminal 2 (ve devami)

Daha önce başarısız olan basamağı tekrarlayın. Doğrulanacaklar:

1. Tüketicilerdeki done toplamı filonun girişine eşit: hiçbir mesaj kaybolmuyor.
2. Kopya yok: $share olmadan her tüketici her mesajı alır ve her şeyi iki kez işlersiniz. Payload'ınız benzersiz ID taşıyorsa bunları sayın.
3. Dengeli dağılım: broker'lar farklı stratejiler kullanır (round robin, rastgele, istemci başına sticky). Yükün kabaca dengeli olduğunu kontrol edin.
4. Failover: çalışma sırasında bir tüketiciyi sonlandırın; diğerleri onun payını üstlenmelidir. Sonlandırılan tüketiciye yolda olan mesajlar QoS'a ve broker'a bağlı olarak yeniden teslim edilebilir veya kaybolabilir.
5. Sıra: cihaz başına sıra grup genelinde garanti edilmez. Önemliyse tasarımınızı buna göre yapın.

Semantik için shared subscriptions yazısına, dağılım ve failover testlerinin ayrıntıları için shared subscription testi yazısına bakın.

Tek sürecin ötesinde

Darboğaz hedefse ve ani yükler kaçınılmazsa (bir kesintiden sonra yeniden bağlanan filo, biriktirdiği çok sayıda veriyi bir anda gönderir), MQTT ile depolama arasına kalıcı bir buffer koyun. MQTT'yi Kafka'ya köprülemek yaygın bir desendir: MQTT'den Kafka'ya yazısına bakın. Ardından ani yük senaryosunu açıkça test edin: filo çalışırken tüketiciyi beş dakika durdurun, yeniden başlatın ve backlog'un erimesinin ne kadar sürdüğünü ölçün. emqtt-bench'in sunduğundan daha gerçekçi cihaz davranışı için gerçek payload'larınızla bir cihaz simülatörü kullanın.

Büyük bir çalıştırmadan önce online MQTT istemcisinde yapacağınız hızlı bir kontrol (testmqtt/fanin/# topic'ine abone olup mesajların gelişini izlemek), filonun düşündüğünüz yere publish ettiğini doğrular.

Fan-in nadiren broker'ı bozar; bozduğu şey tüketicidir. Ham alım hızları yerine backlog'u ve gecikmeyi ölçün, yavaş işleri ağ thread'inden uzak tutun ve tek bağlantı yetmediğinde shared subscription'larla ölçeklendirin.

Sıkça sorulan sorular