MQTT Store and Forward: Kesintilerde Edge Verisini Tamponlamak ve Sırasıyla Yeniden Oynatmak

MQTT store and forward, edge ile bulut arasındaki ağ çöktüğünde endüstriyel verinin bozulmadan kalmasını sağlayan desendir. Edge cihazı, broker'a ulaşılamadığı sürece mesajları düşürmek yerine onları yerel depolamaya yazar ve bağlantı geri geldiğinde sırasıyla yeniden oynatır. Bu yazı desenin nasıl çalıştığını ve önemli tasarım kararlarını açıklıyor; ayrıca SQLite destekli, çalıştırılabilir bir Python uygulaması sunuyor.

Neden Tek Başına MQTT Yeterli Değil?

MQTT'nin güçlü teslim özellikleri vardır: QoS 1 ve 2 onayları, kalıcı oturumlar (persistent session) ve retained mesajlar. Ancak bunların hepsi ancak bir mesaj cihazdan çıktıktan sonra devreye girer. Bir tesiste iki saatlik bir WAN kesintisi sırasında ne olduğunu düşünün:

• Gateway makineleri sorgulamaya ve değer üretmeye devam eder.
• MQTT client bağlanamaz; dolayısıyla mesajların teslim edileceği bir oturum yoktur.
• Çoğu client kütüphanesi gönderilmemiş mesajları boyut sınırıyla bellekte tutar ve bir yeniden başlatma veya elektrik kesintisinde kaybeder.

Broker üzerindeki kalıcı oturumlar diğer yönde yardımcı olur: çevrim dışı bir abone için mesajları tutarlar. Broker'a ulaşamayan bir yayıncı için ise hiçbir şey yapmazlar. Store and forward'ın doldurduğu boşluk budur ve tamponlamanın her IIoT edge gateway için temel bir görev olarak sayılmasının nedeni de budur.

MQTT Store and Forward Nasıl Çalışır?

  Data source --> [ write to local queue (disk) ] --> forwarder --> MQTT broker
                         ^                               |
                         |      delete only after        |
                         +------- PUBACK received -------+

1. Her mesaj önce kalıcı bir yerel kuyruğa yazılır; bir sıra numarası ve kaynak zaman damgasıyla birlikte.
2. Bir iletici (forwarder) en eski mesajları okur ve QoS 1 ile yayınlar.
3. Mesaj ancak broker onayladıktan sonra (PUBACK) kuyruktan silinir.
4. Bağlantı koptuğunda iletici durur, kuyruk büyür ve veri kaynakları etkilenmeden çalışmaya devam eder.
5. Yeniden bağlanınca iletici birikmiş kuyruğu kontrollü bir hızla en eskiden başlayarak boşaltır, ardından canlı veriyle devam eder.

Yalnızca kesinti sırasında üretilen veriyi değil, tüm veriyi kuyruktan geçirmek sıralamayı basit tutar: tek bir yol vardır ve bu yol her zaman ilk giren ilk çıkar (FIFO) mantığıyla çalışır.

Önemli Tasarım Kararları

Tamponu Boyutlandırmak

Bunu üç sayıdan tahmin edin: normal çalışmadaki saniye başına mesaj sayısı, ek yük dahil ortalama mesaj boyutu ve atlatmak istediğiniz en uzun kesinti. Örneğin 24 saatlik bir kesinti boyunca saniyede 200 baytlık 20 mesaj, depolama ek yükü hariç yaklaşık 345 MB payload demektir. Bu çok fazlaysa, değişimde raporlama (report by exception) ile trafiği kaynağında azaltmak gereksinimi çoğu zaman çarpıcı biçimde küçültür.

Düşürme Politikası Seçmek

Tampon dolduğunda seçim yapmanız gerekir: en eski veriyi düşürmek (son durumu korumak) ya da en yeniyi düşürmek (kesintinin başlangıcından itibaren kesintisiz bir geçmiş korumak). İzleme için genellikle en eskiyi düşürmek doğrudur. Uyumluluk veya parti kayıtları için ise düşük öncelikli yeni verileri kabul etmeyi durdurup başlangıcı korumayı tercih edebilirsiniz. Her iki durumda da boşluğun görünür olması için düşürülen mesajların sayacını yayınlayın.

Yeniden Oynatma ve Retained Son Değer

Panellerin "güncel değer" olarak da kullandığı bir topic'e saatlerce geçmişi yeniden oynatmak, bir göstergenin eski değerler arasında zıplamasına neden olabilir. Yaygın iki çözüm vardır: birikmiş veriyi .../history gibi ayrı bir topic'e göndermek ya da kaynak zaman damgasını ekleyip tüketicilerin görüntüleme için daha eski değerleri yok saymasını sağlamak. MQTT 5 ile bayatladığında değersiz olan değerlere bir mesaj sona erme aralığı (message expiry interval) da ayarlayabilirsiniz; böylece broker onları geç teslim etmek yerine atar.

