gmqtt ile Python'da MQTT 5

gmqtt, en baştan MQTT 5 etrafında tasarlanmış, Python için bir asyncio MQTT istemcisidir. Her MQTT 5 property'si — session expiry, message expiry, user property, response topic, correlation data, subscription identifier — sade bir keyword argümanıdır ve gelen property'ler bir sözlük (dict) olarak ulaşır. MQTT 5 özelliklerini Python'da elle Properties nesneleri kurmadan denemek istiyorsanız gmqtt en kısa yoldur.

Bu eğitimde MQTT 5 ile bağlanmayı, property'lerle publish etmeyi, abonelik seçeneklerini, request/response'u, reason code'ları okumayı ve hepsini gerçek bir broker'a karşı test etmeyi göreceksiniz.

Kurulum ve bağlantı

pip install gmqtt

gmqtt, asyncio event loop'u üzerinde çalışan callback'ler kullanır. Bağlanın, on_connect içinde abone olun ve programı bir asyncio.Event ile canlı tutun:

import asyncio
import os
from gmqtt import Client as MQTTClient
from gmqtt.mqtt.constants import MQTTv50

STOP = asyncio.Event()

def on_connect(client, flags, rc, properties):
    print("bağlandı, rc =", rc, "broker property'leri:", properties)
    client.subscribe("testmqtt/gmqtt/#", qos=1)

def on_message(client, topic, payload, qos, properties):
    print("mesaj", topic, payload.decode(), properties)
    return 0  # PUBACK için reason code

def on_disconnect(client, packet, exc=None):
    print("bağlantı kesildi", exc)

async def main():
    client = MQTTClient(
        "testmqtt-gmqtt-demo-01",
        session_expiry_interval=3600,   # bağlantı kesildikten sonra oturumu 1 saat tut
        receive_maximum=100,            # en fazla 100 onaylanmamış QoS 1/2 mesajı
    )
    client.on_connect = on_connect
    client.on_message = on_message
    client.on_disconnect = on_disconnect
    client.set_auth_credentials(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])

    await client.connect("broker.connectmq.com", 1883, keepalive=30, version=MQTTv50)
    await STOP.wait()
    await client.disconnect()

asyncio.run(main())

İki şey gmqtt'ye özgüdür. MQTTClient yapıcısına verilen keyword argümanları CONNECT property'lerine dönüşür — burada session_expiry_interval ve receive_maximum. on_connect'in properties argümanı ise broker'ın CONNACK'te geri gönderdiklerini taşır: broker'a bağlı olarak kendi receive maximum'u, topic alias maximum'u, maximum packet size'ı ya da atanmış bir client identifier. Bunu bir kez yazdırmak, broker'ınızın gerçekte neleri desteklediğini görmenin en hızlı yoludur.

TLS için ssl=True (veya bir ssl.SSLContext) ve 8883 portunu verin: await client.connect("broker.connectmq.com", 8883, ssl=True).

MQTT 5 property'leriyle publish

Publish property'leri client.publish()'in keyword argümanlarıdır:

import json, time

client.publish(
    "testmqtt/gmqtt/sensor-1/temperature",
    json.dumps({"value": 21.7, "t": time.time()}),
    qos=1,
    content_type="application/json",
    message_expiry_interval=60,                     # 60 sn içinde teslim edilmezse at
    user_property=("firmware", "2.4.1"),
)

Subscriber tarafında aynı değerler properties içinde görünür. Yapıya dikkat edin — değerler listedir, çünkü MQTT 5 bazı property'lerin birden fazla kez bulunmasına izin verir:

message testmqtt/gmqtt/sensor-1/temperature {"value": 21.7, "t": 1792223400.12}
  {'content_type': ['application/json'], 'message_expiry_interval': [60],
   'user_property': [('firmware', '2.4.1')], 'dup': 0, 'retain': 0}

Bunları savunmacı biçimde okuyun, örneğin ctype = properties.get("content_type", [None])[0]. Mesaj teslimattan önce bir kuyrukta beklediyse broker kalan süreyi gönderir; yani subscriber 60'tan küçük bir değer görebilir — message expiry'nin çalıştığını doğrulamanın pratik bir yolu. User property'lerin ne zaman doğru araç olduğunu ve ne zaman olmadığını MQTT user property'leri yazısında bulabilirsiniz.

Abonelik seçenekleri

MQTT 5, abonelik başına seçenekler ekler. gmqtt bunları Subscription nesneleriyle sunar:

from gmqtt import Subscription

client.subscribe(
    [
        Subscription("testmqtt/gmqtt/+/temperature", qos=1),
        Subscription("testmqtt/gmqtt/chat/#", qos=1, no_local=True),
        Subscription("testmqtt/gmqtt/config/#", qos=1, retain_handling_options=2),
    ],
    subscription_identifier=7,
)

Bir mesaj yukarıdaki abonelikle eşleştiğinde properties["subscription_identifier"] değeri [7] olur. Bu, on_message içindeki yönlendirmeyi topic string'lerini yeniden karşılaştırmadan ucuz ve açık hale getirir.

Response topic ve correlation data ile request/response

