Paho MQTT Yeniden Bağlanma: Python'da Sağlam Backoff, Yeniden Subscribe ve Çevrimdışı Tamponlama

Paho MQTT yeniden bağlanma davranışı, kararsız bir ağda ayakta kalan bir istemci ile gece saat 3'te sessizce çalışmayı bırakan bir istemci arasındaki farktır. Bu rehber paho-mqtt 2.x'in otomatik olarak ne yaptığını ve açık bıraktığı üç boşluğu açıklıyor, ayrıca size üstel backoff, bağlantıda yeniden subscribe, bir çevrimdışı tampon ve kalıcı MQTT 5 oturumları içeren test edilmiş bir Python istemcisi sunuyor.

Bozuk Yeniden Bağlanma Mantığının Belirtileri

Yeniden bağlanma hataları bir programı nadiren çökertir. Kolayca yanlış teşhis edilen daha sessiz arızalar olarak ortaya çıkarlar:

• İstemci bir Wi-Fi kopmasından sonra yeniden bağlanır, "connected" loglar, ancak bir daha hiç mesaj almaz.
• Ağ kapalıyken başlatılan bir betik beklemek yerine bir soket hatasıyla hemen çıkar.
• Kısa bir kesinti sırasında alınan ölçümler kaybolur ve grafiklerde boşluklar bırakır.
• Yüzlerce cihaz bir broker yeniden başlatmasından sonra tam olarak aynı anda yeniden bağlanır ve ağı aşırı yükler.
• İki işlem birkaç saniyede bir birbirinin bağlantısını keser.

Bunların her birinin belirli bir nedeni ve belirli bir çözümü vardır.

Paho'nun Otomatik Olarak Yaptıkları

Network loop'u loop_start() veya loop_forever() ile çalıştırdığınızda paho-mqtt, kurulmuş bir bağlantı koptuktan sonra kendi başına yeniden bağlanır. Denemeler arasındaki gecikme min_delay değerinden başlar ve reconnect_delay_set() ile yapılandırılan max_delay değerine ulaşana kadar her başarısızlıktan sonra iki katına çıkar. Varsayılanlar 1 ve 120 saniyedir. Başarılı bir bağlantı gecikmeyi sıfırlar.

Paho ayrıca broker'ın henüz onaylamadığı uçuştaki (in-flight) QoS 1 ve QoS 2 mesajlarını tutar ve oturum sürdürüldüğü sürece yeniden bağlandıktan sonra bunları yeniden iletir.

Bu pek çok şeyi kapsar, ama her şeyi değil. Kalan boşluklar sizin sorumluluğunuzdadır.

Boşluk 1: İlk Bağlantı Başarısız Olur

client.connect() ilk bağlantıyı senkron olarak gerçekleştirir ve broker'a ulaşılamazsa genellikle ConnectionRefusedError, socket.gaierror veya TimeoutError gibi bir istisna fırlatır. Hiç bağlantı olmadığı için otomatik yeniden bağlanma hiç başlamaz. Ağ hazır olmadan, açılışta başlatılan betiklerin sıklıkla ölmesinin nedeni budur.

Çözüm, connect_async() ve ardından loop_start() kullanmaktır. Bağlantı denemesi, aynı backoff kurallarıyla yeniden deneyen network loop'a taşınır.

Boşluk 2: Abonelikler Kaybolur

Clean session ile broker, bağlantı sona erdiğinde aboneliklerinizi atar. Paho bunları kaydetmez ve sizin yerinize yeniden subscribe olmaz. connect() sonrasında subscribe() fonksiyonunu bir kez çağırırsanız, istemci mutlu bir şekilde yeniden bağlanır ve sonra hiçbir şey almaz.

Çözüm basittir ve pazarlığa kapalıdır: on_connect içinde subscribe olun. Bu callback, yeniden bağlanmalar dahil her başarılı bağlantıdan sonra çalışır; böylece abonelikler her zaman geri yüklenir.

