Python'da MQTT üzerinden JSON gönderme

MQTT JSON'ın ne olduğunu bilmez: bir payload yalnızca byte'lardan oluşur. Python'da MQTT üzerinden JSON göndermek için nesnenizi publish()'ten önce json.dumps() ile serileştirin ve on_message içinde json.loads(msg.payload) çağırın. Bu rehberin geri kalanı bunu güvenle yapmakla ilgili: geleni doğrulamak, bozuk mesajlara dayanmak, tarihleri ele almak ve payload'ı küçük tutmak.

Tüm örnekler paho-mqtt 2.x ve test broker'ı broker.connectmq.com kullanır.

En basit gidiş-dönüş

import json
import time

import paho.mqtt.client as mqtt

TOPIC = "testmqtt/json-demo/sensor-42/telemetry"


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(TOPIC, qos=1)


def on_message(client, userdata, msg):
    data = json.loads(msg.payload)  # bytes girer, dict çıkar
    print(msg.topic, data["temperature"], type(data["temperature"]))


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883)
client.loop_start()
time.sleep(1)

reading = {"device_id": "sensor-42", "temperature": 21.5, "humidity": 48.2}
info = client.publish(TOPIC, json.dumps(reading), qos=1)
info.wait_for_publish()

time.sleep(1)
client.loop_stop()
client.disconnect()

Üç gerçek, yeni başlayanların hatalarının çoğunu açıklar:

• json.dumps() bir str döndürür; paho string'leri göndermeden önce UTF-8 olarak kodlar. Dict'in kendisini vermek TypeError: payload must be a string, bytearray, int, float or None. hatasıyla sonuçlanır.
• msg.payload her zaman bytes'tır. json.loads() UTF-8 bytes kabul eder, bu yüzden önce .decode() gerekmez.
• JSON'da tuple, set, datetime veya Decimal yoktur. Tuple'lar liste olarak geri gelir; diğerleri TypeError: Object of type datetime is not JSON serializable hatası verir.

Mesajları online MQTT istemcisinde testmqtt/json-demo/# topic'ine abone olarak izleyebilirsiniz. Payload'ı metin olarak gösterir; böylece kodunuzun hatta tam olarak ne koyduğunu görürsünüz.

Bozuk bir payload subscriber'ınızı asla öldürmemeli

Bir topic'te publish yetkisi olan herkes size her şeyi gönderebilir: düz bir string, yarım kalmış bir mesaj, eski bir cihazdan Latin-1 metin. json.loads() on_message içinde exception fırlatırsa bu yukarı taşınır ve ağ döngünüz durur. Önemli olan iki hatayı yakalayın:

import json
import logging

log = logging.getLogger("mqtt")


def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload)
    except UnicodeDecodeError:
        log.warning("payload on %s is not UTF-8: %r", msg.topic, msg.payload[:50])
        return
    except json.JSONDecodeError as exc:
        log.warning("invalid JSON on %s: %s (%r)", msg.topic, exc, msg.payload[:50])
        return

    if not isinstance(data, dict):
        log.warning("expected an object on %s, got %s", msg.topic, type(data).__name__)
        return

    handle_reading(msg.topic, data)

isinstance kontrolü göründüğünden daha önemlidir: 42, "ok" ve null geçerli JSON'dır ve data["temperature"] yapan kod bunlarda TypeError ile çöker. Loglarda göreceğiniz tipik mesajlar, örneğin boş veya JSON olmayan bir payload için Expecting value: line 1 column 1 (char 0), MQTT JSON parse hataları yazısında açıklanıyor.

Yapıyı pydantic ile doğrulayın

Sözdizimi hatalarını yakalamak yetmez: {"temperature": "hot"} sorunsuz parse edilir. Bir şema sözleşmeyi açık hale getirir ve pydantic v2 bytes'ı tek adımda parse edip doğrulayabilir:

from datetime import datetime

from pydantic import BaseModel, Field, ValidationError


class Reading(BaseModel):
    device_id: str = Field(min_length=1, max_length=64)
    ts: datetime
    temperature: float = Field(ge=-50, le=150)
    humidity: float | None = Field(default=None, ge=0, le=100)


def on_message(client, userdata, msg):
    try:
        reading = Reading.model_validate_json(msg.payload)
    except ValidationError as exc:
        log.warning("rejected message on %s: %s", msg.topic, exc.errors(include_url=False))
        return
    store(reading)  # reading.ts gerçek bir datetime, reading.temperature bir float

Publisher tarafında aynı model, datetime dahil kendini serileştirir:

from datetime import datetime, timezone

reading = Reading(device_id="sensor-42", ts=datetime.now(timezone.utc), temperature=21.5)
client.publish("testmqtt/json-demo/sensor-42/telemetry", reading.model_dump_json(), qos=1)

Publisher ve subscriber arasında tek bir model modülü paylaşmak, contract testing'in en ucuz biçimidir: biri bir alanın adını değiştirirse test paketi cihazlardan önce hata verir.

Tarihler, ondalıklar ve diğer zorlu tipler

pydantic kullanmıyorsanız json.dumps()'a bilmediği tipleri nasıl işleyeceğini söyleyin:

import json
from datetime import datetime
from decimal import Decimal


