MQTT Yük Testi Araçları: emqtt-bench, MQTTX Bench ve Locust Karşılaştırması

MQTT yük testi araçları, binlerce cihaz aynı anda bağlandığında, bir firmware hatası bir topic'i mesaja boğduğunda veya bir backend subscriber geride kaldığında sisteminizin ayakta kalıp kalmadığını gösterir. Bu rehber emqtt-bench, MQTTX Bench ve Locust'u karşılaştırıyor, her biri için komutları gösteriyor ve uçtan uca gecikmeyi ve mesaj kaybını ölçen küçük bir Python probu ekliyor. Ayrıca hangi metriklerin önemli olduğunu ve sorumlu bir şekilde nasıl test edeceğinizi öğreneceksiniz: her zaman kendi özel broker'ınıza karşı, asla paylaşılan public bir broker'a karşı değil.

İlk Kural: Kendi Broker'ınızı Test Edin

Mevcut MQTT yük testi araçları arasından seçim yapmak işin kolay kısmıdır; asıl önemli olan onları doğru hedefe yöneltmektir. Yük testleri saldırıya benzeyen trafik üretir. Bunları yalnızca sahibi olduğunuz veya test etmeye açıkça yetkili olduğunuz altyapıya karşı çalıştırın.

• Public broker'a yük testi yapmayın. broker.connectmq.com, client başına sıkı limitleri olan paylaşımlı bir test servisidir (client başına saniyede 10 mesaj, client başına 10 subscription, 64 KB mesaj). Orada yapılan bir yük testi sisteminizi değil bu limitleri ölçer ve servisi diğer herkes için yavaşlatır.
• Teste uygun boyutta özel bir broker kullanın. ConnectMQ üzerinde yük testi için ayrı bir broker oluşturun; böylece üretimdeki cihazlar etkilenmez.
• Başlamadan önce planınızın kotalarını bilin. Ücretsiz Serverless planı en fazla 25 bağlantıya ve QoS 1'e kadar izin verir; ücretli Serverless en fazla 1.000 bağlantı, QoS 2 ve 1 MB mesaj sunar; Dedicated ise izole bir cluster üzerinde çalışır. Bir kotanın aşılması broker hatası değildir: MQTT 5 client'ları 0x97 (quota exceeded) reason code'unu alır ve testiniz bunu gerçek hatalardan ayrı kaydetmelidir.
• Client ID kurallarını kontrol edin. ConnectMQ client ID'leri verdiği kimlik bilgilerine bağlarken, yük araçları ID'leri bir önek ve bir sayaçtan üretir. Test kimlik bilgilerinizin hangi client ID'lere izin verdiğini portalda doğrulayın ve aracın client ID seçeneğini buna göre ayarlayın; aksi halde her bağlantı yetkisiz olarak reddedilir.
• Koruma katmanlarını göz önünde bulundurun. Yönetilen broker'lar flood koruması uygular ve tekrarlanan başarısız girişlerden sonra IP'leri engeller. Yanlış kimlik bilgileriyle hatalı yapılandırılmış bir test, test makinenizin IP'sinin engellenmesine yol açabilir; bu yüzden kimlik bilgilerini önce tek bir client ile doğrulayın.

Neleri Ölçmeli?

Tek başına verim (throughput) yanıltıcı bir sayıdır. Aşağıdakileri toplayın ve herhangi bir şey çalıştırmadan önce başarı kriterlerinize karar verin:

Yük üretecini her zaman izleyin. CPU'su yüzde 100'deyse gecikme rakamları broker'ı değil dizüstü bilgisayarınızı yansıtır. Büyük testleri birkaç makineye bölün.

emqtt-bench

emqtt-bench, EMQX projesinden çıkan, client başına çok az bellekle çok sayıda bağlantı açmak için tasarlanmış hafif bir Erlang aracıdır. Üç modu vardır: conn, pub ve sub.

