Python ve SQLite ile Çevrimdışı MQTT Mesaj Tamponlama (Store and Forward)

Python'da MQTT mesajlarını çevrimdışıyken tamponlamak için her okumayı önce yerel bir SQLite kuyruğuna yazın ve ayrı bir göndericinin bu kuyruğu yalnızca bağlantı varken broker'a boşaltmasına izin verin. Bu eğitimde paho-mqtt 2.x ile bu store-and-forward yayıncıyı oluşturacaksınız: ağ kesintilerinden ve yeniden başlatmalardan sağ çıkar, mesajları orijinal sırasıyla yeniden gönderir, yalnızca broker onayladıktan sonra siler ve disk kullanımını sınırlı tutar. Raspberry Pi'da veya herhangi bir Linux makinede çalışır.

paho-mqtt'nin Yerleşik Kuyruğu Neden Yetmez?

paho-mqtt yeniden bağlanırken giden mesajları zaten kuyruğa alır; peki neden bir şey inşa edelim? Çünkü bu kuyruk bellekte yaşar. Süreç yeniden başlarsa, Pi yeniden açılırsa veya bir kesinti sırasında elektrik giderse içindeki her şey kaybolur. Hücresel bağlantıdaki bir gateway, Wi-Fi'ı güvenilmez bir saha veya tünellerden geçen bir araç için önemli olan anlar tam da bunlardır.

Broker'ın yayıncılara nasıl davrandığıyla ilgili ince bir sorun da vardır. Kalıcı oturumlar ve QoS 1 veya 2, geçici olarak çevrimdışı olan abonelere giden mesajları korur. Broker'a hiç ulaşamayan bir yayıncı için hiçbir şey yapmazlar. Yayıncı tarafındaki dayanıklılık client'ın işidir.

Her iki sorunu da çözen tasarım klasik bir outbox'tır:

1. Uygulama asla doğrudan yayınlamaz. Bir SQLite tablosuna satır ekler: topic, payload, QoS, retain bayrağı ve zaman damgası.
2. Bir gönderici thread en eski satırları okur, bunları QoS 1 ile yayınlar ve broker'ın PUBACK'ini bekler.
3. Yalnızca onaylanan satırlar silinir. Geri kalan her şey bir sonraki deneme için kalır.

SQLite bir dosya olduğu için kuyruk yeniden başlatmalardan sağ çıkar. Gönderici satırları tek bir bağlantı üzerinden ekleme sırasına göre işlediği için sıra korunur.

Outbox Tablosunun Tasarımı

outbox
  id        INTEGER PRIMARY KEY AUTOINCREMENT   -- defines replay order
  topic     TEXT NOT NULL
  payload   BLOB NOT NULL
  qos       INTEGER NOT NULL DEFAULT 1
  retain    INTEGER NOT NULL DEFAULT 0
  created   REAL NOT NULL                      -- unix time of the reading

İki SQLite ayarı bir Pi üzerinde büyük fark yaratır. journal_mode=WAL, gönderici okurken üreticinin ekleme yapmasına izin verir ve işlem başına yazma sayısını azaltır. synchronous=NORMAL, WAL modunda güvenlik ile SD kart aşınması arasında makul bir dengedir. Salt okunur bir kök dosya sistemi kullanıyorsanız veritabanını bir SSD'ye veya yazılabilir bir bölüme koyun.

Store-and-Forward Yayıncı

Aşağıdaki modül kendi başına yeterlidir. OfflineBufferedPublisher sınıfını içe aktarın, uygulamanızın herhangi bir yerinden publish() çağırın; gerisini o halleder.

import json
import os
import sqlite3
import threading
import time
import uuid

import paho.mqtt.client as mqtt


