MQTT Mesajlarını Python ile PostgreSQL ve TimescaleDB'ye Kaydedin

MQTT mesajlarını PostgreSQL'de saklamak için topic'lerinize abone olan, her payload'ı ayrıştıran ve mesaj başına bir transaction yerine satırları toplu olarak ekleyen küçük bir tüketici çalıştırın. Üzerine TimescaleDB eklediğinizde aynı tablo; zaman bölümleme, sıkıştırma ve saklama politikalarına sahip bir hypertable'a dönüşür.

Bu eğitimde bu tüketiciyi Python'da paho-mqtt 2.x ve psycopg 3 ile oluşturacağız: hem düz Postgres hem de Timescale'de çalışan bir şema, boyuta ve süreye göre gruplayan bir yazıcı thread'i, yeniden başlatmalar için kalıcı bir oturum ve verinin geldiğini kanıtlayan sorgular.

Mimari

devices ──► MQTT broker ──► Python consumer ──► PostgreSQL / TimescaleDB
                            (on_message → queue → batch writer)

MQTT callback'i hızlı dönmelidir, çünkü paho onu ağ thread'inde çalıştırır. Bu yüzden on_message yalnızca payload'ı ayrıştırır ve bir satırı kuyruğa koyar; ayrı bir yazıcı thread'i kuyruğu boşaltır ve toplu yazar. Tasarımın tamamı bu ayrımdan ibarettir:

İstemci kütüphanesinde yeniyseniz paho-mqtt 2.x eğitimiyle başlayın.

Postgres ve TimescaleDB için şema

Yerel bir TimescaleDB başlatın (eklentisi önceden kurulmuş düz bir PostgreSQL'dir):

docker run -d --name tsdb -p 5432:5432 \
  -e POSTGRES_USER=mqtt -e POSTGRES_PASSWORD=mqtt -e POSTGRES_DB=telemetry \
  timescale/timescaledb:latest-pg16

Sık sorguladığınız birkaç tipli sütunu ve yeni alanların aktarımı hiç bozmaması için payload'ın tamamını jsonb olarak saklayın:

CREATE TABLE mqtt_messages (
  time       timestamptz NOT NULL,
  topic      text        NOT NULL,
  device_id  text        NOT NULL,
  payload    jsonb       NOT NULL
);

-- Yalnızca TimescaleDB (2.13+ sözdizimi); düz PostgreSQL'de atlayın
SELECT create_hypertable('mqtt_messages', by_range('time'));

CREATE INDEX ON mqtt_messages (device_id, time DESC);

Düz PostgreSQL'de (device_id, time DESC) indeksi, orta hacimde "cihaz başına son değerler" ve zaman aralığı sorguları için yeterlidir; tablo büyüdüğünde aylık native bölümleme ekleyin. Eski Timescale sürümlerinde create_hypertable('mqtt_messages', 'time') kullanın.

Toplu yazıcı

Bağımlılıkları pip install "paho-mqtt>=2" "psycopg[binary]>=3.1" ile kurun ve çalıştırın:

import json, os, queue, threading, time
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties
import psycopg
from psycopg.types.json import Jsonb

BROKER = os.environ.get("MQTT_HOST", "broker.connectmq.com")
TOPIC = "testmqtt/pg/+/telemetry"
DSN = os.environ.get("PG_DSN", "postgresql://mqtt:mqtt@localhost:5432/telemetry")
BATCH_SIZE, FLUSH_SECS = 500, 1.0
INSERT = "INSERT INTO mqtt_messages (time, topic, device_id, payload) VALUES (%s, %s, %s, %s)"

rows: queue.Queue = queue.Queue(maxsize=50_000)
stop = threading.Event()


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"connect failed: {reason_code}")
        return
    print(f"connected, session present: {flags.session_present}")
    client.subscribe(TOPIC, qos=1)


def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload)
    except (UnicodeDecodeError, json.JSONDecodeError):
        print(f"skipping non-JSON payload on {msg.topic}")
        return
    ts = data.get("ts") if isinstance(data, dict) else None
    when = (datetime.fromtimestamp(ts / 1000, timezone.utc)
            if isinstance(ts, (int, float)) else datetime.now(timezone.utc))
    device_id = msg.topic.split("/")[2]
    rows.put((when, msg.topic, device_id, Jsonb(data)))   # doluysa bekler = backpressure


def writer():
    conn, batch = None, []
    while not (stop.is_set() and rows.empty() and not batch):
        deadline = time.monotonic() + FLUSH_SECS
        while len(batch) < BATCH_SIZE and (left := deadline - time.monotonic()) > 0:
            try:
                batch.append(rows.get(timeout=left))
            except queue.Empty:
                break
        if not batch:
            continue
        try:
            if conn is None or conn.closed:
                conn = psycopg.connect(DSN)
            with conn.cursor() as cur:
                cur.executemany(INSERT, batch)
            conn.commit()
            print(f"wrote {len(batch)} rows")
            batch.clear()
        except psycopg.OperationalError as e:      # veritabanı yok: toplu işi tut, tekrar dene
            print(f"database unavailable, retrying: {e}")
            conn = None
            time.sleep(2)
        except psycopg.Error as e:                 # hatalı veri: bu toplu işi at
            print(f"insert failed, dropping {len(batch)} rows: {e}")
            conn.rollback()
            batch.clear()


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id="pg-writer-01", protocol=mqtt.MQTTv5)
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

props = Properties(PacketTypes.CONNECT)
props.SessionExpiryInterval = 3600          # broker oturumu çevrimdışıyken 1 saat tutar
client.connect(BROKER, 8883, clean_start=False, properties=props)

worker = threading.Thread(target=writer)
worker.start()
client.loop_start()
try:
    while not stop.wait(1):
        pass