# Ramp up 500 connections, one every 10 ms
emqtt_bench conn -h YOUR_BROKER_HOST -p 1883 -c 500 -i 10 \
  -u YOUR_USERNAME -P YOUR_PASSWORD

# 50 subscribers on a wildcard, QoS 1
emqtt_bench sub -h YOUR_BROKER_HOST -p 1883 -c 50 -i 10 \
  -t 'loadtest/#' -q 1 -u YOUR_USERNAME -P YOUR_PASSWORD

# 200 publishers, each sending a 256-byte message every 1000 ms to its own topic
emqtt_bench pub -h YOUR_BROKER_HOST -p 1883 -c 200 -i 10 -I 1000 \
  -t 'loadtest/device/%i' -s 256 -q 1 -u YOUR_USERNAME -P YOUR_PASSWORD

Temel parametreler: -c client sayısı, -i yeni bağlantılar arasındaki milisaniye cinsinden aralık, -I client başına publish aralığı, -t %i (client sıra numarası) ve %c (client ID) yer tutucularıyla topic, -s payload boyutu, -q QoS. TLS'i test etmek için --ssl ve 8883 portunu ekleyin; bunu yapmalısınız, çünkü yeniden bağlanma fırtınalarında asıl maliyet çoğu zaman TLS handshake'leridir. Araç birkaç saniyede bir hızları yazdırır.

Güçlü yanları: bağlantı başına çok verimlidir; bağlantı sayısı ve yeniden bağlanma fırtınası testleri için idealdir. Zayıf yanları: payload'lar yapaydır ve gecikme ölçümü sınırlıdır; bu yüzden onu bir probla (aşağıda) birlikte kullanın.

MQTTX Bench

MQTTX CLI, aynı üç moda sahip ve daha anlaşılır çıktı veren bir bench komutu içerir. Ekibiniz zaten MQTTX kullanıyorsa başlamanın en kolay yolu budur.

mqttx bench conn -c 200 -i 10 -h YOUR_BROKER_HOST -p 8883 -l mqtts \
  -u YOUR_USERNAME -P YOUR_PASSWORD

mqttx bench sub -c 20 -t 'loadtest/#' -q 1 -h YOUR_BROKER_HOST -p 8883 -l mqtts \
  -u YOUR_USERNAME -P YOUR_PASSWORD

mqttx bench pub -c 100 -i 10 -im 1000 -t 'loadtest/device/%i' -q 1 \
  -m '{"temp": 21.5}' -h YOUR_BROKER_HOST -p 8883 -l mqtts \
  -u YOUR_USERNAME -P YOUR_PASSWORD

Burada -i bağlantı aralığı, -im ise client başına mesaj aralığıdır. Node.js üzerinde çalıştığı için her client emqtt-bench'e göre daha fazla bellek tüketir; bu nedenle çok yüksek bağlantı sayıları daha fazla üreteç makinesi gerektirir. Seçenekler geliştiği için kendi sürümünüzde mqttx bench pub --help çıktısını kontrol edin.

Locust

Locust, Python tabanlı bir yük testi framework'üdür. Yerel MQTT desteği yoktur, ancak paho-mqtt etrafında bir User sınıfı yazıp sonuçları Locust'un web arayüzüne raporlayabilirsiniz. Bu da onu en esnek seçenek yapar: gerçekçi payload'lar, cihaza özgü davranışlar, artış şekilleri ve doğrulamalar tamamen Python kodundan ibarettir.

# locustfile.py  (pip install locust paho-mqtt)
import json
import os
import random
import time
import uuid

import paho.mqtt.client as mqtt
from locust import User, between, events, task

HOST = os.environ.get("MQTT_HOST", "YOUR_BROKER_HOST")
USER = os.environ.get("MQTT_USER", "YOUR_USERNAME")
PASS = os.environ.get("MQTT_PASS", "YOUR_PASSWORD")