Python ile Bir Store and Forward Uygulaması

Aşağıdaki betik kalıcı kuyruk olarak SQLite'ı, client olarak da paho-mqtt 2.x'i kullanır. Bir üretici iş parçacığı, bağlantı durumundan bağımsız olarak her saniye bir ölçüm üretir. İletici en eski satırları QoS 1 ile yayınlar, her onayı bekler, teslim edilen satırları siler ve genel broker'ın saniyede 10 mesaj sınırının altında kalmak için yeniden oynatmayı kısıtlar.

import json
import sqlite3
import threading
import time
import random
from datetime import datetime, timezone

import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PORT = 1883
CLIENT_ID = "saf-gateway-2b7c"
TOPIC = "connectmq/demo/saf-2b7c/plant1/compressor1/telemetry"
DB_PATH = "saf_queue.db"
MAX_ROWS = 500_000          # tampon sınırı; bunun ötesinde en eski satırlar düşürülür
REPLAY_RATE = 8             # saniyede mesaj (genel broker 10'a izin verir)
ACK_TIMEOUT = 10            # PUBACK için beklenecek saniye

db_lock = threading.Lock()
db = sqlite3.connect(DB_PATH, check_same_thread=False)
db.execute("PRAGMA journal_mode=WAL")
db.execute("""CREATE TABLE IF NOT EXISTS queue (
                seq INTEGER PRIMARY KEY AUTOINCREMENT,
                topic TEXT NOT NULL,
                payload TEXT NOT NULL)""")
db.commit()


def enqueue(topic, data):
    with db_lock:
        db.execute("INSERT INTO queue (topic, payload) VALUES (?, ?)",
                   (topic, json.dumps(data)))
        count = db.execute("SELECT COUNT(*) FROM queue").fetchone()[0]
        if count > MAX_ROWS:
            db.execute("DELETE FROM queue WHERE seq IN "
                       "(SELECT seq FROM queue ORDER BY seq LIMIT ?)", (count - MAX_ROWS,))
        db.commit()


def oldest(n=50):
    with db_lock:
        return db.execute("SELECT seq, topic, payload FROM queue ORDER BY seq LIMIT ?",
                          (n,)).fetchall()


def delete(seq):
    with db_lock:
        db.execute("DELETE FROM queue WHERE seq = ?", (seq,))
        db.commit()


def producer():
    """Simulated data source: never blocks on the network."""
    while True:
        enqueue(TOPIC, {
            "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
            "pressure_bar": round(7 + random.gauss(0, 0.1), 3),
            "motor_temp_c": round(65 + random.gauss(0, 0.5), 2),
        })
        time.sleep(1)


def on_connect(client, userdata, flags, reason_code, properties):
    print("Connected:", reason_code)


def on_disconnect(client, userdata, flags, reason_code, properties):
    print("Disconnected:", reason_code, "- buffering locally")


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.reconnect_delay_set(min_delay=1, max_delay=60)
# Üretim: client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD"); client.tls_set(); PORT = 8883
client.connect_async(BROKER, PORT, keepalive=30)
client.loop_start()

threading.Thread(target=producer, daemon=True).start()

interval = 1.0 / REPLAY_RATE
while True:
    if not client.is_connected():
        time.sleep(1)
        continue
    rows = oldest()
    if not rows:
        time.sleep(0.5)
        continue
    for seq, topic, payload in rows:
        data = json.loads(payload)
        data["seq"] = seq                      # tüketicilerin tekilleştirmesini sağlar
        try:
            info = client.publish(topic, json.dumps(data), qos=1)
            info.wait_for_publish(timeout=ACK_TIMEOUT)
        except (RuntimeError, ValueError) as exc:
            print("Publish failed:", exc)
            break
        if not info.is_published():
            print("No PUBACK, will retry from seq", seq)
            break                              # satırı tut, daha sonra sırayla yeniden dene
        delete(seq)
        time.sleep(interval)
    with db_lock:
        backlog = db.execute("SELECT COUNT(*) FROM queue").fetchone()[0]
    if backlog > 1:
        print("Backlog:", backlog)

Test etmek için betiği çalıştırın, topic'i başka bir terminalde izleyin, ardından ağ bağlantınızı bir iki dakikalığına kesin:

mosquitto_sub -h broker.connectmq.com -p 1883 -i saf-2b7c-viewer -t "connectmq/demo/saf-2b7c/#" -v

Çevrim dışıyken betik bağlantı kopmasını yazdırır ve kuyruğa almaya devam eder. Ağ geri geldiğinde client otomatik olarak yeniden bağlanır ve birikmiş kuyruk sıra numarasına göre boşalır. ts alanı her değerin ne zaman ölçüldüğünü gösterir ve seq boşluksuz artar. Kesinti sırasında betiği tamamen de durdurabilirsiniz; yeniden başlattığınızda SQLite kuyruğu hâlâ yerindedir.

