MQTT Broker Kesintisinden Etkilenmeyen IoT İstemcileri Tasarlamak

MQTT broker kesintisi "olur mu" değil "ne zaman olur" sorusudur: ağlar bölünür, sertifikalar yenilenir, broker'lar güncellenir ve bulut bölgelerinin de kötü günleri olur. Bu rehber, cihaz ve backend istemcilerini kısa ya da uzun bir broker kesintisinde önemli hiçbir veriyi kaybetmeyecek ve mükerrer işlem yapmayacak şekilde nasıl tasarlayacağınızı gösteriyor. Kalıplar ConnectMQ dahil her broker için geçerlidir ve çalışan Python koduyla birlikte gelir.

MQTT Broker Kesintisi Sırasında Gerçekte Ne Olur?

İstemcinin bakış açısından bir kesinti üç şekilde görünür: TCP bağlantısı kopar, bağlantı keepalive süresi dolana kadar sessizce asılı kalır ya da yeni bağlantı denemeleri reddedilir. Sebep broker'ın kendisi, sizinle broker arasındaki ağ, DNS, TLS, hatta kendi kimlik bilgileriniz olabilir. İstemci hangisi olduğunu her zaman ayırt edemez; bu yüzden hepsini aynı şekilde ele almalıdır: durumu hızla fark etmek, yerelde çalışmaya devam etmek ve kibarca yeniden bağlanmak.

Dört şey risk altındadır:

1. Bağlantı yokken yayınlanan giden veri, tamponlamazsanız kaybolur.
2. Siz çevrimdışıyken gönderilen gelen komutlar, oturum ayarlarına bağlı olarak teslim edilebilir de edilmeyebilir de.
3. Diğer sistemler cihazı hâlâ çevrimiçi sanıyorsa varlık (presence) bilgisi yanlış hale gelir.
4. Binlerce cihaz aynı anda yeniden bağlandığında broker'ın kendisi aşırı yüklenebilir.

İyi bir istemci tasarımı bunların her birini ele alır. Oturumlar ve kuyruğa alınmış mesajlar gibi terimler size yabancıysa önce bir MQTT broker'ın ne yaptığını okumanızda fayda var.

Kalıp 1: Exponential Backoff ve Jitter ile Yeniden Bağlanma

Saf yaklaşım, sonsuza kadar her saniye yeniden denemektir. Bunu on bin cihazla çarpın: broker geri geldiği anda bir eşzamanlı bağlantı duvarıyla karşılaşır ve bu onu hemen yeniden devirebilir. Buna thundering herd (sürü saldırısı) problemi denir.

Çözüm, rastgele jitter eklenmiş exponential backoff'tur: 1 sn, ardından 2 sn, 4 sn, 8 sn şeklinde bir üst sınıra kadar bekleyin ve cihazların zamana yayılması için her gecikmeyi rastgeleleştirin. Paho 2.x, reconnect_delay_set ile yerleşik destek sunar ancak jitter eklemez; bu nedenle pek çok ekip tam kontrol için döngüyü kendisi yönetir.

Kimlik bilgilerinize de dikkat edin. ConnectMQ gibi tekrarlanan başarısız girişlerden sonra IP adreslerini engelleyen broker'lar, yanlış parolayla her saniye yeniden deneyen bir cihazı saldırı olarak değerlendirir. Kimlik doğrulama hatalarında ağ hatalarına göre çok daha agresif şekilde geri çekilin ve bu hataları bir operatöre bildirin.

Kalıp 2: Giden Veriyi Yerelde Tamponlama

MQTT istemci kütüphaneleri iletimdeki (in-flight) mesajları bellekte tutar; cihaz kesinti sırasında yeniden başlarsa bunlar yok olur. Kaybetmeyi göze alamayacağınız telemetri için her okumayı önce küçük bir yerel depoya yazın ve yalnızca broker PUBACK ile onayladıktan sonra silin. Gateway'ler ve Linux cihazlarda SQLite iyi bir seçimdir; mikrodenetleyicilerde flash bellekteki bir ring buffer aynı işi görür.

Tampon dolduğunda ne olacağına baştan karar verin. Çoğu telemetri için en eski okumaları atmak doğrudur. Faturalama veya güvenlik olaylarında ise kritik olmayan verilerin örneklenmesini durdurmayı tercih edebilirsiniz.

Kalıp 3: Last Will ve Retained Durum ile Varlık Bildirme

Bağlanırken bir Last Will and Testament (LWT) ayarlayın: temiz bir disconnect olmadan ortadan kaybolursanız broker'ın sizin adınıza yayınlayacağı retained bir offline mesajı. Her başarılı bağlantıda aynı topic'e retained bir online yayınlayın. Böylece diğer sistemler polling yapmadan doğru varlık bilgisini görür.

