Pratikte Önemli MQTT 5 Özellikleri: Reason Code'lar, Shared Subscription'lar, Expiry ve Daha Fazlası

MQTT 5, MQTT standardının güncel sürümüdür ve ekiplerin MQTT 3.1.1'i üretimde çalıştırırken karşılaştığı birçok sıkıntıyı giderir. Bu rehberde hangi MQTT 5 özelliklerinin günlük mühendislik işini gerçekten değiştirdiğini öğrenecek, bunları 3.1.1 ile yan yana görecek ve Mosquitto CLI araçları ile Python kullanarak broker.connectmq.com public broker'ı üzerinde kendiniz deneyeceksiniz.

Protokole yeniyseniz önce MQTT broker'ın ne olduğunu okuyun ve buraya dönün.

MQTT 5 Neden Var?

MQTT 3.1.1 küçük, kararlı ve her yerdedir. Ancak önemli soruları yanıtsız bırakır. Bağlantım neden reddedildi? Aboneliğim neden başarısız oldu? Tek bir topic'teki mesajları birden fazla worker'a nasıl dağıtırım? Broker oturumumu veya eski bir komutu ne kadar süre tutmalı?

3.1.1'de yanıtlar genellikle "broker soketi kapattı" ya da "bunu payload'a kendin yaz" şeklindeydi. 2019'da OASIS tarafından standartlaştırılan MQTT 5, aynı publish/subscribe modelini korurken protokole açık geri bildirim, meta veri ve limitler ekler. Topic'leriniz, QoS seviyeleriniz ve retained message'larınız eskisi gibi çalışır. Tam spesifikasyona OASIS üzerinden ulaşabilirsiniz.

Bir Bakışta MQTT 3.1.1 ve MQTT 5

Reason Code'lar: Bir Şeyin Neden Başarısız Olduğunu Bilmek

Operasyon açısından en kullanışlı MQTT 5 değişikliği reason code'lardır. CONNACK, PUBACK, PUBREC, SUBACK, UNSUBACK ve DISCONNECT paketlerinin tümü bir baytlık bir kod taşır ve sunucu bununla birlikte insan tarafından okunabilir bir reason string gönderebilir.

0x80'in altındaki kodlar başarıyı, 0x80 ve üzeri ise başarısızlığı belirtir. Sahadaki cihazlar için bu çok önemlidir: firmware "yanlış parola" durumuna (dur ve alarm ver) "sunucu meşgul" durumundan (geri çekil ve yeniden dene) farklı tepki verebilir. ConnectMQ, 0x97 Quota exceeded ve 0x89 Server busy gibi reason code'lar döndürür; böylece client'larınız bir kota sorununu geçici bir aşırı yükten ayırt edebilir.

Session Expiry ve Message Expiry

Session Expiry Interval

3.1.1'de kalıcı bir oturum (clean session = false) client geri dönene kadar, muhtemelen sonsuza dek yaşar. MQTT 5 bunun yerine iki ayar getirir: Clean Start (sıfırdan başla veya devam et) ve Session Expiry Interval (broker'ın oturumu bağlantı kesildikten sonra kaç saniye tutacağı). On dakika uyuyan bir sensör 15 dakikalık bir oturum isteyebilir ve uyandığında kuyruktaki QoS 1 mesajlarını yine alır; terk edilmiş oturumlar ise kendiliğinden kaybolur.

ConnectMQ'da session expiry plan başına sınırlandırılmıştır; public test broker'ında maksimum 30 dakikadır.

Message Expiry Interval

Publisher her mesaja saniye cinsinden bir ömür ekleyebilir. Bu süre dolduğunda mesaj hâlâ bir kuyrukta bekliyorsa veya retained olarak saklanıyorsa broker onu siler. Bu, bir saat geç çalıştırılması tehlikeli olan "vanayı aç" gibi komutlar için tam olarak istediğiniz şeydir.

mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -i demo-a1b2-pub \
  -t "connectmq/demo/a1b2/commands/valve1" -m "open" -q 1 \
  -D publish message-expiry-interval 60

Shared Subscription'lar: Yerleşik Yük Dengeleme

Normal bir abonelikte her subscriber her mesajı alır. Tek bir backend örneği binlerce cihaza yetişemediğinde bu bir sorundur. MQTT 5 shared subscription'ları birden fazla client'ın bir gruba katılmasını sağlar; broker her mesajı grubun yalnızca bir üyesine iletir.

Söz dizimi $share/{group}/{filter} şeklindedir. Publisher'lar hiçbir şeyi değiştirmez: yine düz topic'e yayın yaparlar.

Deneyin: Tek Bir Topic'i Paylaşan İki Worker

İki terminal açın ve aynı workers grubunda iki tüketici başlatın. Public alan herkes tarafından okunabildiği için a1b2 yerine kendi rastgele ID'nizi yazın.

# Terminal 1
mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 -i demo-a1b2-w1 \
  -t '$share/workers/connectmq/demo/a1b2/jobs' -v

