MQTT Yeniden Bağlanma Mantığını ve Backoff'u Test Etmek

Yeniden bağlanma kodu, bir MQTT istemcisinin geliştirme sırasında en az, production'da ise en çok çalışan kısmıdır. MQTT yeniden bağlanma mantığını test etmek için bağlantıyı gerçek dünyanın kopardığı şekillerde bilerek koparın — broker yeniden başlatması, reddedilen bağlantılar, sessiz bir ağ kopması — ve dört şeyi kontrol edin: istemci bunu fark ediyor mu, giderek artan aralıklarla bekliyor mu, yeniden abone oluyor mu ve uzakta olduğu sürede hiçbir QoS 1 mesajı kaybolmuyor mu?

Bu rehber size her şeyi kaydeden bir Python test istemcisi, dakikalar içinde yeniden üretebileceğiniz üç arıza senaryosu ve bir doğrulama kontrol listesi sunar.

"Doğru" yeniden bağlanma davranışı neye benzer

Tasarım tarafı — bekleme sürelerinin ve oturum ayarlarının nasıl seçileceği — MQTT yeniden bağlanma en iyi uygulamaları yazısında ele alınıyor. Burada istemcinizin bunu gerçekten yaptığını kanıtlamaya odaklanıyoruz.

Yaygın kütüphaneler nasıl yeniden bağlanır

• paho-mqtt (Python 2.x), loop_forever() ve loop_start() içinde otomatik olarak yeniden bağlanır. Üstel backoff'u reconnect_delay_set(min_delay, max_delay) kontrol eder; jitter yoktur.
• MQTT.js 5 her reconnectPeriod milisaniyede bir (varsayılan 1000) otomatik olarak yeniden bağlanır — üstel değil, sabit bir aralık. Varsayılan olarak önceki topic'lere yeniden abone olur (resubscribe: true).
• Eclipse Paho Java, setAutomaticReconnect(true) ayarlandığında yeniden bağlanır ve bekleme süresini yapılandırılabilir bir üst sınıra kadar ikiye katlar.
• MQTTnet yeniden bağlanmayı sizin kodunuza bırakır: bağlantı kopma olayını yakalar ve kendi bekleme sürenizle connect'i yeniden çağırırsınız.

Kütüphane hangisi olursa olsun, varsayılanlar nadiren production'da istediğiniz şeydir — test edilmeyi hak etmelerinin nedeni de tam olarak budur.

Her geçişi kaydeden bir test istemcisi

Bu paho-mqtt 2.x istemcisi kalıcı oturumlu MQTT 5 kullanır, her bağlantıda yeniden abone olur ve her olay için monotonik bir zaman damgası yazdırır; böylece backoff'u doğrudan çıktıdan okuyabilirsiniz:

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

HOST = os.getenv("MQTT_HOST", "localhost")
PORT = int(os.getenv("MQTT_PORT", "1883"))
TOPIC = "testmqtt/reconnect/probe/#"
T0 = time.monotonic()

def log(event, **info):
    print(f"{time.monotonic() - T0:8.2f}s  {event:<15} {info}", flush=True)

def on_connect(client, userdata, flags, reason_code, properties):
    log("connected", rc=str(reason_code), session_present=flags.session_present)
    if not reason_code.is_failure:
        client.subscribe(TOPIC, qos=1)   # her zaman yeniden abone ol

def on_disconnect(client, userdata, flags, reason_code, properties):
    log("disconnected", rc=str(reason_code))

def on_connect_fail(client, userdata):
    log("connect_failed")

def on_message(client, userdata, msg):
    log("message", topic=msg.topic, payload=msg.payload.decode(), dup=msg.dup)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id="reconnect-probe-1", protocol=mqtt.MQTTv5)
if os.getenv("MQTT_USER"):
    client.username_pw_set(os.getenv("MQTT_USER"), os.getenv("MQTT_PASS"))