class OfflineBufferedPublisher:
    def __init__(self, host, port=1883, client_id=None, username=None, password=None,
                 db_path="outbox.db", max_rows=200_000, batch=20, max_rate=8.0):
        self.db_path = db_path
        self.max_rows = max_rows
        self.batch = batch
        self.min_interval = 1.0 / max_rate if max_rate else 0
        self._lock = threading.Lock()
        self._wake = threading.Event()
        self._stop = threading.Event()

        self.db = sqlite3.connect(db_path, check_same_thread=False)
        self.db.execute("PRAGMA journal_mode=WAL")
        self.db.execute("PRAGMA synchronous=NORMAL")
        self.db.execute("""CREATE TABLE IF NOT EXISTS outbox (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            topic TEXT NOT NULL, payload BLOB NOT NULL,
            qos INTEGER NOT NULL DEFAULT 1, retain INTEGER NOT NULL DEFAULT 0,
            created REAL NOT NULL)""")
        self.db.commit()

        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                                  client_id=client_id or f"buffered-{uuid.uuid4().hex[:8]}")
        if username:
            self.client.username_pw_set(username, password)
        if port == 8883:
            self.client.tls_set()
        self.client.reconnect_delay_set(1, 60)
        self.client.on_connect = self._on_connect
        self.client.on_disconnect = self._on_disconnect
        self.client.connect_async(host, port, keepalive=30)
        self.client.loop_start()

        self.sender = threading.Thread(target=self._drain, daemon=True)
        self.sender.start()

    # ---- üretici tarafı ------------------------------------------------
    def publish(self, topic, payload, qos=1, retain=False):
        if not isinstance(payload, (bytes, bytearray)):
            payload = json.dumps(payload).encode() if not isinstance(payload, str) else payload.encode()
        with self._lock:
            self.db.execute("INSERT INTO outbox (topic, payload, qos, retain, created) VALUES (?,?,?,?,?)",
                            (topic, payload, qos, int(retain), time.time()))
            self._enforce_limit()
            self.db.commit()
        self._wake.set()

    def pending(self):
        with self._lock:
            return self.db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]

    def _enforce_limit(self):
        count = self.db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]
        if count > self.max_rows:
            self.db.execute("DELETE FROM outbox WHERE id IN (SELECT id FROM outbox ORDER BY id LIMIT ?)",
                            (count - self.max_rows,))

    # ---- gönderici tarafı ----------------------------------------------
    def _on_connect(self, client, userdata, flags, reason_code, properties):
        print("connected:", reason_code)
        self._wake.set()

    def _on_disconnect(self, client, userdata, flags, reason_code, properties):
        print("disconnected:", reason_code)

    def _drain(self):
        while not self._stop.is_set():
            if not self.client.is_connected():
                self._wake.wait(2)
                self._wake.clear()
                continue
            with self._lock:
                rows = self.db.execute(
                    "SELECT id, topic, payload, qos, retain FROM outbox ORDER BY id LIMIT ?",
                    (self.batch,)).fetchall()
            if not rows:
                self._wake.wait(5)
                self._wake.clear()
                continue
            for row_id, topic, payload, qos, retain in rows:
                info = self.client.publish(topic, payload, qos=qos, retain=bool(retain))
                try:
                    info.wait_for_publish(timeout=10)
                except (ValueError, RuntimeError):
                    break
                if not info.is_published():
                    break  # bağlantı koptu; bu satırı ve sonrasındakileri sakla
                with self._lock:
                    self.db.execute("DELETE FROM outbox WHERE id = ?", (row_id,))
                    self.db.commit()
                if self.min_interval:
                    time.sleep(self.min_interval)

    def close(self):
        self._stop.set()
        self._wake.set()
        self.sender.join(timeout=5)
        self.client.loop_stop()
        self.client.disconnect()
        self.db.close()


