MQTT Çevrimdışı Mesajlar: Broker'lar Bağlantısı Kesik İstemciler için Mesajları Nasıl Kuyruğa Alır?

MQTT çevrimdışı mesajlar, broker'ın geçici olarak bağlantısı kesilen bir subscriber için tuttuğu ve subscriber geri döndüğünde teslim ettiği yayınlardır. Bu yazı tam olarak hangi mesajların bu kapsama girdiğini, gerçek broker'larda hangi sınırların geçerli olduğunu açıklıyor ve kuyruğa alma ile teslimatı kendi gözünüzle görebilmeniz için tekrarlanabilir bir testi adım adım gösteriyor.

MQTT Çevrimdışı Mesajları için Dört Koşul

Kuyruğa alma otomatik değildir. Broker, çevrimdışı bir istemci için bir mesajı yalnızca aşağıdakilerin tümü doğru olduğunda kuyruğa alır:

Herhangi bir koşul sağlanmazsa mesaj o istemciye basitçe teslim edilmez. MQTT tasarım gereği publisher'ları subscriber'lardan ayırdığı için publisher herhangi bir hata almaz. "Cihazım bir komutu kaçırdı" hatalarının bu kadar yaygın olmasının nedeni budur: publisher broker'dan başarılı bir PUBACK görür, ancak broker'ın mesajı o subscriber için saklaması için bir nedeni yoktu.

Peki QoS 0?

QoS 0 "en fazla bir kez" demektir. Spesifikasyon, broker'ın çevrimdışı oturumlar için QoS 0 mesajlarını kuyruğa almasına izin verir ama bunu zorunlu tutmaz ve çoğu broker bunu yapmaz. QoS 0'ı yalnızca canlı veri olarak düşünün. Seviyeler arasındaki farklar MQTT QoS seviyeleri açıklandı yazısında ayrıntılı olarak ele alınıyor.

Peki Retained Mesajlar?

Retained mesajlar ayrı bir mekanizmadır. Broker her topic için son retained mesajı saklar ve oturumu olsun ya da olmasın abone olan herkese gönderir. Bunlar "ne kaçırdım?" sorusunu değil, "güncel değer nedir?" sorusunu çözer. Temiz oturumla yeniden bağlanan bir istemci, abone olduğunda en son retained değeri yine alır, ancak aradaki güncellemelerin hiçbirini almaz. İyi bir tasarım çoğu zaman ikisini birlikte kullanır: durum için retained mesajlar, olaylar ve komutlar için kuyruğa alınan QoS 1 mesajları.

Çevrimdışı Kuyruklara Uygulanan Sınırlar

Her broker kuyruklara sınır koyar; aksi halde hiç geri dönmeyen bağlantısı kesik bir istemci sonsuza kadar bellek tüketirdi.

