QoS 1 ve QoS 2 Teslimat Garantileri Nasıl Test Edilir

MQTT teslimat garantilerini test etmek için sıra numaralı mesajlar yayınlayın, mesajlar yoldayken bağlantıyı koparın ve neyin geldiğini sayın. Hiçbir sıra numarası eksik değilse QoS 1 testi geçer (kopyalar sorun değildir). Hiçbir şey eksik ve hiçbir şey kopya değilse QoS 2 testi geçer. Geri kalan her şey — kalıcı oturumlar, onay paketleri, broker kuyruk limitleri — bu deneyi dürüst kılmakla ilgilidir.

Bu rehber onay akışını nasıl izleyeceğinizi, boşlukları ve kopyaları sayan bir Python test düzeneğini, denemeye değer hata senaryolarını ve sonuçların nasıl okunacağını gösteriyor.

Her seviyenin vaadi

Test sonuçlarının çoğunu iki ayrıntı belirler. Birincisi, QoS atlama (hop) başınadır: publisher'ın QoS'u publisher → broker yolunu, aboneliğin QoS'u broker → subscriber yolunu kapsar ve etkin teslimat QoS'u ikisinden düşük olanıdır. QoS 1 aboneliğe QoS 2 ile yapılan bir publish, QoS 1 teslimattır. İkincisi, garantiler bağlantı kopmasından ancak oturum da hayatta kalırsa sağ çıkar — aksi hâlde broker yoldaki durumu çöpe atar. Teori için QoS seviyeleri rehberine bakın; burada onu test ediyoruz.

Adım 1: onayları izleyin

Herhangi bir şeyi otomatikleştirmeden önce paketlere bakın. Mosquitto client'ları -d ile her kontrol paketini yazdırır:

# Terminal 1: subscribe at QoS 2 with debug output
mosquitto_sub -h broker.connectmq.com -p 1883 -i testmqtt-qos-sub -t 'testmqtt/qos/demo' -q 2 -d

# Terminal 2: publish one QoS 2 message with debug output
mosquitto_pub -h broker.connectmq.com -p 1883 -i testmqtt-qos-pub -t 'testmqtt/qos/demo' -m 'hello' -q 2 -d

Publisher buna benzer bir akış yazdırır:

Client testmqtt-qos-pub sending CONNECT
Client testmqtt-qos-pub received CONNACK (0)
Client testmqtt-qos-pub sending PUBLISH (d0, q2, r0, m1, 'testmqtt/qos/demo', ... (5 bytes))
Client testmqtt-qos-pub received PUBREC (Mid: 1)
Client testmqtt-qos-pub sending PUBREL (m1)
Client testmqtt-qos-pub received PUBCOMP (Mid: 1)
Client testmqtt-qos-pub sending DISCONNECT

d0 DUP bayrağı, q2 QoS, m1 ise packet identifier'dır. Aynısını -q 1 ile tekrarlayın; üç adımlı el sıkışma yerine tek bir PUBACK görmelisiniz. Subscriber tarafında alınan PUBLISH içindeki q değerini kontrol edin — QoS 1 ile abone olduysanız publisher QoS 2 kullansa bile q1 yazar.

Daha derin inceleme için trafiği yakalayın ve Wireshark'ta mqtt.msgtype >= 3 && mqtt.msgtype <= 7 (PUBLISH'ten PUBCOMP'a) ya da yalnızca yeniden gönderimleri görmek için mqtt.dupflag == 1 filtresini kullanın. MQTT 5'te onaylar ayrıca bir reason code taşır: 0x10 (No matching subscribers) içeren bir PUBACK hata değil, başarıdır — kodlara reason code referansından bakabilirsiniz.

Adım 2: boşluk ve kopya sayacı

Düzenek üç parçadan oluşur: aldığı her sıra numarasını kaydeden kalıcı oturumlu bir subscriber, bilinen bir aralığı gönderip onayları sayan bir publisher ve kısa bir analiz betiği.

import sys
import paho.mqtt.client as mqtt

QOS = int(sys.argv[1]) if len(sys.argv) > 1 else 1
TOPIC = "testmqtt/qos-test/seq"
log = open(f"received-q{QOS}.log", "a", buffering=1)  # line-buffered: survives kill -9

def on_connect(client, userdata, flags, reason_code, properties):
    print(f"connected: {reason_code}, session present: {flags.session_present}")
    client.subscribe(TOPIC, qos=QOS)

def on_message(client, userdata, msg):
    log.write(f"{msg.payload.decode()} {int(msg.dup)}\n")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"testmqtt-qos-sub-q{QOS}", clean_session=False)
# client.username_pw_set("username", "password")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=10)
client.loop_forever()   # reconnects automatically after a network error

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

QOS = int(sys.argv[1]) if len(sys.argv) > 1 else 1
N = int(sys.argv[2]) if len(sys.argv) > 2 else 2000
TOPIC = "testmqtt/qos-test/seq"
acked = 0

def on_publish(client, userdata, mid, reason_code, properties):
    global acked
    acked += 1          # QoS 1: on PUBACK, QoS 2: on PUBCOMP

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"testmqtt-qos-pub-q{QOS}", clean_session=False)
client.on_publish = on_publish
client.connect("broker.connectmq.com", 1883, keepalive=10)
client.loop_start()

for seq in range(N):
    client.publish(TOPIC, str(seq), qos=QOS)   # queued by paho while disconnected
    time.sleep(0.005)