Boşluk 3: Çevrimdışıyken Publish Edilen Mesajlar

Bağlantı kopukken üretilen ölçümlerin gidecek bir yere ihtiyacı vardır. QoS 0 mesajları "gönder ve unut" mantığıyla çalışır, bu yüzden bağlantı yokken publish etmek onları düşürür. Daha yüksek QoS seviyeleri için paho'nun dahili kuyruğuna güvenmek de iyi bir plan değildir, çünkü sınırsızdır ve uygulamanız tarafından görülemez. Yeniden bağlandıktan sonra boşalttığınız açık ve sınırlı bir çevrimdışı tampon, size bellek üzerinde ve bir kesinti çok uzun sürdüğünde neyin atılacağı üzerinde kontrol sağlar.

Sağlam Bir Paho MQTT Yeniden Bağlanma İstemcisi

Aşağıdaki sınıf üç boşluğun tamamını kapatır ve bir cihaz filosunun hep birlikte yeniden bağlanmaması için backoff'a jitter ekler. Broker'ın kısa kesintiler sırasında abonelikleri ve kuyruktaki QoS 1 mesajlarını tutması için session expiry ile MQTT 5 kullanır.

pip install "paho-mqtt>=2.0"

import json
import logging
import random
import threading
import time
import uuid
from collections import deque

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

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("mqtt")


class ResilientClient:
    def __init__(self, host, port, client_id, subscriptions,
                 username=None, password=None, use_tls=False,
                 session_expiry_s=1800, buffer_size=1000):
        self.host = host
        self.port = port
        self.subscriptions = subscriptions  # (topic, qos) listesi
        self.session_expiry_s = session_expiry_s
        self.offline = deque(maxlen=buffer_size)  # dolduğunda en eskiler atılır
        self.lock = threading.Lock()
        self.on_payload = None

        c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                        client_id=client_id, protocol=mqtt.MQTTv5)
        if username:
            c.username_pw_set(username, password)
        if use_tls:
            c.tls_set()
        # Üstel backoff; rastgele bir alt sınır yeniden bağlanan filoları dağıtır
        c.reconnect_delay_set(min_delay=random.randint(1, 5), max_delay=60)
        c.on_connect = self._on_connect
        c.on_disconnect = self._on_disconnect
        c.on_connect_fail = self._on_connect_fail
        c.on_message = self._on_message
        self.client = c

    def start(self):
        props = Properties(PacketTypes.CONNECT)
        props.SessionExpiryInterval = self.session_expiry_s
        self.client.connect_async(self.host, self.port, keepalive=30,
                                  clean_start=False, properties=props)
        self.client.loop_start()

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

    # --- callback'ler ----------------------------------------------------
    def _on_connect(self, client, userdata, flags, reason_code, properties):
        if reason_code.is_failure:
            log.warning("connect refused: %s", reason_code)
            return
        log.info("connected, session_present=%s", flags.session_present)
        if self.subscriptions:
            client.subscribe(self.subscriptions)  # her zaman yeniden subscribe ol
        self._flush()

    def _on_disconnect(self, client, userdata, flags, reason_code, properties):
        if reason_code == 0:
            log.info("clean disconnect")
        else:
            log.warning("connection lost: %s, reconnecting", reason_code)

    def _on_connect_fail(self, client, userdata):
        log.warning("connection attempt failed, retrying with backoff")

    def _on_message(self, client, userdata, msg):
        if self.on_payload:
            self.on_payload(msg.topic, msg.payload)

    # --- publish ---------------------------------------------------------
    def publish(self, topic, payload, qos=1, retain=False):
        if self.client.is_connected():
            info = self.client.publish(topic, payload, qos=qos, retain=retain)
            if info.rc == mqtt.MQTT_ERR_SUCCESS:
                return True
        with self.lock:
            self.offline.append((topic, payload, qos, retain))
        return False

    def _flush(self):
        with self.lock:
            pending = list(self.offline)
            self.offline.clear()
        if pending:
            log.info("flushing %d buffered messages", len(pending))
        for topic, payload, qos, retain in pending:
            self.client.publish(topic, payload, qos=qos, retain=retain)