if __name__ == "__main__":
    pub = OfflineBufferedPublisher(
        host=os.getenv("MQTT_HOST", "broker.connectmq.com"),
        port=int(os.getenv("MQTT_PORT", "1883")),
        client_id=os.getenv("MQTT_CLIENT_ID"),
        username=os.getenv("MQTT_USERNAME"),
        password=os.getenv("MQTT_PASSWORD"),
    )
    prefix = os.getenv("TOPIC_PREFIX", "connectmq/demo/q5w8e1/sensor1")
    n = 0
    try:
        while True:
            n += 1
            pub.publish(f"{prefix}/reading", {"seq": n, "ts": round(time.time(), 3), "value": 20 + n % 5})
            print("queued", n, "pending", pub.pending())
            time.sleep(2)
    except KeyboardInterrupt:
        pub.close()

Bir Kesintiyi Test Etme

Sıra numaralarını yazdıran bir abone başlatın, ardından yayıncıyı çalıştırın:

mosquitto_sub -h broker.connectmq.com -p 1883 -i buffer-viewer-q5w8e1 -t "connectmq/demo/q5w8e1/#" -v
python buffered_publisher.py

Şimdi bir kesinti simüle edin. Bir Pi üzerinde ağı sudo ip link set wlan0 down ile kapatabilirsiniz (bunu aynı arayüz üzerinden SSH ile değil, yerel bir konsoldan yapın); bir dakika bekleyin, ardından sudo ip link set wlan0 up ile geri açın. Kesinti sırasında "pending" sayacı artar. paho yeniden bağlandığında gönderici birikmiş kuyruğu boşaltır ve abone eksik her sıra numarasını sırasıyla, ardından canlı okumaları alır. Kesinti sırasında betiği durdurup yeniden başlatmayı deneyin: birikmiş kuyruk hâlâ oradadır.

Sıralama, Tekrarlar ve Zaman Damgaları

Sıra. Gönderici her seferinde bir mesaj yayınlar ve silip devam etmeden önce onayını bekler. Tek bir MQTT bağlantısı üzerinde bu, orijinal sırayı korur. Birden fazla uçuştaki (in-flight) mesaja izin vererek verimi artırabilirsiniz, ancak sıra önemliyse tek bir gönderici thread ve tek bir bağlantı kullanın.

Tekrarlar. QoS 1 en az bir kez teslimattır. Bağlantı, broker mesajı aldıktan sonra ama PUBACK gelmeden önce koparsa satır yeniden bağlandıktan sonra tekrar gönderilir. Tüketicilerin tekrarları atabilmesi için payload'a bir sıra numarası veya benzersiz bir ID ekleyin. MQTT QoS seviyeleri yazısı ödünleşimleri ve QoS 2'nin neden yine de sizi bir client yeniden başlatmasından korumadığını açıklar.

Zaman damgaları. Yeniden gönderilen bir mesaj dakikalar ya da saatler gecikmeli gelir. Ölçüm zamanını her zaman payload'a koyun ve tüketicilerin varış zamanı yerine onu kullanmasını sağlayın. MQTT 5 ile, canlı konum gibi bir süre sonra anlamını yitiren okumalar için bir mesaj sona erme aralığı (message expiry interval) da ayarlayabilirsiniz.

Gerçek Kurulumlar için Ayarlama

Kuyruk dolduğunda bu uygulama en eski satırları atar; bu, en yeni verinin en önemli olduğu telemetri için uygundur. Olay kayıtları için bunun yerine yeni satır kabul etmeyi durdurup bir alarm üretmeyi tercih edebilirsiniz. Her iki durumda da pending() sayısını bir sağlık metriği olarak yayınlayın; böylece büyüyen bir birikme panonuzda görünür olur.

Yayıncıyı, Python MQTT betiğini açılışta çalıştırma yazısında gösterildiği gibi cihazla birlikte yeniden başlaması için bir systemd servisi olarak çalıştırın. MQTT verisini yerel olarak saklamanın diğer yolları için abone tarafını ele alan Python ile MQTT'den SQLite'a yazısına bakın.

Alternatif: Bridge ile Yerel Bir Broker