deadline = time.time() + 60
while acked < N and time.time() < deadline:
    time.sleep(0.2)
print(f"published {N}, acknowledged {acked}")
client.disconnect()
client.loop_stop()

#!/usr/bin/env bash
LOG=${1:-received-q1.log}; N=${2:-2000}
echo "received lines : $(wc -l < "$LOG")"
echo "unique         : $(cut -d' ' -f1 "$LOG" | sort -u | wc -l)"
echo "duplicates     : $(cut -d' ' -f1 "$LOG" | sort | uniq -d | wc -l)"
echo "with DUP flag  : $(awk '$2 == 1' "$LOG" | wc -l)"
echo "missing        : $(comm -23 <(seq 0 $((N - 1)) | sort) <(cut -d' ' -f1 "$LOG" | sort -u) | wc -l)"

Önce python sub.py 1, sonra python pub.py 1, ardından ./analyze.sh received-q1.log çalıştırın. Hatasız bir çalıştırmayla başlayın: 2.000 benzersiz ve 0 eksik göstermelidir. Temel çalıştırma zaten başarısızsa hata enjekte etmeden önce bunu düzeltin (abonelikte yanlış QoS, bir ACL, mesaj boyutu limiti).

Adım 3: gönderim sırasında bir şeyleri bozun

Her senaryoyu pub.py gönderim yaparken çalıştırın, sonra analiz edin. Her çalıştırmada yeni bir log dosyası kullanın.

BROKER_IP=$(dig +short broker.connectmq.com | head -1)
sudo iptables -I OUTPUT -p tcp -d "$BROKER_IP" --dport 1883 -j DROP
sleep 20
sudo iptables -D OUTPUT -p tcp -d "$BROKER_IP" --dport 1883 -j DROP

Kara delik daha gerçekçi bir hatadır: hiçbir RST gelmez, bu yüzden client durumu yalnızca keep-alive ile fark eder (broker tarafında 10 saniyelik keep-alive'ın yaklaşık 1,5 katı) ve ardından yeniden bağlanır. Yeniden bağlanınca subscriber session present: True yazdırmalı, broker da onaylanmamış mesajları DUP=1 ile, ardından kuyrukta biriken mesajları yeniden göndermelidir. Daha fazla kesinti tekniği için ağ kesintisiyle QoS testi yazısına bakın.

Süreç çökmesi QoS 2 için ilginç olan durumdur. paho-mqtt oturum durumunu bellekte tutar; dolayısıyla kill -9, QoS 2 el sıkışmasının client tarafındaki yarısını siler. Bir kütüphanenin mesajı kodunuza PUBLISH alındığında mı yoksa PUBREL alındığında mı ilettiğine bağlı olarak, yanlış anda gelen bir çökme kopya bir mesaja ya da eksik bir mesaja yol açabilir. Exactly-once, durumunu koruyan iki uç nokta arasındaki protokolün bir özelliğidir — uygulamanız bir çökmeden sonra aynı mesajı iki kez işlememeliyse, kendi depolamanızda bir mesaj ID'si üzerinden tekilleştirme yapın.

Sonuçları okumak

• Ağ kesintisinden sonra QoS 1/2'de eksik mesajlar — oturum korunmamıştır: log'daki session present değerini, client ID'yi (yeniden bağlanırken birebir aynı olmalı) ve MQTT 5 session expiry değerini kontrol edin.
• Yalnızca uzun kesintilerden sonra eksik mesajlar — broker'ın client başına kuyruğu dolmuş ve politika gereği mesajları atmıştır (örneğin Mosquitto'da max_queued_messages). Bu bir QoS hatası değil, kapasite sınırıdır; bkz. mesaj kuyruğu dolu.
• QoS 2'de kopyalar — subscriber'daki etkin QoS'u kontrol edin (mosquitto_sub -d içinde q2), ardından çökmeleri ya da aynı client ID'yi paylaşan ikinci bir client'ı arayın. Diğer nedenler için kopya mesajlar.
• QoS 1'de DUP bayrağı olmayan kopyalar — genellikle publisher yeniden bağlandıktan sonra tekrar göndermiştir; bu broker açısından yeni bir mesajdır. At-least-once için normaldir.
• acknowledged değeri published'dan düşük — publisher birikmiş mesajlar boşalmadan vazgeçmiştir; süre sınırını artırın ya da broker tarafında bir inflight limiti olup olmadığına bakın.

Hızlı bir manuel kontrol

Betik kullanmadan hızlı bir kontrol için online MQTT client üzerinde QoS 1 ile abone olun, bir terminalden mosquitto_pub -q 1 -d ile yayın yapın ve iki tarafta görünen QoS'u karşılaştırın. Ardından tarayıcıdaki client'ın bağlantısını kesin, birkaç mesaj daha yayınlayın ve aynı client ID ile clean session kapalı olarak yeniden bağlanın: kuyruktaki mesajlar gelmelidir.

QoS teslimatlarını tarayıcınızda izleyin

QoS 1 sıfır boşlukla, QoS 2 ise sıfır boşluk ve sıfır kopyayla kanıtlanır — spesifikasyondan varsayılarak değil, gerçek bağlantı kopmaları boyunca sıra numaralarıyla ölçülerek. Düzeneği deponuzda tutun ve broker'ı, client kütüphanesini ya da oturum ayarlarını her değiştirdiğinizde yeniden çalıştırın.

Sıkça sorulan sorular