MQTT ve Redis: Cihazların Son Durumunu Önbelleğe Alın

MQTT'den gelen son cihaz durumunu Redis'te önbelleğe almak için telemetri topic'lerinizi dinleyen ve her mesajı, key'i cihaz ID'si olan bir Redis hash'ine yazan küçük bir subscriber çalıştırın. API'leriniz, dashboard'larınız ve arka plan işleriniz güncel durumu MQTT bağlantısı açmadan ve bir sonraki mesajı beklemeden tek bir HGETALL ile okur.

Bu eğitim söz konusu subscriber'ı Python ile kuruyor, onu sırasız güncellemelere karşı koruyor, yakın geçmiş için bir Redis Stream ekliyor ve yaklaşımı retained mesajlar ve Redis Pub/Sub ile karşılaştırıyor.

Redis önbelleği ne zaman mantıklı

MQTT'nin zaten bir "son değer" özelliği var: retained mesajlar. Yeni bir subscriber, her topic'teki son retained mesajı hemen alır. O zaman neden Redis ekleyelim?

Son değere yalnızca MQTT istemcileri ihtiyaç duyuyorsa retained mesajlar yeterlidir. MQTT konuşmayan kod durumu okumaya başladığı anda Redis önbelleği daha basit ve daha hızlıdır.

Veri modeli

Sıkıcı ve öngörülebilir tutun:

• device:<id>: son alanları içeren bir hash: temp, humidity, battery, ts (cihaz timestamp'i), topic, seen (köprünün yazdığı an).
• devices:online: status topic'lerinden beslenen, isteğe bağlı bir cihaz ID'leri set'i.
• telemetry: replay ve hata ayıklama için son ham mesajları tutan isteğe bağlı bir stream.

Cihazlar testmqtt/redis/<device>/state topic'ine JSON publish eder; örneğin {"ts": 1760700000123, "temp": 21.4, "battery": 87}. Cihazdan gelen bir timestamp (veya sıra numarası), ileride eski güncellemeleri reddetmenizi sağlayan şeydir. Parse etme ipuçları için Python'da JSON payload'ları yazısına bakın.

Subscriber

Bu köprü paho-mqtt 2.x ve redis-py kullanır. Lua script'i hash'i yalnızca gelen timestamp kayıtlı olandan yeniyse yazar ve sessiz cihazların kendiliğinden kaybolması için bir TTL'i yeniler:

import json, os, time
import paho.mqtt.client as mqtt
import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

# KEYS[1] = cihaz key'i; ARGV = ts, ttl, ardından alan/değer çiftleri
UPSERT_IF_NEWER = r.register_script("""
local cur = redis.call('HGET', KEYS[1], 'ts')
if cur and tonumber(cur) >= tonumber(ARGV[1]) then
  return 0
end
redis.call('HSET', KEYS[1], 'ts', ARGV[1], unpack(ARGV, 3))
redis.call('EXPIRE', KEYS[1], ARGV[2])
return 1
""")

TTL_SECONDS = 900   # 15 dakika sessiz kalan cihazları unut

def on_connect(client, userdata, flags, reason_code, properties):
    print("bağlandı:", reason_code)
    client.subscribe("testmqtt/redis/+/state", qos=1)

def on_message(client, userdata, msg):
    device = msg.topic.split("/")[2]
    try:
        data = json.loads(msg.payload)
    except ValueError:
        print("JSON olmayan payload atlandı:", msg.topic)
        return

    ts = int(data.pop("ts", time.time() * 1000))
    fields = []
    for k, v in data.items():
        plain = isinstance(v, (str, int, float)) and not isinstance(v, bool)
        fields += [k, v if plain else json.dumps(v)]   # bool/None/nesneler JSON olarak
    fields += ["topic", msg.topic, "seen", int(time.time() * 1000)]

    written = UPSERT_IF_NEWER(keys=[f"device:{device}"], args=[ts, TTL_SECONDS, *fields])
    if not written:
        print(f"{device} için eski güncelleme yok sayıldı (ts={ts})")

    # isteğe bağlı kısa geçmiş, ~100k kayıtla sınırlı
    r.xadd("telemetry", {"device": device, "payload": msg.payload}, maxlen=100_000, approximate=True)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="redis-state-bridge")
client.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
client.tls_set()
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 8883)
client.loop_forever()

Tasarımla ilgili notlar:

• Atomik compare-and-set. Lua script'leri Redis'te atomik çalışır; iki köprü instance'ı (veya bir QoS 1 yeniden iletimi) yarışıp daha eski bir durumu yazamaz.
• Düz alanlar. String'ler ve sayılar olduğu gibi yazılır; boolean'lar, null değerler ve iç içe nesneler JSON string olarak saklanır. Redis hash'leri tek seviyelidir, buna karşı savaşmayın.
• Çevrimdışı sinyali olarak TTL. Bir key'in süresi dolarsa cihaz TTL_SECONDS boyunca veri göndermemiş demektir. Açık bir sinyal için bunu bir status topic'indeki Last Will mesajıyla birleştirin.
• Kopyalar zararsızdır. Aynı durumu iki kez yazmak hiçbir şeyi değiştirmez; bu da bu hedefi doğal olarak idempotent yapar.

Durumu okuma

Artık her servis, MQTT olmadan "X cihazı şu anda ne durumda?" sorusunu yanıtlayabilir:

redis-cli HGETALL device:sensor-1
redis-cli HGET device:sensor-1 temp
redis-cli TTL device:sensor-1
redis-cli --scan --pattern 'device:*'

Canlı güncellenen bir dashboard için ilk sayfa yüklemesini Redis'ten okuyan bir REST endpoint'iyle sunun, sonraki değişiklikler için tarayıcının MQTT'ye güvenli WebSocket üzerinden (port 8084, path /mqtt) subscribe olmasına izin verin. Önbellek "şimdi"yi yanıtlar, MQTT "sonraki"ni iletir.

Yakın geçmiş için Redis Streams

XADD satırı ham mesajların boyutu sınırlı bir log'unu tutar. Stream'ler consumer group'ları destekler; böylece birden çok worker güncellemeleri onaylı şekilde işleyebilir ve hata ayıklarken bir zaman aralığını yeniden oynatabilirsiniz:

redis-cli XLEN telemetry
redis-cli XREVRANGE telemetry + - COUNT 5

redis-cli XGROUP CREATE telemetry alerts $ MKSTREAM
redis-cli XREADGROUP GROUP alerts worker-1 COUNT 10 BLOCK 5000 STREAMS telemetry '>'

Stream'i kısa tutun (MAXLEN); bellekte yaşar. Uzun vadeli geçmiş için bir zaman serisi veritabanına veya Postgres'e yazın; MQTT'den PostgreSQL'e ve MQTT veri saklama ve depolama yazılarına bakın.

MQTT vs Redis Pub/Sub

Redis'in de bir Pub/Sub özelliği var ve MQTT'yi tamamen atlamak cazip gelebilir. İkisi birbirinin yerine geçmez:

• Redis Pub/Sub, o anda Redis'e bağlı süreçler arasında fire-and-forget çalışır. QoS, retained değer, Last Will veya cihazlar için tasarlanmış topic bazlı bir ACL modeli yoktur; istemcilerin bir Redis bağlantısı tutması gerekir.
• MQTT, güvenilmez ağlardaki cihazlar için tasarlanmıştır: QoS 1 ve 2, kalıcı session'lar, retained mesajlar, keep alive, Last Will ve mikrodenetleyiciler için hafif istemciler.

Yaygın bir ayrım, uçta MQTT ve backend içinde Redis kullanmaktır. Ayrıntılı karşılaştırma MQTT vs Redis Pub/Sub yazısında.

Test edin

Redis'i yerelde başlatın ve köprüyü çalıştırın:

docker run -d --name redis -p 6379:6379 redis:7
python mqtt_redis_state.py

Önce yeni bir durum, ardından daha küçük timestamp'li eski bir durum publish edin:

mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
  -t testmqtt/redis/sensor-1/state -m '{"ts": 2000, "temp": 22.5, "battery": 90}'

mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
  -t testmqtt/redis/sensor-1/state -m '{"ts": 1000, "temp": 19.0, "battery": 95}'

redis-cli HGETALL device:sensor-1

temp hâlâ 22.5 olmalı ve köprü sensor-1 için eski güncelleme yok sayıldı diye log basmalıdır. Payload'ları online MQTT istemcisinden elle de gönderebilir ve köprünün çalıştırdığı her komutu görmek için redis-cli MONITOR çıktısını izleyebilirsiniz.

Test durumlarını tarayıcıdan publish edin

Redis durum önbelleği, bir MQTT sistemine yapılabilecek en ucuz iyileştirmelerden biridir: tek subscriber, cihaz başına tek hash ve her servisin "şimdi"ye anında erişimi. Timestamp kontrolünü ilk günden ekleyin, geçmişi hash'in dışında tutun ve MQTT'nin en iyi yaptığı işi, yani güncellemeleri taşımayı sürdürmesine izin verin.

Sıkça sorulan sorular