# Terminal 2
mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 -i demo-a1b2-w2 \
  -t '$share/workers/connectmq/demo/a1b2/jobs' -v

Şimdi üçüncü bir terminalden altı iş yayınlayın:

for i in 1 2 3 4 5 6; do
  mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -i demo-a1b2-pub \
    -t "connectmq/demo/a1b2/jobs" -m "job-$i" -q 1
done

Bunu broker.connectmq.com üzerinde çalıştırdığımızda worker 1 job-1, job-3, job-5 ve worker 2 job-2, job-4, job-6 aldı. Her iş bir kez işlendi; üçüncü bir worker eklemek ise publisher'lara dokunmadan yükü daha da dağıtır. Bash'te kabuğun $share ifadesini genişletmemesi için topic'in etrafındaki tek tırnaklar önemlidir.

Tipik kullanımlar; yatay ölçeklenen veri alım (ingestion) servisleri, komut işleyiciler ve birden fazla container replikası olarak çalıştırdığınız her tüketicidir.

Topic Alias'lar: Kısıtlı Bağlantılarda Daha Küçük Paketler

Endüstriyel topic'ler uzun olabilir; örneğin acme/izmir/machining/line1/cnc1/telemetry. Bu dizeyi her PUBLISH'te göndermek hücresel veya uydu bağlantılarında bant genişliğini israf eder. Topic alias ile client tam topic'i bir kez bir tamsayı alias ile birlikte gönderir, ardından yalnızca alias numarasını gönderir. Her taraf kaç alias kabul ettiğini (Topic Alias Maximum) CONNECT ve CONNACK içinde bildirir. Çoğu client kütüphanesi, etkinleştirdiğinizde alias'ları sizin için yönetir.

User Property'ler: Payload'a Dokunmadan Meta Veri

User property'ler bir pakete eklenen keyfi UTF-8 anahtar/değer çiftleridir. Bunları içerik türü ipuçları, şema sürümleri, izleme (trace) ID'leri veya göndericinin firmware sürümü için kullanın; payload ise temiz bir sensör ölçümü olarak kalır.

mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -i demo-a1b2-pub \
  -t "connectmq/demo/a1b2/sensors/t1" -m '{"temperature":22.4}' \
  -D publish user-property schema v2 -D publish user-property fw 1.4.0

Bir subscriber bunları mosquitto_sub -V mqttv5 -F "%t %p %P" ile yazdırabilir.

Response Topic ve Correlation Data ile İstek/Yanıt

Birçok IoT akışı aslında RPC çağrısıdır: "yapılandırmanı oku", "yeniden başlat", "bana son 10 değeri ver". MQTT 5 bunu standartlaştırır:

1. İstekte bulunan, kendi yanıt topic'ine abone olur.
2. İsteği, o yanıt topic'ini gösteren bir Response Topic property'si ve isteği tanımlayan Correlation Data ile yayınlar.
3. Yanıtlayan, response topic üzerinden yanıt verir ve correlation data'yı geri kopyalar.

Response topic subscriber'lara iletildiği için yanıtlayan tarafın sabit kodlanmış herhangi bir yönlendirme bilgisine ihtiyacı yoktur. Property'lerin geldiğini şununla görebilirsiniz:

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 -i demo-a1b2-s \
  -t "connectmq/demo/a1b2/#" -F "%t %p | response=%R props=%P"

Akış Kontrolü ve Sunucunun Bildirdiği Limitler

Receive Maximum

Receive Maximum, aynı anda kaç QoS 1 ve QoS 2 mesajının yolda (gönderilmiş ancak henüz onaylanmamış) olabileceğini sınırlar. Client limitini sunucuya CONNECT'te, sunucu da kendi limitini client'a CONNACK'te bildirir. Küçük bir mikrodenetleyici hiçbir zaman mesaj selinde boğulmamak için düşük bir değer ayarlayabilir; broker da onaylayabileceğinden hızlı yayın yapan bir client'a karşı kendini koruyabilir. Onaylar hakkında daha fazlası için MQTT QoS seviyeleri yazısına bakın.

CONNACK'teki limitler

3.1.1'de bir client limitleri başarısız olarak keşfederdi. MQTT 5'te sunucu bunları Maximum QoS, Retain Available, Wildcard Subscription Available, Shared Subscription Available, Receive Maximum ve Topic Alias Maximum dahil CONNACK property'lerinde bildirir. İyi yazılmış bir client bunları bir kez okur ve kendini uyarlar; örneğin mevcut planda QoS 2 sunulmuyorsa QoS 1'e iner.

Paho 2.x ile Python'da MQTT 5

Aşağıdaki betik MQTT 5 ile bağlanır, CONNACK'te bildirilen limitleri yazdırır ve message expiry, user property'ler, response topic ve correlation data kullanarak eksiksiz bir istek/yanıt döngüsü çalıştırır.

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