client.on_connect, client.on_disconnect = on_connect, on_disconnect
client.on_connect_fail, client.on_message = on_connect_fail, on_message
client.reconnect_delay_set(min_delay=1, max_delay=32)

props = Properties(PacketTypes.CONNECT)
props.SessionExpiryInterval = 3600       # oturumu 1 saat tut
client.connect_async(HOST, PORT, keepalive=10, clean_start=False, properties=props)
client.loop_forever(retry_first_connection=True)

Test sırasında client ID'yi sabit tutun — kalıcı bir oturum bir client ID'ye aittir ve aynı ID'ye sahip iki süreç birbirini sürekli düşürür. Kısa bir keepalive=10, sessiz kopmaların hızla görünmesini sağlar; son çalıştırmada production değerinizi kullanın.

Senaryo 1: broker yeniden başlatma ve kesinti

İstediğiniz an durdurabileceğiniz, atılabilir bir Mosquitto'yu yerelde çalıştırın:

docker run -d --name mqtt-reconnect -p 1883:1883 eclipse-mosquitto:2 \
  mosquitto -c /mosquitto-no-auth.conf
python reconnect_probe.py            # terminal 1

docker stop mqtt-reconnect && sleep 60 && docker start mqtt-reconnect   # terminal 2

Broker geri gelene kadar ikiye katlanan bekleme süreleriyle beklenen çıktı:

    0.03s  connected       {'rc': 'Success', 'session_present': False}
   12.40s  disconnected    {'rc': 'Unspecified error'}
   13.41s  connect_failed  {}
   15.42s  connect_failed  {}
   19.43s  connect_failed  {}
   27.44s  connect_failed  {}
   43.45s  connect_failed  {}
   75.46s  connected       {'rc': 'Success', 'session_present': False}

Reason code metinleri sürüme göre değişebilir, ancak denemeler arasındaki aralıklar 1, 2, 4, 8, 16 ve 32 saniye olmalıdır. Yeniden başlatmadan sonraki session_present: False değerine dikkat edin: bu Mosquitto'da kalıcılık (persistence) yok, dolayısıyla oturum süreçle birlikte öldü. İstemciniz yeniden abone olmalıdır — bu testten sonra mesajlar gelmeyi bırakıyorsa hata budur. Mosquitto'da persistence true ayarını etkinleştirin (bkz. Mosquitto kalıcılığı) ve oturumun yeniden başlatmadan sağ çıktığını görmek için tekrarlayın.

Senaryo 2: sessiz ağ kopması

Broker yeniden başlatması soketleri temiz şekilde kapatır. Takılan bir Wi-Fi erişim noktası veya bir NAT timeout'u ise kapatmaz — paketler yalnızca kaybolur ve istemci bunu sadece keep-alive üzerinden öğrenir. Bir Linux makinesinde bunu bir güvenlik duvarı kuralıyla simüle edebilirsiniz:

sudo iptables -I OUTPUT -p tcp --dport 1883 -j DROP
sleep 45
sudo iptables -D OUTPUT -p tcp --dport 1883 -j DROP

Anında bir reset'i simüle etmek için -j DROP yerine -j REJECT --reject-with tcp-reset kullanın. Aradaki fark önemlidir: DROP ile sonda, kural eklendikten yaklaşık 10–20 saniye sonra (bir ila iki keep-alive aralığı) disconnected log'lamalı, ardından kural kaldırılana kadar backoff uygulamalıdır. Hiç bağlantı kopması log'lamıyorsa keep-alive çalışmıyordur — bkz. MQTT keep alive.

Gecikme, paket kaybı, bant genişliği sınırları veya platformlar arası kurulumlar için bir proxy iptables'tan daha esnektir; Toxiproxy ile test bu arızaları istemci ile broker arasına nasıl enjekte edeceğinizi gösterir.

Senaryo 3: kuyruktaki mesajlarla oturum devamı