class MqttDevice(User):
    wait_time = between(1, 2)

    def on_start(self):
        self.device_id = uuid.uuid4().hex[:8]
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                                  client_id=f"lt-{self.device_id}",
                                  protocol=mqtt.MQTTv5)
        self.client.username_pw_set(USER, PASS)
        self.client.tls_set()
        start = time.perf_counter()
        try:
            self.client.connect(HOST, 8883, keepalive=30)
            self.client.loop_start()
            events.request.fire(request_type="MQTT", name="connect",
                                response_time=(time.perf_counter() - start) * 1000,
                                response_length=0, exception=None, context={})
        except Exception as exc:
            events.request.fire(request_type="MQTT", name="connect",
                                response_time=(time.perf_counter() - start) * 1000,
                                response_length=0, exception=exc, context={})
            raise

    @task
    def publish_telemetry(self):
        payload = json.dumps({"id": self.device_id,
                              "temp": round(random.uniform(18, 26), 2),
                              "ts": time.time()})
        start = time.perf_counter()
        info = self.client.publish(f"loadtest/device/{self.device_id}", payload, qos=1)
        exc = None
        try:
            info.wait_for_publish(timeout=5)  # QoS 1'de PUBACK'i bekler
            if not info.is_published():
                exc = TimeoutError("no PUBACK within 5 s")
        except Exception as e:
            exc = e
        events.request.fire(request_type="MQTT", name="publish qos1",
                            response_time=(time.perf_counter() - start) * 1000,
                            response_length=len(payload), exception=exc, context={})

    def on_stop(self):
        self.client.loop_stop()
        self.client.disconnect()

Bunu locust -f locustfile.py --headless -u 100 -r 5 --run-time 10m ile çalıştırın; burada -u simüle edilen cihaz sayısı, -r ise saniyedeki başlatma hızıdır. Her cihaz kendi paho ağ thread'ini çalıştırdığından Locust, bağlantı başına emqtt-bench'ten daha ağırdır; onu ham bağlantı sayıları yerine yüzlerce cihazla davranışı zengin testler için kullanın. Buradaki publish süresi, broker'ın yanıt verme hızı için iyi bir gösterge olan PUBACK gidiş-dönüş süresini ölçer.

Python ile Gecikme ve Kayıp Probu

Yukarıdaki araçların hiçbiri varsayılan olarak güvenilir uçtan uca gecikme ve kayıp verisi sunmaz. Bu probu herhangi bir yük testiyle birlikte çalıştırın: sıra numaralı, zaman damgalı mesajlar publish eder ve gelenleri ölçer.

# probe.py  (pip install paho-mqtt)
import json, os, statistics, time, uuid
import paho.mqtt.client as mqtt

HOST, PORT = os.environ.get("MQTT_HOST", "YOUR_BROKER_HOST"), 8883
USER, PASS = os.environ.get("MQTT_USER", "YOUR_USERNAME"), os.environ.get("MQTT_PASS", "YOUR_PASSWORD")
TOPIC = f"loadtest/probe/{uuid.uuid4().hex[:6]}"
COUNT, INTERVAL = 600, 0.5

latencies, seen = [], set()

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    latencies.append((time.time() - data["ts"]) * 1000)
    seen.add(data["seq"])

def make(name):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"probe-{name}-{uuid.uuid4().hex[:6]}")
    c.username_pw_set(USER, PASS)
    c.tls_set()
    c.connect(HOST, PORT, keepalive=30)
    c.loop_start()
    return c

sub = make("sub")
sub.on_message = on_message
sub.subscribe(TOPIC, qos=1)
time.sleep(1)
pub = make("pub")

for seq in range(COUNT):
    pub.publish(TOPIC, json.dumps({"seq": seq, "ts": time.time()}), qos=1)
    time.sleep(INTERVAL)

time.sleep(5)
q = statistics.quantiles(latencies, n=100)
print(f"received {len(seen)}/{COUNT}, lost {COUNT - len(seen)}")
print(f"latency ms p50={q[49]:.1f} p95={q[94]:.1f} p99={q[98]:.1f} max={max(latencies):.1f}")

