MQTT Oturum Süresi ve Çevrimdışı Mesaj Kuyrukları Nasıl Test Edilir

Kalıcı bir MQTT oturumu, bir cihazın çevrimdışı olmasına rağmen kaçırdığı her QoS 1 mesajı daha sonra almasını sağlar. Bunun çalışıp çalışmadığı, kolayca yanlış ayarlanan dört şeye bağlıdır: client ID, Clean Start, Session Expiry Interval ve broker'ın kuyruk limitleri. Güvenilir test yöntemi şudur: kalıcı bir oturumla abone olun, bağlantıyı kesin, yayın yapın, yeniden bağlanın, ardından Session Present değerini ve aldığınız mesajları birebir doğrulayın. Sonra oturumun süresinin dolmuş olması gereken bir anda aynısını tekrarlayın.

Bu rehber size mosquitto_sub ile manuel bir test, üç davranışın hepsini kontrol eden bir Python betiği ve karşılaşacağınız hata kalıplarının bir listesini sunuyor.

Bir oturumun hayatta kalması için neler doğru olmalı

MQTT 5'te Session Expiry Interval'ın varsayılan değeri 0'dır. Clean Start'ı false yapıp aralığı unutan bir istemci, bağlantı kesildiği anda oturumunu yine kaybeder. İki protokol sürümü arasındaki fark Clean Session ve Clean Start yazısında, genel kavram ise MQTT kalıcı oturumlar yazısında anlatılıyor.

Broker sizin değerinizi de geçersiz kılabilir: istediğinizden daha azına izin veriyorsa bir MQTT 5 broker'ı CONNACK içinde kendi Session Expiry Interval değerini döndürür. Bunu her zaman loglayın.

Broker bir oturumda neleri saklar

• İstemcinin abonelikleri (QoS ve seçenekleriyle birlikte).
• İstemci çevrimdışıyken gelen QoS 1 ve QoS 2 mesajlar.
• Tamamen onaylanmamış, yoldaki (in-flight) QoS 1 ve QoS 2 mesajlar.
• Broker bu şekilde yapılandırıldıysa isteğe bağlı olarak QoS 0 mesajlar.

Bu listenin dışındaki her şey, örneğin retained mesajlar, oturumdan bağımsız yaşar. Kendi mesaj süresi dolması ayarı olan bir mesaj, oturumdan önce kuyruktan kaybolabilir.

mosquitto_sub ile manuel test

Benzersiz bir client ID ve topic seçip oturumu oluşturun:

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 \
  -i sess-demo-41c2 -c -x 120 -q 1 \
  -t 'testmqtt/session/41c2/#' -v
# press Ctrl+C after "subscribed"

for i in 1 2 3 4 5; do
  mosquitto_pub -h broker.connectmq.com -V mqttv5 -q 1 \
    -t testmqtt/session/41c2/cmd -m "msg $i"
done

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 \
  -i sess-demo-41c2 -c -x 120 -q 1 \
  -t 'testmqtt/session/41c2/#' -v -d

-d ile karşılıklı gönderilen tüm paketleri görürsünüz. Her Mosquitto sürümü Session Present bayrağını yazdırmaz; bu yüzden buradaki kanıt biriken mesajlardır: CONNACK'in hemen ardından sırasıyla msg 1'den msg 5'e kadar. Aşağıdaki Python testi bayrağı açıkça kontrol eder. Şimdi istemciyi durdurun, yeniden yayın yapın, 120 saniyeden uzun bekleyin ve 3. adımı bir kez daha çalıştırın: oturum yok olmuş olmalı ve yeni mesajların hiçbiri gelmemelidir.

Burada iki şey kafa karıştırır. Birincisi, 3. adımdaki topic biriken mesajlar için önemli değildir: onları saklanan abonelik teslim eder. İkincisi, başka bir istemci aynı ID'yi kullanırken 3. adımı çalıştırmak ikisinden birini düşürür. Bu bir oturum devralma (session takeover) durumudur, oturum testi değildir.