except KeyboardInterrupt:
    pass
client.disconnect()        # önce mesaj almayı durdur...
client.loop_stop()
stop.set()                 # ...sonra yazıcının kuyruğu boşaltmasına izin ver
worker.join()

Anlamaya değer birkaç ayrıntı:

• psycopg 3'teki executemany toplu işi pipeline modunda verimli biçimde gönderir; mesaj başına bir execute ve commit'ten çok daha hızlıdır. Çok yüksek hızlar için cur.copy("COPY mqtt_messages (time, topic, device_id, payload) FROM STDIN") ve write_row() kullanın.
• Jsonb(data) ayrıştırılmış dict'i jsonb türüne uyarlar. Ham payload'ı saklamak, ileride tipli sütunlar ekleyip bunları SQL ile doldurmanıza olanak tanır.
• Sınırlı kuyruk güvenlik vananızdır: Postgres yavaşlarsa rows.put() MQTT thread'ini bekletir, istemci soketi okumayı bırakır ve süreciniz belleği tüketmek yerine mesajları broker tutar.
• Kapanış sırası önemlidir: önce MQTT bağlantısını kesin, sonra kuyruğu boşaltıp flush edin; böylece alınan hiçbir şey yazılmadan kalmaz.

Yeniden başlatmalar ve teslim garantileri

clean_start=False, sabit bir client ID ve bir oturum süresiyle broker, tüketici kapalıyken QoS 1 mesajlarını kuyruğa alır ve yeniden bağlanınca teslim eder (logda session present: True). Ayrıntılar için kalıcı oturumlar yazısına bakın. broker.connectmq.com dahil paylaşımlı herkese açık test endpoint'leri çevrimdışı oturum tutmaz; bu davranışı kendi özel broker'ınızda test edin.

Bir boşluk kalıyor: paho, bir QoS 1 mesajını on_message döner dönmez, yani satır commit edilmeden önce onaylar. Süreç çökerse kuyrukta bekleyen satırlar kaybolur. Bu önemliyse istemciyi manual_ack=True ile oluşturun, her satırla birlikte msg.mid ve msg.qos değerlerini taşıyın ve conn.commit() sonrasında her mesaj için alınma sırasıyla client.ack(mid, qos) çağırın. Broker'ın aynı anda yalnızca sınırlı sayıda onaylanmamış mesaj gönderdiğini unutmayın (Receive Maximum veya broker'ın inflight limiti); FLUSH_SECS değerini kısa tutun, yoksa throughput bu pencereyle sınırlanır.

At-least-once teslim, yeniden bağlantılardan sonra tekrarlanan mesajların mümkün olduğu anlamına gelir. Sizin için önemliyse cihaz başına bir sıra numarası yayınlayın, (device_id, time, (payload->>'seq')) gibi benzersiz bir indeks ekleyin ve ON CONFLICT DO NOTHING ile ekleyin. Hypertable üzerinde benzersiz indeksler zaman sütununu içermelidir. Nedenleri için tekrarlanan mesajlar yazısına bakın.

Pipeline'ı test edin

1. Veritabanını başlatın ve schema.sql dosyasını psql postgresql://mqtt:mqtt@localhost:5432/telemetry -f schema.sql ile uygulayın.
2. python mqtt_to_pg.py komutunu çalıştırın.
3. Test verisi yayınlayın; online MQTT istemcisinden ya da bir shell döngüsüyle:

for i in $(seq 1 20); do
  mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
    -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 -t testmqtt/pg/sensor-0$((i % 3))/telemetry \
    -m "{\"ts\":$(date +%s000),\"seq\":$i,\"temperature\":$((20 + i % 5)).5}"
done

1. Satırları kontrol edin:

SELECT device_id, count(*), max(time) FROM mqtt_messages GROUP BY device_id;

SELECT time_bucket('1 minute', time) AS minute, device_id,
       avg((payload->>'temperature')::float) AS avg_temp
FROM mqtt_messages
WHERE time > now() - interval '1 hour'
GROUP BY 1, 2 ORDER BY 1;

time_bucket bir TimescaleDB fonksiyonudur; düz PostgreSQL'de date_trunc('minute', time) kullanın. Yeniden başlatmaları test etmek için tüketiciyi durdurun, on mesaj daha yayınlayın, yeniden başlatın ve sayının on arttığını doğrulayın (oturum tutan bir broker'da).

Test ölçümlerini tarayıcınızdan yayınlayın

Saklama süresi ve ölçekleme

• Sıkıştırma ve saklama (TimescaleDB):

ALTER TABLE mqtt_messages SET (timescaledb.compress, timescaledb.compress_segmentby = 'device_id');
SELECT add_compression_policy('mqtt_messages', INTERVAL '7 days');
SELECT add_retention_policy('mqtt_messages', INTERVAL '90 days');

• Daha fazla tüketici: her biri kendi client ID'sine sahip birkaç yazıcıyı $share/pg-writers/testmqtt/pg/+/telemetry gibi bir paylaşımlı abonelikle çalıştırın; broker mesajları aralarında dağıtır. Ayrıntılar paylaşımlı abonelikler yazısında.
• Dashboard'lar: Grafana'nın PostgreSQL data source'u bu tabloyu doğrudan okur; üstte canlı bir panel için onu Grafana MQTT data source ile birleştirin.

İyi bir MQTT'den SQL'e pipeline bilinçli olarak sıkıcıdır: hızlı ayrıştırır, sınırlı kuyruğa alır, toplu yazar ve yeniden başlatmaları broker'ın kalıcı oturumuna bırakır. Önce birkaç yayınlanmış mesajla, sonra bir yeniden başlatmayla test edin; kötü bir günde verinize tam olarak ne olacağını bilirsiniz.

Sıkça sorulan sorular