def default(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    if isinstance(obj, Decimal):
        return str(obj)
    if isinstance(obj, set):
        return sorted(obj)
    raise TypeError(f"Cannot serialize {type(obj).__name__}")


payload = json.dumps(data, default=default, allow_nan=False)

NaN satırı gerçek bir tuzaktır: Python hiç şikâyet etmeden {"t": NaN} üretir, ama JavaScript'in JSON.parse'ı, Go ve katı parser'ların çoğu bunu reddeder. allow_nan=False bunu publisher tarafında, düzeltmesi kolay olan yerde bir ValueError'a çevirir.

Kompakt kodlama

Hücresel bağlantılarda ve pille çalışan cihazlarda her byte önemlidir ve MQTT payload'ın bedelini her atlamada öder. İşte aynı ölçümün dört farklı kodlaması (boyutlar Python 3 ile ölçüldü):

İlk üçü tam anahtarları kullanır: device_id, timestamp (mikrosaniye ve offset içeren bir ISO string), temperature, humidity, battery ve status. Sonuncusu {"id":"sensor-42","ts":1792229400,"t":21.46,"h":48.2,"b":0.87}. Farkı yaratan değişiklikler:

• Kompakt ayırıcılar , ve : sonrasındaki boşlukları kaldırır.
• Kısa anahtarlar küçük mesajlarda en büyük kazançtır; bunları şemanızda belgeleyin.
• ISO string'leri yerine Unix timestamp'leri her biri yaklaşık 20 byte kazandırır.
• Float'ları sensörünüzün gerçekten sahip olduğu hassasiyete yuvarlayın. 21.456789 bilgi değil gürültü taşır.
• ensure_ascii=False, ASCII olmayan metni ü kaçışları (2 yerine 6 byte) yerine UTF-8 olarak ("Zürich") korur.

Hız ve kolaylık için orjson doğrudan bir alternatiftir: varsayılan olarak kompakt çıktı üretir, bytes döndürür (paho bunu doğrudan kabul eder) ve datetime'ı yerel olarak serileştirir.

import orjson

client.publish(TOPIC, orjson.dumps(reading_dict), qos=1)

def on_message(client, userdata, msg):
    data = orjson.loads(msg.payload)  # raises orjson.JSONDecodeError (a ValueError subclass)

Topic ve başlıklar dahil tam paket boyutunu MQTT paket boyutu hesaplayıcısı ile kontrol edin. Kompakt JSON hâlâ büyükse sıradaki adım ikili formatlardır; bkz. MQTT üzerinden CBOR ve MessagePack ve Protobuf payload'ları.

Subscriber'lara bunun JSON olduğunu söyleyin (MQTT 5)

MQTT 5, publisher'ın payload'ı etiketlemesine izin verir; böylece genel amaçlı araçlar ve diğer servisler tahmin yürütmek zorunda kalmaz:

import json

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
client.connect("broker.connectmq.com", 1883)
client.loop_start()

props = Properties(PacketTypes.PUBLISH)
props.ContentType = "application/json"
props.PayloadFormatIndicator = 1  # 1 = UTF-8 metin

info = client.publish("testmqtt/json-demo/sensor-42/telemetry",
                      json.dumps({"temperature": 21.5}), qos=1, properties=props)
info.wait_for_publish()

Alıcı tarafta msg.properties.ContentType değerini okuyun (yalnızca publisher ayarladıysa bulunduğu için önce hasattr ile kontrol edin). Payload format indicator 1 olduğunda bir MQTT 5 broker'ı geçerli UTF-8 olmayan payload'ları reddedebilir; ayrıntılar MQTT UTF-8 kodlama hataları yazısında.

JSON işlemenizi test edin

Bir subscriber'a güvenmeden önce ona işleri bozan mesajları gönderin. mosquitto_pub ile (veya online istemciyle):

T=testmqtt/json-demo/sensor-42/telemetry
H=broker.connectmq.com
mosquitto_pub -h $H -t $T -m '{"device_id":"sensor-42","ts":"2026-10-17T09:30:00Z","temperature":21.5}'
mosquitto_pub -h $H -t $T -m 'hello'                          # JSON değil
mosquitto_pub -h $H -t $T -n                                 # boş payload
mosquitto_pub -h $H -t $T -m '[1,2,3]'                        # geçerli JSON, yanlış yapı
mosquitto_pub -h $H -t $T -m '{"temperature":"hot"}'          # yanlış tip
printf '{"city":"Z\xfcrich"}' | mosquitto_pub -h $H -t $T -s  # Latin-1, UTF-8 değil

Subscriber ilk mesajı kaydetmeli, diğerlerinin her biri için açık bir uyarı loglamalı ve bağlı kalmalıdır. Aynı durumları, on_message fonksiyonunuzu sahte bir mesaj nesnesiyle doğrudan çağıran bir unit test'e koyun; böylece regresyonları broker olmadan yakalarsınız.

JSON payload'larını online istemcide inceleyin

Payload'a güvenilmeyen byte'lar olarak davrandığınızda MQTT üzerinden JSON basittir: açıkça serileştirin, savunmacı parse edin, bir şemaya göre doğrulayın ve kompakt tutun. Zarflar, timestamp'ler ve topic yapısı gibi daha büyük tasarım soruları için MQTT payload tasarımı en iyi uygulamaları ile devam edin.

Sıkça sorulan sorular