LWT semantiği konusunda net olun. Will mesajı, broker oturumun anormal şekilde sona erdiğini tespit ettiğinde yayınlanır; bu genellikle keepalive aralığının 1,5 katı kadar sürer. Broker'ın kendisi çökmüşse o geri gelene kadar hiçbir şey yayınlanmaz ve broker yeniden başladıktan sonra retained durumun istemciler tarafından yeniden oluşturulması gerekebilir. MQTT 5, kısa bir yeniden bağlanma sırasında gidip gelen offline mesajlarını bastırmanıza olanak tanıyan bir will delay interval ekler.

Kalıp 4: Komutları Idempotent Yapma

QoS 1 ile bir mesaj en az bir kez teslim edilir. Kesinti ve yeniden bağlanma sırasında mükerrer mesajlar olası hale gelir. İki kez teslim edilen "röleyi değiştir" komutu röleyi yeniden kapatır; iki kez teslim edilen "10 ml dağıt" komutu 20 ml dağıtır. Çare, her komutu idempotent yapmaktır:

• Göreli eylemler (toggle) yerine mutlak durumlar ({"relay": "on"}) gönderin.
• Her komuta benzersiz bir command_id ekleyin ve cihazın yakın zamanda çalıştırdığı ID'leri hatırlamasını sağlayın.
• Bir son geçerlilik süresi ekleyin: MQTT 5 message expiry interval ya da bir valid_until zaman damgası; böylece uzun bir kesinti sırasında kuyruğa alınan bir komut, artık güvenli olmadığı saatler sonra çalıştırılmaz.

Dayanıklı Bir Python İstemcisi

Aşağıdaki istemci dört kalıbı bir araya getirir. Telemetriyi SQLite'ta tamponlar, rastgeleleştirilmiş exponential backoff ile yeniden bağlanır, LWT ile retained bir durum topic'ini sürdürür ve komutlardaki mükerrerleri ayıklar. Hemen çalıştırabilmeniz için herkese açık MQTT test broker'ını kullanır; canlı ortamda port 8883'e, TLS'e ve portal kimlik bilgilerinize geçin.

import json
import random
import sqlite3
import time
import uuid

import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
DEVICE = f"dev-{uuid.uuid4().hex[:6]}"
ROOT = f"connectmq/demo/{DEVICE}"
STATUS, TELEMETRY, COMMANDS = f"{ROOT}/status", f"{ROOT}/telemetry", f"{ROOT}/cmd"

db = sqlite3.connect("outbox.db", check_same_thread=False)
db.execute("CREATE TABLE IF NOT EXISTS outbox (id INTEGER PRIMARY KEY, payload TEXT)")
db.commit()

pending = {}            # mid -> outbox satır id'si
seen_commands = []      # mükerrer ayıklama için son komut ID'leri
connected = False


def on_connect(client, userdata, flags, reason_code, properties):
    global connected
    if reason_code.is_failure:
        print("Connect refused:", reason_code)
        return
    connected = True
    print("Connected")
    client.publish(STATUS, "online", qos=1, retain=True)
    client.subscribe(COMMANDS, qos=1)


def on_disconnect(client, userdata, flags, reason_code, properties):
    global connected
    connected = False
    pending.clear()  # onaylanmamış satırlar outbox'ta kalır ve yeniden gönderilir
    print("Disconnected:", reason_code)


def on_publish(client, userdata, mid, reason_code, properties):
    row_id = pending.pop(mid, None)
    if row_id is not None:
        db.execute("DELETE FROM outbox WHERE id = ?", (row_id,))
        db.commit()


def on_message(client, userdata, msg):
    try:
        cmd = json.loads(msg.payload)
    except ValueError:
        print("Ignoring malformed command")
        return
    if cmd.get("command_id") in seen_commands:
        print("Duplicate command ignored:", cmd["command_id"])
        return
    if cmd.get("valid_until", float("inf")) < time.time():
        print("Expired command ignored:", cmd.get("command_id"))
        return
    seen_commands.append(cmd.get("command_id"))
    del seen_commands[:-100]
    print("Executing:", cmd)


def flush_outbox(client):
    rows = db.execute("SELECT id, payload FROM outbox ORDER BY id LIMIT 20").fetchall()
    for row_id, payload in rows:
        if row_id in pending.values():
            continue
        info = client.publish(TELEMETRY, payload, qos=1)
        pending[info.mid] = row_id


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=DEVICE)
client.will_set(STATUS, "offline", qos=1, retain=True)
client.on_connect, client.on_disconnect = on_connect, on_disconnect
client.on_publish, client.on_message = on_publish, on_message

# Yeniden bağlanmalar arasında exponential backoff; rastgele minimum cihazları zamana yayar
client.reconnect_delay_set(min_delay=random.randint(1, 5), max_delay=120)
client.connect_async(HOST, PORT, keepalive=30)  # başarısız olursa döngü tarafından yeniden denenir
client.loop_start()