if __name__ == "__main__":
    run_id = uuid.uuid4().hex[:8]
    prefix = f"connectmq/demo/{run_id}"
    rc = ResilientClient(
        host="broker.connectmq.com", port=1883,
        client_id=f"resilient-{run_id}",
        subscriptions=[(f"{prefix}/cmd/#", 1)],
    )
    rc.on_payload = lambda t, p: log.info("command on %s: %s", t, p)
    rc.start()
    print(f"publishing to {prefix}/telemetry, send commands to {prefix}/cmd/...")
    try:
        n = 0
        while True:
            n += 1
            payload = json.dumps({"seq": n, "ts": int(time.time() * 1000)})
            sent = rc.publish(f"{prefix}/telemetry", payload, qos=1)
            log.info("seq %d %s (buffered: %d)", n, "sent" if sent else "buffered", len(rc.offline))
            time.sleep(2)
    except KeyboardInterrupt:
        rc.stop()

Her parça neden orada

Her payload'daki sıra numarası testi kolaylaştırır: bir kesintiden sonra seq içindeki boşluklar veya tekrarlar tam olarak ne olduğunu gösterir. QoS 1 en az bir kez teslimat olduğu için tüketiciler, örneğin daha önce gördükleri sıra numaralarını yok sayarak ara sıra oluşan bir tekrara tolerans göstermelidir. MQTT QoS seviyeleri açıklamamız tekrarların neden mümkün olduğunu ele alır.

Yeniden Bağlanma Mantığı Nasıl Test Edilir

Bilerek bozmadığınız yeniden bağlanma koduna güvenmeyin. Betiği ve bir subscriber'ı yan yana çalıştırın:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-<run-id> -t "connectmq/demo/<run-id>/#" -v

Ardından her arıza modunu sırayla deneyin:

1. Ağ kopması. Wi-Fi'ı devre dışı bırakın veya kabloyu 30 saniyeliğine çıkarın. Log önce "connection lost", ardından giderek artan gecikmelerle başarısız denemeler, sonra da "connected" ve "flushing N buffered messages" gösterir. Subscriber, tamponlanan sıra numaralarının bir patlama halinde geldiğini görür.
2. Linux'ta engellenmiş port. Ağın geri kalanı çalışırken broker'a ulaşılamayan bir durumu simüle edin:

sudo iptables -A OUTPUT -p tcp --dport 1883 -j DROP
# 60 saniye bekleyin, ardından kuralı kaldırın
sudo iptables -D OUTPUT -p tcp --dport 1883 -j DROP

Düşürülen paketler reddedilen bir bağlantıdan daha sert bir testtir, çünkü istemci durumu yalnızca keep-alive aralığının yaklaşık bir buçuk katı olan keep-alive zaman aşımı yoluyla fark eder. 3. Çevrimdışı başlatma. Betiği başlatmadan önce ağın bağlantısını kesin. connect_async() ile betik çıkmak yerine bekler ve ağ geri geldiğinde bağlanır. 4. Oturumun ele geçirilmesi. Aynı client ID ile ikinci bir istemci başlatın, örneğin mosquitto_sub -i resilient-<run-id> -t x. Broker orijinal istemcinin bağlantısını keser; o da yeniden bağlanır ve sırası gelince davetsiz misafirin bağlantısını keser. Bu "birbiriyle çekişen iki işlem" belirtisidir; bunu bilerek görmek, daha sonra tanımayı kolaylaştırır. 5. Kesinti sırasında komutlar. Betik çevrimdışıyken onun cmd topic'ine bir QoS 1 komutu publish edin. Oturum korunduğu için komut, istemci geri döndüğünde teslim edilir.

İşlenmeye Değer Reason Code'lar

MQTT 5 size bir bağlantının neden sona erdiğini söyler. Reason code'u loglayın ve yeniden denemenin düzeltmeyeceği durumlara tepki verin.