Uygulamaya İlişkin Notlar

• Aynı anda yalnızca bir mesajın yolda olması sıralamayı basit tutar ve çoğu gateway için yeterince hızlıdır. Daha yüksek verim için küçük bir pencere yayınlayın ve onaylar geldikçe satırları silin.
• Tam olarak bir kez değil, en az bir kez. Broker mesajı kabul ettikten sonra PUBACK kaybolursa satır yeniden gönderilir. seq alanı bir historian'ın veya abonenin tekrarları atmasını sağlar. MQTT QoS 2 hat üzerindeki tekrarları önler, ancak bir gateway yeniden başlatmasında önlemez; bu yüzden uygulama düzeyinde kimliklere yine de sahip olmaya değer. Ödünleşimler için MQTT QoS seviyeleri yazısına bakın.
• SQLite'taki WAL modu, okuma ve yazmaların iyi bir dayanıklılıkla eşzamanlı ilerlemesine olanak tanır.
• Kısıtlanmış yeniden oynatma broker'ı ve sonraki tüketicileri korur. Paylaşımlı planlarda client başına sınırlar vardır ve ConnectMQ'nun ücretsiz Serverless planı, bu desenin tam olarak ihtiyaç duyduğu QoS 1'i destekler.

MQTT Store and Forward Kontrol Listesi

1. Her mesajı göndermeye çalışmadan önce diske yazın.
2. Her payload'a bir kaynak zaman damgası ve bir sıra numarası ekleyin.
3. Tampondan ancak broker onayladıktan sonra silin.
4. En eskiden başlayarak, broker ve tüketici sınırlarının altında kısıtlanmış şekilde yeniden oynatın.
5. Maksimum bir tampon boyutu ve düşürme politikası tanımlayın, düşürmeleri raporlayın.
6. Yeniden oynatılan geçmişin "güncel değer" topic'leriyle nasıl etkileşeceğine karar verin.
7. Gerçek kesintilerle test edin: kabloyu çekin, kesintinin ortasında gateway'i yeniden başlatın, tamponu doldurun.

Hücresel Bağlantılarda ve Uzak Sahalarda Store and Forward

Hücresel bağlantılardaki uzak varlıklar bağlantıyı daha sık kaybeder ve bayt başına ödeme yapar. Orada store and forward; gruplama (mesaj başına birden fazla ölçüm) ve sıkıştırma ile doğal olarak birleşir. Hücresel ağ üzerinden MQTT rehberi bu bağlantılar için keep alive ayarını ve veri bütçelerini ele alıyor; MQTT ile endüstriyel IoT rehberi ise tamponlamanın eksiksiz bir tesis mimarisinde nereye oturduğunu gösteriyor.

Diğer Uçta Güvenilir Bir Broker

Store and forward bağlantının edge tarafını korur. Bulut tarafında ise erişilebilir, TLS ile güvenli hale getirilmiş ve yeniden oynatma patlamalarına göre boyutlandırılmış bir broker'a ihtiyacınız vardır. Barındırılan bir MQTT broker broker operasyonlarını üzerinizden alır; buluttaki MQTT broker rehberi de seçenekleri nasıl karşılaştıracağınızı açıklıyor.

ConnectMQ ile Başlayın

Tamponlama yapan gateway'lerinizi ConnectMQ'ya yönlendirin: 8883 portunda TLS, verilmiş kimlik bilgileri, MQTT 5 mesaj sona erme süresi ve bir kesintiden sonra birikmiş kuyruğun boşalmasını izlemek için canlı bir topic gezgini. app.connectmq.com adresinde kredi kartı gerekmeden ücretsiz bir Serverless broker oluşturun.

Sık Sorulan Sorular

MQTT'de yerleşik store and forward var mı?

Broker'a ulaşamayan yayıncılar için yok. QoS ve kalıcı oturumlar, mesaj broker'a ulaştıktan sonra çalışır; ondan önceki tamponlama edge uygulamasının işidir.

Bellekte mi yoksa diskte mi tamponlamalıyım?

Diskte. Bellek tamponları bir yeniden başlatmada veya elektrik kesintisinde kaybolur ve bunlar çoğu zaman ağı çökerten olaylarla aynı anda yaşanır.

Tüketiciler yeniden oynatmadan sonra tekrarlanan mesajları nasıl ele alır?

Her payload'a bir sıra numarası veya benzersiz kimlik ekleyin ve tüketicilerin zaten sakladıkları kimlikleri yok saymasını sağlayın. Historian'lar genellikle zaman damgası ve tag'e göre de tekilleştirme yapar.

Büyük bir yeniden oynatma broker'ı aşırı yükler mi?

Hepsi bir anda gönderilirse hız sınırlarına takılabilir veya aboneleri boğabilir. Örnekteki gibi yeniden oynatmayı sabit bir hızla kısıtlayın ve hızı planınıza göre ayarlayın.