while True:
    reading = {"temperature": round(random.uniform(20, 25), 2), "ts": int(time.time())}
    db.execute("INSERT INTO outbox (payload) VALUES (?)", (json.dumps(reading),))
    db.commit()
    if connected:
        flush_outbox(client)
    time.sleep(5)

Ağ bağlantınızı bir dakikalığına keserek test edin. Okumalar outbox.db içinde birikir, ağ döngüsü artan gecikmelerle yeniden denemeye devam eder ve ağ geri geldiğinde birikmiş kayıtlar her döngüde en fazla yirmi satır olmak üzere sırayla boşaltılır. Paho denemeler arasındaki gecikmeyi max_delay değerine kadar ikiye katlar ancak kendi başına jitter eklemez; bu yüzden her cihaz rastgele bir başlangıç gecikmesi seçer. Büyük bir filoda daha da ileri gidip kendi yeniden bağlanma döngünüzle her denemeye rastgele jitter ekleyebilirsiniz.

Oturum Durumu ve Varsaymamanız Gerekenler

Kalıcı oturumlar (MQTT 3.1.1'de clean session false ya da MQTT 5'te bir session expiry interval), broker'ın bağlantısı kopmuş bir istemci için QoS 1 ve 2 mesajlarını kuyruğa almasını sağlar. İstemci tarafı kesintilerde işe yararlar. Broker tarafı kesintilerde ise yalnızca broker oturumları yeniden başlatmalar arasında kalıcı tutuyorsa yardımcı olurlar; her broker bunu yapmaz. ConnectMQ yeniden başlatmalar arasında kalıcılığı garanti etmez ve herkese açık broker'da oturumların süresi en fazla 30 dakika sonra dolar.

Bu yüzden savunmacı tasarlayın:

1. Broker'ın abonelikleri hatırlamasına güvenmek yerine on_connect içinde her seferinde yeniden abone olun.
2. Yeniden bağlandıktan sonra retained durumu ve yapılandırmayı yeniden yayınlayın.
3. Kaybetmeyi göze alamayacağınız veriler için doğruluk kaynağı olarak broker kuyruklarını değil, cihaz tarafı tamponlamayı kabul edin.
4. Bir kesintiden sonra durumu backend'in uzlaştırmasına izin verin; örneğin cihazlardan güncel yapılandırmalarını yeniden yayınlamalarını isteyerek.

Kesintiye Hazırlık Kontrol Listesi

Bu istemci kalıplarını tamamlayan broker tarafı önlemler için MQTT broker yüksek erişilebilirlik ve MQTT broker izleme rehberlerimize göz atın ve yayına almadan önce MQTT canlı ortam kontrol listesinden geçin.

Dayanıklı İstemcilerinizi ConnectMQ Üzerinde Çalıştırın

Bir yönetilen MQTT broker, süresi dolmuş sertifikalar ve yamalanmamış sunucular gibi en yaygın kendi kendine yol açılan kesintileri ortadan kaldırır; geri kalanını istemcileriniz halleder. app.connectmq.com adresinde kredi kartı gerekmeden Serverless planında ücretsiz bir broker oluşturun, bu kalıpları onun üzerinde test edin ve yeniden bağlanmaları izlemek için portalın $SYS bağlantı durumunu ve alarm günlüğünü kullanın. ConnectMQ hakkında daha fazla bilgi edinin ya da seçenekleri bulut MQTT broker rehberimizde karşılaştırın.

Sık Sorulan Sorular

Bir IoT istemcisi MQTT broker kesintisinden sonra ne kadar hızlı yeniden bağlanmalı?

Önce hızlıca deneyin, ardından her gecikmeye rastgele jitter ekleyerek bir ila iki dakikalık bir üst sınıra kadar üstel olarak geri çekilin. Bu, kısa aksaklıklardan hızla toparlanmayı sağlarken uzun bir kesintiden sonra broker'ı aşırı yüklemez.

QoS 1, broker kesintisi sırasında veriyi korur mu?

Yalnızca istemci kütüphanesine zaten teslim edilmiş mesajlar için ve yalnızca cihaz güç almaya devam ettiği sürece. Kaybetmeyi göze alamayacağınız veriler için PUBACK gelene kadar veriyi kalıcı yerel depolamada tamponlayın.

Last Will mesajı ne zaman gönderilir?

Broker anormal bir bağlantı kopmasını tespit ettiğinde; bu genellikle keepalive aralığının yaklaşık 1,5 katı sonra olur. Temiz bir disconnect'te gönderilmez ve broker'ın kendisi çökmüşken hiçbir şey gönderilmez.

Mükerrerleri önlemek için komutlarda QoS 2 mi kullanılmalı?

QoS 2 istemci ile broker arasındaki mükerrerleri önler, ancak yeniden bağlanmalar veya uygulama düzeyindeki yeniden denemelerden sonra uçtan uca mükerrerler yine de oluşabilir. Benzersiz ID'lere sahip idempotent komutlar daha sağlam bir çözümdür.