Python ile otomatik oturum testi

Bu paho-mqtt 2.x betiği tek seferde üç test yapar: oturum hayatta kalır ve mesajları sırayla kuyruğa alır, oturumun süresi gerçekten dolar ve kuyruk taştığında ne olur.

import sys, time, uuid
import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

HOST, PORT = "broker.connectmq.com", 1883
RUN = uuid.uuid4().hex[:8]
BURST = int(sys.argv[1]) if len(sys.argv) > 1 else 2000   # for the queue-limit test

def connect(cid, clean_start, session_expiry, inbox=None):
    """Connect with MQTT 5, wait for CONNACK, return (client, connack state)."""
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=cid, protocol=mqtt.MQTTv5)
    state = {}
    c.on_connect = lambda cl, u, flags, rc, props: state.update(
        present=flags.session_present, rc=str(rc),
        expiry=getattr(props, "SessionExpiryInterval", None))   # set only if the broker overrides it
    if inbox is not None:
        c.on_message = lambda cl, u, msg: inbox.append(int(msg.payload))
    props = Properties(PacketTypes.CONNECT)
    props.SessionExpiryInterval = session_expiry
    c.connect(HOST, PORT, clean_start=clean_start, properties=props)
    c.loop_start()
    for _ in range(100):
        if state: break
        time.sleep(0.05)
    return c, state

def close(c, end_session=False):
    props = None
    if end_session:
        props = Properties(PacketTypes.DISCONNECT)
        props.SessionExpiryInterval = 0
    c.disconnect(properties=props)
    c.loop_stop()

def go_offline(cid, topic, session_expiry):
    c, state = connect(cid, True, session_expiry)
    c.subscribe(topic, qos=1)
    time.sleep(1)
    close(c)
    return state

def publish(topic, count):
    pub, _ = connect(f"sess-pub-{RUN}", True, 0)
    info = None
    for i in range(count):
        info = pub.publish(topic, str(i), qos=1)
    info.wait_for_publish(timeout=60)
    close(pub)

def drain(inbox, quiet=2.0):
    """Wait until no new message has arrived for `quiet` seconds."""
    last, since = -1, time.time()
    while time.time() - since < quiet:
        if len(inbox) != last:
            last, since = len(inbox), time.time()
        time.sleep(0.1)

def report(name, ok, detail=""):
    print(f"{'PASS' if ok else 'FAIL'} {name} {detail}")

# Test 1: the session survives a disconnect and queues QoS 1 messages
cid, topic = f"sess-a-{RUN}", f"testmqtt/session/{RUN}/a"
go_offline(cid, topic, 300)
publish(topic, 50)
inbox = []
c, s = connect(cid, False, 300, inbox)
drain(inbox)
close(c, end_session=True)
report("session present after reconnect", s.get("present") is True, s)
report("all 50 queued messages delivered in order", inbox == list(range(50)), f"got {len(inbox)}")
if not s.get("present"):
    raise SystemExit("ABORT: the broker does not keep sessions, the other tests are meaningless")

# Test 2: the session really expires
cid, topic = f"sess-b-{RUN}", f"testmqtt/session/{RUN}/b"
go_offline(cid, topic, 5)
time.sleep(10)
c, s = connect(cid, False, 5)
close(c, end_session=True)
report("session gone after 5 s expiry", s.get("present") is False, s)

# Test 3: what happens when the offline queue overflows
cid, topic = f"sess-c-{RUN}", f"testmqtt/session/{RUN}/c"
go_offline(cid, topic, 300)
publish(topic, BURST)
inbox = []
c, s = connect(cid, False, 300, inbox)
drain(inbox, quiet=3)
close(c, end_session=True)
kept = f"first={inbox[0]} last={inbox[-1]}" if inbox else ""
print(f"INFO queue limit: published {BURST}, delivered {len(inbox)} {kept}")