Yaygın bir diğer store-and-forward tasarımı, Pi'nin kendisine bir Mosquitto broker koyar. Yerel uygulamalar localhost'a yayın yapar, Mosquitto da topic'leri bulut broker'ına köprüler ve uplink kesikken mesajları kuyruğa alır. Aynı cihazdaki birkaç süreç veri yayınladığında veya uygulama kodunu değiştiremediğinizde iyi bir seçimdir. Mosquitto bridge yapılandırması rehberi ayarları ele alır.

Ödünleşimleri bilmekte fayda var. Artık her gateway'de kendi yapılandırması, güncellemeleri ve kimlik bilgileri olan bir broker işletirsiniz. Mosquitto'nun kuyruk limitleri ve kalıcılık davranışı yeniden başlatmalardan sağ çıkmak için dikkatli ayar gerektirir ve veri boşluklarını ayıklamak bir yerine iki broker'a bakmak demektir. Python outbox ise her şeyi tamamen kontrol ettiğiniz ve bir birim testiyle test edebileceğiniz tek bir süreçte tutar; birçok ekip bunu tek amaçlı bir gateway için daha basit bulur. Birçok yerel üretici tek bir uplink'i paylaşıyorsa bridge'i, veri tek bir uygulamaya aitse outbox'ı seçin.

Üretim Broker Ayarları

Ücretsiz genel MQTT broker bu test için uygundur, ancak orada yayınlanan her şey herkese açıktır. Gerçek veriler için ConnectMQ üzerinde özel bir broker oluşturun, ardından yayıncıyı portaldan alınan MQTT_PORT=8883, MQTT_USERNAME=YOUR_USERNAME, MQTT_PASSWORD=YOUR_PASSWORD ve MQTT_CLIENT_ID=YOUR_CLIENT_ID değerleriyle başlatın. Yönetilen bir MQTT broker sunucu tarafında çalışma süresini üstlenirken bu outbox client tarafındaki boşlukları kapatır. paho-mqtt dokümantasyonu, MQTTMessageInfo ve yeniden bağlanma seçeneklerini ayrıntılı olarak açıklar.

Bu desen, CAN bus okuyucularından LoRa alıcılarına kadar Raspberry Pi MQTT eğitimi serisindeki her gateway için temel bir yapı taşıdır.

ConnectMQ ile Tek Bir Okumayı Bile Kaybetmeyin

Dayanıklı bir yerel outbox'ı güvenilir bir broker ile birleştirin, kesintiler panolarınız için görünmez hale gelsin. Kredi kartı gerekmeden app.connectmq.com adresinde ücretsiz bir Serverless broker oluşturun, bu yayıncıyı TLS üzerinden ona yönlendirin ve bir sonraki ağ kesintinizden sonra birikmiş kuyruğun ConnectMQ canlı topic gezginine boşalmasını izleyin.

Sık Sorulan Sorular

MQTT kalıcı oturumu bir yayıncı için mesajları tamponlar mı?

Hayır. Kalıcı oturumlar, çevrimdışı olan aboneler için mesajları saklar. Broker'a ulaşamayan bir yayıncı yerel olarak tamponlama yapmalıdır; bu outbox'ın yaptığı da budur.

Yeniden gönderilen mesajlar doğru sırayla gelir mi?

Evet; bu koddaki gibi tek bir gönderici satırları tek bir bağlantı üzerinden ID sırasına göre yayınladığı ve her onayı beklediği sürece.

Bunu asyncio ile kullanabilir miyim?

Evet. Aynı outbox tasarımı aiomqtt ile çalışır: boşaltma döngüsünü bir async görev olarak çalıştırın ve aiosqlite kullanın ya da SQLite çağrılarını bir thread executor'da çalıştırın.

SD kartın aşınmasını nasıl önlerim?

WAL modunu kullanın, payload'ları kompakt tutun ve mümkünse veritabanını bir SSD'ye yerleştirin. Birkaç saniyede bir okuma yazmak, iyi bir depolama için mütevazı bir yüktür.