MQTT 5, request/response'u standartlaştırır: istekte bulunan bir response topic ve correlation data belirler; yanıtlayan cevabını oraya publish eder ve correlation data'yı geri kopyalar.

def on_connect(client, flags, rc, properties):
    client.subscribe("testmqtt/gmqtt/rpc/time", qos=1)

def on_message(client, topic, payload, qos, properties):
    reply_to = properties.get("response_topic", [None])[0]
    corr = properties.get("correlation_data", [None])[0]
    if reply_to:
        client.publish(reply_to, str(time.time()), qos=1, correlation_data=corr)
    return 0

import uuid

corr = uuid.uuid4().bytes
client.subscribe("testmqtt/gmqtt/rpc/replies/client-01", qos=1)
client.publish(
    "testmqtt/gmqtt/rpc/time", "",
    qos=1,
    response_topic="testmqtt/gmqtt/rpc/replies/client-01",
    correlation_data=corr,
)

İstekte bulunanın on_message'ında, yanıtı isteğiyle eşleştirmek için properties["correlation_data"][0] ile corr'u karşılaştırın. Zaman aşımları dahil kalıbın tamamı MQTT request/response yazısında.

Reason code'lar ve temiz bağlantı kesme

MQTT 5, eski tek baytlık dönüş kodlarının yerine CONNACK, SUBACK, PUBACK ve DISCONNECT'te ayrıntılı reason code'lar getirir. gmqtt'de bunları birkaç yerde görürsünüz:

• on_connect içindeki rc — 0 başarıdır; 134 (Bad User Name or Password) veya 135 (Not authorized) gibi değerler reddedilen bir bağlantıyı açıklar.
• on_subscribe(client, mid, qos, properties) içindeki qos demeti — bir topic için 128 veya üzeri değer, broker'ın o aboneliği genellikle bir ACL nedeniyle reddettiği anlamına gelir.
• on_message'ın dönüş değeri, geri gönderdiğiniz PUBACK/PUBREC'te reason code olarak kullanılır; 0 başarı demektir.

Herhangi bir sayıyı MQTT reason code sorgulama aracı ile arayın.

Last Will'in yine de gönderilmesini sağlayarak ayrılmak için (izlemeyi test ederken işe yarar) reason code 4 ile bağlantıyı kesin: await client.disconnect(reason_code=4). Will'in kendisi, isteğe bağlı bir will delay interval ile bir Message nesnesi olarak yapılandırılır:

from gmqtt import Client as MQTTClient, Message

will = Message("testmqtt/gmqtt/sensor-1/status", "offline", qos=1, retain=True, will_delay_interval=10)
client = MQTTClient("testmqtt-gmqtt-sensor-1", will_message=will)

Yeniden bağlanma ve session expiry

aiomqtt'nin aksine gmqtt otomatik olarak yeniden bağlanır. Kaç kez deneyeceğini ve ne kadar bekleyeceğini ayarlayabilirsiniz:

client.set_config({"reconnect_retries": 10, "reconnect_delay": 5})

Yeniden bağlanmayı session_expiry_interval ve sabit bir client ID ile birleştirin: siz yokken broker aboneliklerinizi saklar ve QoS 1 mesajlarını kuyruğa alır; CONNACK bayrakları da bir oturumun devam ettirilip ettirilmediğini söyler. Yeni bir bağlantıda on_connect içindeki flags session present bilgisini verir; bir kesintiden sonra 0 ise oturumun süresi dolmuştur ve yeniden abone olmanız gerekir (yukarıdaki gibi on_connect içinde abone olmak bunu halleder). Uygulamalı kontrol için session expiry testi yazısına bakın.

gmqtt istemcinizi test etmek

1. İlk örnekteki subscriber'ı çalıştırın.
2. Online MQTT istemcisini açın, MQTT 5 ile broker.connectmq.com'a bağlanın ve bir user property ile message expiry ekleyerek testmqtt/gmqtt/sensor-1/temperature topic'ine publish edin.
3. Terminalinizdeki property sözlüğünün tam olarak gönderdiğinizi içerdiğini doğrulayın.
4. Subscriber'ı durdurun, QoS 1 bir mesaj publish edin ve session expiry penceresi içinde yeniden başlatın — mesaj yeniden bağlanınca gelmelidir.

gmqtt istemcinize MQTT 5 mesajları gönderin

gmqtt ve diğer Python istemcileri

gmqtt; MQTT 5 meta verileri tasarımınızın merkezindeyse — request/response, content type'lar, süre sonu — ve kodunuz zaten asenkronsa iyi bir seçimdir. paho-mqtt'ye göre daha küçük bir ekip tarafından geliştirildiği için son sürümleri kontrol edin ve test ettiğiniz sürümü sabitleyin. Genel tablo için MQTT 5 ve 3.1.1 karşılaştırmasına bakın.

gmqtt, MQTT 5'i sıradan Python gibi hissettirir: property'ler argümandır, sonuçlar sözlüktür ve yeniden bağlanma kendiliğinden olur. CONNACK property'lerini bir kez yazdırın, property değerlerini liste olarak okuyun ve üretimde güvenmeden önce her özelliği ikinci bir istemciyle doğrulayın.

Sıkça sorulan sorular