python session_test.py ile çalıştırın ya da daha büyük bir patlama verin, örneğin python session_test.py 5000. Oturumları tutan bir broker'da sağlıklı bir çalıştırma şöyle görünür:

PASS session present after reconnect {'present': True, 'rc': 'Success', 'expiry': None}
PASS all 50 queued messages delivered in order got 50
PASS session gone after 5 s expiry {'present': False, 'rc': 'Success', 'expiry': None}
INFO queue limit: published 2000, delivered 1000 first=0 last=999

Son satır başarılı/başarısız sonucu değil, bilgidir. Kuyruğun gerçek limitini ve atma politikasını gösterir: first=0 last=999, broker'ın en eski mesajları tutup en yenileri attığı anlamına gelir; first=1000 last=1999 ise tam tersi demektir. expiry bir sayı gösteriyorsa broker istediğiniz aralığı geçersiz kılmıştır.

Betik her oturumu DISCONNECT içinde Session Expiry Interval = 0 ile sonlandırır; böylece test çalıştırmaları geride sahipsiz oturum bırakmaz. Aralığı bu yolla 0'a ancak CONNECT sırasında sıfırdan farklıysa indirebilirsiniz. DISCONNECT içinde 0'dan yukarı çıkarmak bir protokol hatasıdır.

Kendi broker'ınızda kuyruk limitlerini test etmek

Her broker çevrimdışı istemci başına ne kadar veri saklayacağını sınırlar, çünkü sınırsız kuyruklu bin ölü cihaz belleğini tüketirdi. Mosquitto'da ilgili ayarlar şunlardır:

persistence true
persistence_location /mosquitto/data/

# remove persistent sessions of clients offline for more than 2 days
persistent_client_expiration 2d

# per-client offline queue (0 bytes = no byte limit)
max_queued_messages 1000
max_queued_bytes 0

# QoS 0 messages are not queued for offline clients unless enabled
queue_qos0_messages false

Varsaydığınız limiti değil, yapılandırdığınız limiti test edin: max_queued_messages + 500 büyüklüğünde bir patlama yayınlayın, yeniden bağlanın ve karşılaştırın. Sonra cihazlarınızın neye ihtiyaç duyduğuna karar verin. Yalnızca son komutla ilgilenen bir sensör, devasa bir kuyruk yerine mesaj süresi dolmasını ya da retained durumu kullanmalıdır. İstemciler kuyruk dolu hataları görüyorsa MQTT mesaj kuyruğu dolu yazısı çözümleri anlatır. Mosquitto yapılandırma üreticisi bu ayarları sizin için oluşturabilir.

İstemci çevrimdışıyken broker'ı yeniden başlatmayı da test edin. Kalıcılık açık değilse Mosquitto oturumları yalnızca bellekte tutar ve bir yeniden başlatma hepsini kaybettirir.

Hata kalıpları ve anlamları

Daha kapsamlı bir hata ayıklama rehberi için MQTT oturumu neden kalıcı olmuyor yazısına bakın.

Tarayıcıda test etmek

Hızlı bir görsel kontrol için çevrimiçi MQTT istemcisini açın; sabit bir client ID, kapalı Clean Start ve bir oturum süresiyle bağlanın, QoS 1 ile abone olun ve bağlantıyı kesin. Terminalden birkaç mesaj yayınlayın, tarayıcıda yeniden bağlanın ve biriken mesajların gelişini izleyin.

Kalıcı bir oturumu tarayıcıda deneyin

Bir oturum testini çalıştırmak bir dakika sürer ve "cihazım komutu kaçırdı" şikâyetlerinde saatlerce hata ayıklamaktan kurtarır. Betiği deponuzda tutun, devreye aldığınız her broker'a karşı çalıştırın ve kuyruk limiti satırını kapasite planınızın bir parçası olarak görün.

Sıkça sorulan sorular