• Oturum ömrü. MQTT 5'te session expiry interval, kuyruğun ne kadar süre yaşayacağına karar verir. Süre dolduğunda kuyruktaki mesajlar atılır. broker.connectmq.com adresindeki herkese açık MQTT test broker'ında session expiry 30 dakikayla sınırlıdır. Değer seçimi MQTT session expiry interval yazısında ele alınıyor.
• Kuyruk derinliği. Broker'lar oturum başına kuyruğa alınan mesaj sayısını veya bayt miktarını sınırlar. Sınıra ulaşıldığında broker'a bağlı olarak yeni mesajlar düşürülür veya eskileri çıkarılır.
• Mesaj süresi. Bir MQTT 5 publisher'ı Message Expiry Interval ayarlayabilir. Bir mesaj kuyrukta bundan daha uzun süre beklerse broker eskimiş veriyi teslim etmek yerine onu atar.
• Mesaj boyutu. Plan ve broker mesaj boyutu sınırları kuyruktaki mesajlar için de geçerlidir (herkese açık broker'da 64 KB, ConnectMQ Serverless Free planında 256 KB, ücretli Serverless planında 1 MB).
• Broker yeniden başlatmaları. Kuyrukların nerede tutulduğu broker'a bağlıdır. ConnectMQ oturumları ve kuyruklarını bellekte tutar: tanınan oturum ömrü boyunca istemci bağlantı kesilmelerinden sağ çıkarlar, ancak broker yeniden başlatıldığında korunmazlar. İstemciler kaybolan bir oturumu tespit etmeli (CONNACK içinde Session Present = 0), yeniden abone olmalı ve durum için retained mesajlara veya uygulama düzeyinde yeniden senkronizasyona güvenmelidir.

Tekrarlanabilir Test: Çevrimdışı Teslimatı İzleyin

Bu test mosquitto komut satırı araçlarıyla yaklaşık beş dakika sürer. q5t9 değerini kendi rastgele dizenizle değiştirin ve payload'ları zararsız tutun: herkese açık alan herkes tarafından okunabilir.

1. Oturumu oluşturun. Sabit bir kimlik, Clean Start false, 15 dakikalık session expiry ve QoS 1 ile bir subscriber bağlayın:

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 \
  -i offline-sub-q5t9 -c -x 900 -q 1 -v \
  -t 'connectmq/demo/q5t9/cmd/#'

2. Çevrimdışı olun. Ctrl+C tuşlarına basın.

3. Subscriber yokken farklı QoS seviyelerinde mesajlar publish edin:

H="-h broker.connectmq.com -p 1883 -V mqttv5 -i offline-pub-q5t9"
mosquitto_pub $H -q 0 -t 'connectmq/demo/q5t9/cmd/a' -m 'qos0 message'
mosquitto_pub $H -q 1 -t 'connectmq/demo/q5t9/cmd/b' -m 'qos1 message'
mosquitto_pub $H -q 1 -t 'connectmq/demo/q5t9/cmd/c' -m 'short-lived' \
  -D publish message-expiry-interval 5
mosquitto_pub $H -q 1 -t 'connectmq/demo/q5t9/other' -m 'not subscribed'

4. En az 10 saniye bekleyin, ardından 1. adımdaki komutu yeniden çalıştırın.

Yalnızca connectmq/demo/q5t9/cmd/b qos1 message satırını görmelisiniz. Diğerlerinin her birinin neden eksik olduğu şöyle açıklanır:

• cmd/a QoS 0'dı, bu yüzden kuyruğa alınmadı.
• cmd/c 5 saniyelik message expiry'ye sahipti ve kuyrukta atıldı.
• other abonelik filtresiyle eşleşmedi.

5. 1. adımda -c olmadan tekrarlayın. Oturum bağlantı kapandığında atıldığı için yeniden bağlandığınızda hiçbir şey teslim edilmez.

Aynı Kalıp Python ile

Yeniden bağlanan ve kaçırdığı her şeyi işleyen bir cihaz:

import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

TOPIC = "connectmq/demo/q5t9/cmd/#"

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print("Connect failed:", reason_code)
        return
    print("Connected. Session present:", flags.session_present)
    if not flags.session_present:
        # Yeni veya kaybolmuş oturum: abone ol ve durumu başka bir yoldan al
        client.subscribe(TOPIC, qos=1)

def on_message(client, userdata, msg):
    print(f"Processing {msg.topic}: {msg.payload.decode()}")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id="offline-py-q5t9",
                     protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_message = on_message
client.reconnect_delay_set(min_delay=1, max_delay=60)

props = Properties(PacketTypes.CONNECT)
props.SessionExpiryInterval = 900

client.connect("broker.connectmq.com", 1883, keepalive=60,
               clean_start=False, properties=props)
client.loop_forever()

Çevrimdışı İstemcileri Hesaba Katan Tasarım

Kuyruğa alma kısa kesintilerde yardımcı olur, ancak tek güvenilirlik aracınız olmamalıdır. Sağlam bir tasarım birkaç mekanizmayı bir araya getirir:

1. Komutlar ve olaylar için QoS 1, bir sonraki ölçümün öncekinin yerini aldığı yüksek frekanslı telemetri için QoS 0 kullanın.
2. Session expiry'yi gerçekçi tutun. Tipik kesintileri kapsayacak kadar uzun, ölü cihazların kuyruk biriktirmeyeceği kadar kısa olsun.
3. Bir saat sonra geri dönen bir cihazın güncelliğini yitirmiş bir talimatı yürütmemesi için zamana duyarlı komutlarda message expiry ayarlayın.
4. Herhangi bir istemcinin bir kuyruğa bağlı kalmadan anında yeniden senkronize olabilmesi için durumu retained mesajlar olarak publish edin.
5. İşleyicileri idempotent yapın. QoS 1, yeniden bağlanmalardan sonra yinelenen mesajlar teslim edebilir; payload'a bir mesaj kimliği veya sıra numarası ekleyin.
6. Session Present = 0 durumunu bir hata olarak değil, normal bir olay olarak ele alın.
7. Yeniden bağlanmada ani bir mesaj yükü bekleyin. Çevrimdışı kalan istemciler bir anda çok sayıda mesaj alabilir; MQTT 5 Receive Maximum, paralel olarak gelen onaylanmamış mesaj sayısını sınırlar.

Oturumların kendisine dair arka plan bilgisi için MQTT kalıcı oturumlar ve clean session bayrağı yazısını okuyun. Yeni başlıyorsanız MQTT eğitimi önce publish/subscribe temellerini ele alıyor.

Eksik Çevrimdışı Mesajlarda Sorun Giderme

• Yeniden bağlandıktan sonra hiçbir şey gelmiyor: client ID'nin aynı olduğunu, Clean Start'ın false olduğunu ve MQTT 5'te session expiry'nin sıfır olmadığını doğrulayın.
• Bazı mesajlar geliyor, bazıları gelmiyor: her publish'in ve aboneliğin QoS değerini karşılaştırın ve message expiry'yi kontrol edin.
• Mesajlar iki kez geliyor: QoS 1'de yarıda kesilen bir onaydan sonra bu beklenen bir durumdur; uygulamada yinelenenleri ayıklayın.
• Session Present beklenmedik şekilde 0: oturumun süresi dolmuş, client ID değişmiş veya broker yeniden başlamıştır.

Çevrimdışı Mesajları ConnectMQ ile Kuyruğa Alın

ConnectMQ, MQTT 5.0 ve MQTT 3.1.1 istemcileri için kalıcı oturumları, session expiry'yi ve message expiry'yi destekleyen barındırılan bir MQTT broker'dır. Serverless Free planı, çevrimdışı kuyruklamanın gerektirdiği tek şey olan QoS 1'e kadar destek verir; QoS 2 ise ücretli planlarda mevcuttur. Kredi kartı gerektirmeden ücretsiz broker'ınızı oluşturun, YOUR_USERNAME, YOUR_PASSWORD ve YOUR_CLIENT_ID ile bağlanın ve yukarıdaki testi kendi özel uç noktanızda çalıştırın. Sağlayıcıları karşılaştırıyorsanız bir bulut MQTT broker seçerken nelere dikkat etmeniz gerektiğine göz atın.

Sık Sorulan Sorular

Subscriber çevrimdışıyken MQTT mesajları saklar mı?

Yalnızca subscriber'ın kalıcı bir oturumu varsa, eşleşen bir topic'e zaten abone olduysa ve etkin QoS 1 veya 2 ise. Aksi halde mesaj o istemci için tutulmaz.

Cihaz çevrimdışıyken QoS 0 mesajlarım neden kayboluyor?

QoS 0 mesajları normalde çevrimdışı oturumlar için kuyruğa alınmaz; spesifikasyon bunu isteğe bağlı bırakır. Mesajın bir bağlantı kesilmesinden sağ çıkması gerekiyorsa QoS 1 ile publish edin ve abone olun.

Bir broker çevrimdışı mesajları ne kadar süre saklar?

Oturumun süresi dolana, bir kuyruk sınırına ulaşılana, mesajın kendi süresi geçene veya oturumlar bellekte tutuluyorsa broker yeniden başlayana kadar. ConnectMQ'da oturumlar bellekte yaşar ve herkese açık test broker'ı session expiry'yi 30 dakikayla sınırlar.

Retained mesajlar çevrimdışı kuyruklamanın yerini tutar mı?

Hayır. Retained mesaj, yeni bir subscriber'a bir topic'teki en son değeri verirken çevrimdışı kuyruklama, istemcinin kaçırdığı her QoS 1 veya 2 mesajını teslim eder. Çoğu sistem ikisini birlikte kullanır.