Bir 0x86 reddini sonsuza kadar yeniden denemek yalnızca broker'ın loglarını doldurur ve ConnectMQ dahil pek çok broker, tekrarlanan başarısız girişlerden sonra bir IP adresini engeller. _on_connect içinde, kod kalıcı bir soruna işaret ettiğinde client.disconnect() çağırmayı ve bir insanı uyarmayı düşünün.

Üretim Ayarları

Gerçek dağıtımlarda, broker'ınızın verdiği kimlik bilgileriyle 8883 portunda TLS üzerinden bağlanın. ConnectMQ'nun barındırılan MQTT broker hizmetine karşı istemciyi şöyle oluşturun:

rc = ResilientClient(
    host="broker.connectmq.com", port=8883, use_tls=True,
    client_id="YOUR_CLIENT_ID", username="YOUR_USERNAME", password="YOUR_PASSWORD",
    subscriptions=[("connectmq/factory1/production/line1/+/cmd", 1)],
)

Birkaç sınırı aklınızda tutun. Herkese açık broker'da oturumlar en fazla 30 dakika sonra sona erer; bu yüzden yukarıda kullanılan 1.800 saniyelik expiry orada işe yarayan en yüksek değerdir. Ücretsiz Serverless planı, bu istemcinin kullandığı QoS 1'e kadar destekler. Tam olarak bir kez teslimata ihtiyacınız varsa ücretli planlar QoS 2 ekler. Broker ayrıca bağlantı durumunu $SYS/broker/connection/{clientId}/state üzerinde publish eder; bu, bir cihazın kesintiden sonra geri geldiğini doğrulamak için kullanışlıdır.

Daha geniş bir arka plan için paho-mqtt Python eğitimimiz temelleri ele alır, paho-mqtt 2.0 callback API rehberi burada kullanılan sürüm 2 callback imzalarını açıklar ve paho ile asenkron kütüphaneler arasında seçim yapıyorsanız MQTT istemci kütüphaneleri karşılaştırması yardımcı olur. Resmi paho-mqtt dokümantasyonu yeniden bağlanmayla ilgili her seçeneği listeler.

ConnectMQ ile Başlayın

Yeniden bağlanma mantığınızı bugün ücretsiz herkese açık MQTT broker üzerinde test edin, ardından dağıtım yaptığınızda özel kimlik bilgilerine geçin. ConnectMQ; TLS, session expiry ve anlaşılır reason code'lar içeren tam yönetilen bir MQTT 5 broker'ı çalıştırır; böylece bir şeyler ters gittiğinde istemcileriniz faydalı geri bildirim alır. Kredi kartı olmadan app.connectmq.com adresinde ücretsiz bir Serverless broker oluşturun.

Sık Sorulan Sorular

paho-mqtt otomatik olarak yeniden bağlanır mı?

Evet, bir bağlantı kurulduktan ve network loop loop_start() veya loop_forever() ile çalıştıktan sonra. Başarısız bir ilk connect() çağrısını yeniden denemez ve sizin yerinize yeniden subscribe olmaz.

paho istemcim yeniden bağlandıktan sonra neden mesaj almayı bırakıyor?

Abonelikler eski oturumla birlikte kayboldu. subscribe() çağrılarınızı her başarılı bağlantıdan sonra çalışmaları için on_connect içine taşıyın.

Hangi yeniden bağlanma gecikmesini kullanmalıyım?

1 ile 5 saniye arasında başlayın ve üst sınırı 60 ile 120 saniye arasında tutun. Başlangıç gecikmesini cihaz başına rastgele seçmek, tüm filonun aynı anda yeniden bağlanmasını önler.

clean_start=False kullanmalı mıyım?

Broker'ın kısa kesintiler sırasında abonelikleri tutmasını ve QoS 1 mesajlarını kuyruğa almasını istediğinizde bir session expiry interval ile birlikte kullanın. Oturum client ID'ye bağlı olduğu için client ID'yi sabit tutun.