BROKER = "broker.connectmq.com"
PREFIX = "connectmq/demo/a1b2"          # kendi rastgele ID'nizi kullanın
REQ_TOPIC = f"{PREFIX}/rpc/request"
RESP_TOPIC = f"{PREFIX}/rpc/response"


def on_connect(client, userdata, flags, reason_code, properties):
    print("CONNACK:", reason_code)
    for name in ("MaximumQoS", "RetainAvailable", "ReceiveMaximum",
                 "TopicAliasMaximum", "SharedSubscriptionAvailable"):
        if hasattr(properties, name):
            print(f"  {name} = {getattr(properties, name)}")
    client.subscribe(REQ_TOPIC, qos=1)
    client.subscribe(RESP_TOPIC, qos=1)


def on_message(client, userdata, msg):
    props = msg.properties
    if msg.topic == REQ_TOPIC:
        # Yanıtlayan taraf: istenen topic'te yanıt ver, correlation data'yı geri gönder
        reply = Properties(PacketTypes.PUBLISH)
        reply.CorrelationData = props.CorrelationData
        client.publish(props.ResponseTopic, b'{"uptime": 3600}', qos=1,
                       properties=reply)
    elif msg.topic == RESP_TOPIC:
        print("Response:", msg.payload.decode(),
              "correlation:", props.CorrelationData.decode())


connect_props = Properties(PacketTypes.CONNECT)
connect_props.SessionExpiryInterval = 600   # oturumu 10 dakika tut
connect_props.ReceiveMaximum = 10           # en fazla 10 onaylanmamış gelen mesaj

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id="demo-a1b2-py", protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, clean_start=True, properties=connect_props)
client.loop_start()
time.sleep(2)

request = Properties(PacketTypes.PUBLISH)
request.ResponseTopic = RESP_TOPIC
request.CorrelationData = uuid.uuid4().hex.encode()
request.MessageExpiryInterval = 30
request.UserProperty = [("schema", "v1"), ("app", "blog-demo")]
client.publish(REQ_TOPIC, b'{"cmd": "status"}', qos=1, properties=request)

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

Gerçek bir sistemde istekte bulunan ve yanıtlayan farklı client'lardır, ancak kod aynı kalır. Üretim için portaldan aldığınız YOUR_USERNAME, YOUR_PASSWORD ve YOUR_CLIENT_ID ile kendi broker'ınıza, ideal olarak 8883 portunda TLS üzerinden bağlanın (MQTT güvenliği en iyi uygulamaları yazısına bakın).

ConnectMQ'da MQTT 5 Desteği

ConnectMQ broker'ları TCP, TLS ve WebSocket üzerinden hem MQTT 3.1.1 hem de MQTT 5.0 konuşur; böylece eski ve yeni cihazlar aynı broker'ı paylaşabilir. Şu anda uygulanan MQTT 5 özellikleri şunlardır:

ConnectMQ VS Code eklentisi CONNACK, SUBACK ve PUBACK reason code'larını doğrudan gösterir; bu da bir aboneliğin neden reddedildiğini ayıklarken işinize yarar.

ConnectMQ ile Başlayın

Bu yazıdaki her örneği hesap olmadan broker.connectmq.com üzerinde deneyebilirsiniz. Kimlik bilgileri olan, size ait izole bir broker'a hazır olduğunuzda app.connectmq.com adresinden kaydolun: Serverless planı ücretsizdir, kredi kartı gerektirmez ve dakikalar içinde size bir broker verir. Seçenekleri yönetilen MQTT broker nedir yazısında karşılaştırın.

Sık Sorulan Sorular

MQTT 5, MQTT 3.1.1 ile geriye dönük uyumlu mu?

Kablo üzerindeki format farklıdır; bu yüzden bir client bağlanırken tek bir sürüm seçer. ConnectMQ gibi her ikisini de destekleyen bir broker, MQTT 3.1.1 ve MQTT 5 client'larının aynı topic'ler üzerinde mesaj alışverişi yapmasına olanak tanır.

Shared subscription'lar için MQTT 5 gerekli mi?

Shared subscription'lar MQTT 5 standardında tanımlanmıştır. Yük dengeli teslimat için MQTT 5 ile bağlanın (örneğin Mosquitto araçlarında -V mqttv5) ve $share/{group}/{filter} filtresine abone olun.

Session expiry ile message expiry arasındaki fark nedir?

Session expiry, broker'ın bağlantısı kesilmiş bir client'ın oturumunu ve kuyruktaki mesajlarını ne kadar süre tutacağını belirler. Message expiry ise tek bir mesajın broker tarafından silinmeden önce ne kadar süre geçerli kalacağını belirler.

İlk olarak hangi MQTT 5 özelliğini benimsemeliyim?

Reason code'larla başlayın; çünkü hataları sıfır maliyetle görünür kılarlar. Ardından uyku moduna giren cihazlar için session expiry'yi, ölçeklenmesi gereken backend servisleri için de shared subscription'ları ekleyin.