Çoğu ekibin atladığı ve çoğu olayın kaynaklandığı test budur. Sondayı oturumları tutan bir broker'a yönlendirin — örneğin paylaşılan uç nokta broker.connectmq.com — ve:

1. reconnect_probe.py'yi başlatın ve abonelik yerindeyken connected satırını bekleyin.
2. Senaryo 2'deki iptables kuralıyla 1883 portunu engelleyin.
3. Sonda çevrimdışıyken, kuralın engellemediği bir yoldan testmqtt/reconnect/probe/q topic'ine beş QoS 1 mesajı publish edin — örneğin 8084 portunda güvenli WebSocket üzerinden bağlanan online MQTT istemcisi ile.
4. Kuralı kaldırın ve sondanın yeniden bağlanmasını izleyin.

Sonda session_present: True ve ardından beş mesajın tamamını log'lamalıdır. session_present: False ile yeniden bağlanıyorsa broker oturumu atmıştır: clean_start ayarını, oturum sona erme aralığını (session expiry interval) ve client ID'nizin değişip değişmediğini kontrol edin. Mesajlar dup: True ile veya iki kez geliyorsa bu, QoS 1'in at-least-once davranışının tasarlandığı gibi çalışmasıdır — handler'ınızın idempotent olması gerekir. Kalıcı oturumlar broker'ın neyi ne kadar süre tuttuğunu açıklar.

Test edilmeye değer diğer arıza durumları

• Çevrimdışıyken değişen kimlik bilgileri. Broker yeniden bağlanmaya "Not authorized" veya "Bad user name or password" ile yanıt verir. İstemci bunu belirgin şekilde log'lamalı ve yavaşlamalıdır; sonsuza kadar her saniye yeniden denememelidir. Kodları reason code aracında arayın.
• DNS hatası. Ağı tamamen kesin; çözümleme hataları da reddedilen bağlantılarla aynı backoff yolundan geçmelidir.
• Yinelenen client ID. Sondayı iki kez başlatın. Her instance diğerini düşürür ve ikisi de sonsuza kadar yeniden bağlanır — klasik bir production döngüsü.
• Uzun kesintiler. Broker'ı max_delay süresinden uzun süre kapalı bırakın ve denemelerin büyümeye devam etmek veya durmak yerine üst sınırda sabitlendiğini kontrol edin.
• Publish sırasında yeniden bağlanma. Ağı keserken döngü içinde QoS 1 mesajları publish edin ve toparlanmadan sonra bir subscriber'a ne ulaştığını sayın.

Yeniden bağlanma testi kontrol listesi

1. Her bağlantıyı, kopmayı ve başarısız denemeyi monotonik bir zaman damgasıyla log'layın.
2. Broker'ı yeniden başlatın: tespiti, artan bekleme sürelerini ve toparlanmayı doğrulayın.
3. Trafiği sessizce düşürün: keep-alive ile bir ila iki aralıkta tespiti doğrulayın.
4. Çevrimdışıyken QoS 1 mesajlarını kuyruğa alın: session_present değerini ve yeniden bağlandıktan sonraki teslimatı doğrulayın.
5. Kalıcılık olmadan yapılan bir yeniden başlatmadan sonra yeniden aboneliği doğrulayın.
6. Hatalı kimlik bilgilerini, DNS hatasını ve yinelenen client ID'leri test edin.
7. Senaryoları atılabilir bir broker container'ıyla CI'da otomatikleştirin.

Yeniden bağlanma mantığı, en son çalıştığını gördüğünüz an kadar iyidir. Zaman damgası yazdıran bir sonda, durdurabileceğiniz bir broker ve tek bir güvenlik duvarı kuralı; backoff'u, yeniden aboneliği ve oturum devamını kanıtlamaya — ve gerilemeyi bir broker yeniden başlatması sizin yerinize yakalamadan önce yakalamaya — yeter.

Sıkça sorulan sorular