Publisher ve subscriber aynı makinede çalıştığı için saat kayması gecikmeyi bozmaz. Bir taban çizgisi için probu önce boştaki bir broker'da bir kez, ardından her yük aşamasında çalıştırın; aradaki fark yükün maliyetidir. Kimlik bilgileriniz yalnızca belirli client ID'lere izin veriyorsa üretilen ID'leri size verilenlerle değiştirin.

Karşılaştırma Özeti

Sorumlu Bir Test Planı

1. Hedefi yazın: örneğin "2.000 cihaz 2 dakika içinde yeniden bağlanır; p95 gecikme üzerinde anlaşılan eşiğin altında kalır".
2. Tek bir client ile doğrulayın: kimlik bilgileri, TLS, topic yapısı, client ID kuralları.
3. Taban çizgisi: probu boştaki bir broker'da çalıştırın.
4. Kademeli artırın: hedefin yüzde 10, 25, 50 ve 100'ü; her aşamayı birkaç dakika sürdürün.
5. Zorlu senaryoları test edin: herkesin aynı anda yeniden bağlanması, okumayı bırakan bir subscriber, aşırı büyük payload'lar.
6. Kota yanıtlarını ayrı kaydedin: 0x97 bir plan limitine ulaştığınız anlamına gelir; bu bir çökme değil, bilgidir.
7. Temizlik yapın: retained test mesajlarını silin ve tüm client'ların bağlantısını kesin.

Sonuçlar mevcut planınızın sunduğundan daha fazla bağlantıya veya izolasyona ihtiyacınız olduğunu gösteriyorsa, bu bir üst seviyeye geçme sinyalidir; örneğin Serverless'tan Dedicated'a. Yönetilen MQTT broker genel bakışı ve bulut MQTT broker seçim rehberi bu ödünleşimleri açıklar. Yük testinden önceki fonksiyonel testler için MQTT client'larını test etme yazısına bakın; düşük hacimde gerçekçi trafik üretmek içinse bir MQTT cihaz simülatörü kullanın. Komut satırı temelleri MQTT CLI araçları yazısında ele alınır; başka bir dilde özel bir yük üreteci yazmak isterseniz MQTT client kütüphaneleri rehberi yardımcı olur.

ConnectMQ ile Başlayın

app.connectmq.com üzerinde ücretsiz Serverless planıyla, kredi kartı gerekmeden ayrı bir test broker'ı oluşturun ve test betiklerinizi küçük ölçekte doğrulayın. Daha büyük çalıştırmalara hazır olduğunuzda aynı broker kurulumunu yükseltin ve test sürerken bağlantıları izlemek için portal panelini açık tutun.

Sık Sorulan Sorular

Public MQTT broker'a yük testi yapabilir miyim?

Hayır. Public broker'lar paylaşımlıdır ve client başına hız sınırlıdır; bu nedenle sonuçlar anlamsız olur ve diğer kullanıcılar etkilenir. Kendi özel broker'ınızı kullanın.

En iyi MQTT yük testi aracı hangisi?

Yüksek bağlantı sayıları için emqtt-bench, hızlı ve okunabilir testler için MQTTX Bench, gerçekçi cihaz davranışına ihtiyaç duyduğunuzda ise Locust. Birçok ekip ölçek için emqtt-bench'i, gecikme için de küçük bir probu birlikte kullanır.

Yük testi client'larım neden quota exceeded hatasıyla reddediliyor?

Planınızın bir limitine, örneğin maksimum bağlantı sayısına ulaştınız. MQTT 5 client'ları 0x97 reason code'unu alır; yükü azaltın veya daha yüksek limitlere sahip bir plana geçin.

Yük testlerinde TLS kullanılmalı mı?

Evet, üretimde TLS kullanılıyorsa. Yeniden bağlanma fırtınalarında handshake'ler her iki tarafta da ciddi CPU yükü ekler; TLS olmadan test etmek bu maliyeti gizler.