MQTT Throughput Testi: Saniyedeki Mesaj Sayısı Açıklandı

"Saniyedeki mesaj sayısı" en çok alıntılanan ve en çok yanlış anlaşılan MQTT değeridir. Bir broker publisher'lardan 50.000 msg/s kabul edip yine de yalnızca yarısını teslim edebilir ya da hepsini iki saniye gecikmeyle teslim edebilir. Faydalı bir MQTT throughput testi, kendi payload'larınız ve topic yapınızla, gecikme sabit kalırken ve hiçbir şey kaybolmadan sistemin girişte ve çıkışta sürdürebildiği hızı ölçer.

Bu rehber değerleri tanımlar, sürdürülebilir hızı adım adım nasıl bulacağınızı gösterir ve sınırı aştığınızı söyleyen backpressure sinyallerini açıklar. Bağlantı sayıları ve gecikmeyle ilgili genel resim için MQTT broker yük testi yazısına bakın.

Giriş, çıkış ve fan-out

Fan-out katsayısı en çok unutulan kısımdır. Topic'inde 20 subscriber bulunan ve 1.000 msg/s gönderen tek bir publisher, 20.000 msg/s çıkış demektir. Broker'ın iş yükü girişten çok çıkışa yakındır; bu yüzden her zaman ikisini de belirtin. Uç durumlar için ayrı fan-out testi ve fan-in testi yazılarına bakın.

Tepe ve sürdürülebilir throughput

Her sistem, gerçek kapasitesinin üzerindeki bir patlamayı bir süre emebilir: socket buffer'ları, istemci kuyrukları ve broker kuyrukları dolar ve sorunu gizler. Beş saniyelik bir test broker'ı değil, bu buffer'ları ölçer. Sürdürülebilir hız ancak her yük seviyesini kuyrukların kararlı duruma gelmesine yetecek kadar, genellikle birkaç dakika, tuttuğunuzda ortaya çıkar.

Yöntem bir basamaklı rampadır:

1. Sistemin kesinlikle kaldırdığı bir hızla başlayın (örneğin beklenen tepe değerin %20'si).
2. 2–5 dakika tutun; giriş, çıkış, p99 gecikmesi ve hataları kaydedin.
3. Sabit bir adımla artırın (örneğin +%25) ve yeniden tutun.
4. Çıkış girişi izlemeyi bıraktığında, p99 gecikmesi bir basamak içinde artmaya devam ettiğinde veya hatalar göründüğünde durun.
5. Son temiz basamak sizin sürdürülebilir hızınızdır. Üretimi bunun rahat bir pay altında planlayın.

Sunulan hızı gecikmeye karşı çizdiğinizde, düz bir çizginin ardından gecikmenin yukarı kıvrıldığı bir "dirsek" görürsünüz. Aradığınız değer bu dirsektir; yük üreticisinin itebildiği maksimum hız değil.

emqtt-bench ile rampa

emqtt-bench tek bir makineden yüksek hızlar üretir. Önce subscriber'ları, ardından artan hızlarda publisher'ları başlatın. Publish aralığı -I istemci başına ve milisaniye cinsindendir; yani toplam hız istemci × 1000 / I olur:

docker run --rm -it emqx/emqtt-bench sub \
  -h broker.connectmq.com -p 1883 \
  -c 50 -t 'testmqtt/tput/%i' -q 1 \
  -u "$MQTT_USER" -P "$MQTT_PASS"

docker run --rm -it emqx/emqtt-bench pub \
  -h broker.connectmq.com -p 1883 \
  -c 50 -I 50 -t 'testmqtt/tput/%i' -s 256 -q 1 \
  -u "$MQTT_USER" -P "$MQTT_PASS"

Sonraki basamaklarda -I değerini düşürün (50 → 40 → 30 → 20 → 10 ms) veya istemci ekleyin. emqtt-bench her saniye gönderilen ve alınan hızları yazdırır; bunları basamak başına bir tabloya aktarın. Burada her publisher'ın kendi subscriber'ı olduğu için fan-out katsayısı 1'dir ve çıkış girişe eşit olmalıdır.

Boşluklar ve gecikme için Python sondası

Hız sayaçları iki şeyi gizler: hangi mesajların kaybolduğunu ve kalanların ne kadar geç geldiğini. Bu küçük paho-mqtt 2.x sondasını benchmark'ın yanında çalıştırın. Mütevazı bir hızda sıra numaralı ve zaman damgalı mesajlar publish eder ve her saniye boşlukları ve gecikmeyi raporlar; bench çalışırken sondanın gecikmesinin artması sizin backpressure sinyalinizdir:

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

HOST, PORT = "broker.connectmq.com", 1883
TOPIC = f"testmqtt/tput/probe-{uuid.uuid4().hex[:6]}"
RATE = 100                      # saniyedeki sonda mesaji
stats = {"n": 0, "last": -1, "gaps": 0, "lag": []}

def on_message(client, userdata, msg):
    m = json.loads(msg.payload)
    if m["i"] > stats["last"] + 1:
        stats["gaps"] += m["i"] - stats["last"] - 1
    stats["last"] = max(stats["last"], m["i"])
    stats["n"] += 1
    stats["lag"].append((time.perf_counter() - m["t"]) * 1000)

def client(name):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{name}-{uuid.uuid4().hex[:6]}")
    c.connect(HOST, PORT)
    c.loop_start()
    return c

sub = client("probe-sub")
sub.on_message = on_message
sub.subscribe(TOPIC, qos=1)
pub = client("probe-pub")
time.sleep(1)

i, next_report = 0, time.time() + 1
while True:
    pub.publish(TOPIC, json.dumps({"i": i, "t": time.perf_counter()}), qos=1)
    i += 1
    time.sleep(1 / RATE)
    if time.time() >= next_report:
        lag = sorted(stats["lag"]) or [0]
        print(f"rx {stats['n']:>5}/s  gaps {stats['gaps']}  "
              f"lag p50 {lag[len(lag)//2]:.1f} ms  p99 {lag[int(len(lag)*0.99)-1]:.1f} ms")
        stats.update(n=0, lag=[])
        next_report += 1

Tek bir Python süreci saniyede on binlerce mesaj üretmez; bu benchmark aracının işidir. Sonda bir termometredir, ısıtıcı değil.

Backpressure'ı tanımak

MQTT'de genel bir akış kontrolü sinyali yoktur; backpressure bir yerlerde büyüyen kuyruklar olarak ortaya çıkar. Nereye bakacağınızı bilin:

QoS 1 ve 2'de asıl sınırlayıcı inflight penceresidir: bir istemci yalnızca belirli sayıda onaylanmamış mesajı açıkta tutabilir (paho'da max_inflight_messages_set(), MQTT 5'te Receive Maximum). Yolda 20 mesaj ve 50 ms gidiş-dönüş süresiyle tek bir bağlantı, broker ne kadar hızlı olursa olsun 400 msg/s civarında tıkanır. Bu yüzden uzak bağlantılar üzerindeki throughput testleri aynı bölgedekilerden kötü görünür ve tavanı yükselten şey daha hızlı bir döngü değil, daha fazla bağlantıdır. Mekanizmayı Receive Maximum ve akış kontrolü yazısı açıklar.

msg/s'den bant genişliğine

Throughput aynı zamanda bir ağ bütçesidir. Bir MQTT PUBLISH kabaca şunlara mal olur: sabit başlık (2–5 bayt) + topic uzunluğu + 2 bayt + paket tanımlayıcı (QoS > 0 ise 2 bayt) + MQTT 5 özellikleri + payload. Buna TCP/IP başlıklarını ve kullanılıyorsa TLS kayıt ek yükünü ekleyin. QoS 1, ters yönde 4 baytlık bir PUBACK ekler.

Örnek hesap: 256 baytlık payload, 20 baytlık topic, QoS 1 → TCP/IP öncesi PUBLISH başına yaklaşık 285 bayt. 5.000 msg/s çıkışta bu yaklaşık 1,4 MB/s, yani IP ve TLS ek yükü hariç kabaca 11–12 Mbit/s MQTT trafiği eder. Küçük mesajlarda ek yük baskındır; bu yüzden birden fazla ölçümü tek mesajda toplamak, etkin throughput'u çoğu zaman herhangi bir broker ayarından daha fazla artırır. Paket boyutu hesaplayıcı bu hesabı kendi topic ve payload'ınız için yapar.

Sonucu raporlamak

Bir throughput sonucu ancak koşullarını belirtiyorsa yeniden kullanılabilir. En azından şunları kaydedin: broker sürümü ve yapılandırması, QoS, payload boyutu, topic sayısı, fan-out katsayısı, publisher ve subscriber bağlantı sayıları, TLS açık mı kapalı mı, basamak başına süre ve yük üreticisinin bölgesi. Ardından tepe değeri değil, sürdürülebilir giriş ve çıkış hızlarını p99 gecikmeleriyle birlikte raporlayın. Daha sonra çalıştırmaları karşılaştırırken benchmark sonuçlarını yorumlama yazısı yaygın tuzakları ele alır.

Rampaya başlamadan önce kimlik bilgilerini ve ACL'leri doğrulamak için topic'lerinizi online MQTT istemcisiyle kısa bir smoke testten geçirin; yük altındaki bir yetki hatası tam olarak bir throughput sorunu gibi görünür.

Throughput testi en büyük sayıya ulaşma yarışı değil, dirseği arama işidir. Basamaklarla artırın, her basamağı yeterince uzun tutun, yalnızca sayaçlara değil gecikmeye ve boşluklara bakın ve çıkışı girişle birlikte raporlayın; kapasite planlamasını bu değere göre yapabilirsiniz.

Sıkça sorulan sorular