paho-mqtt ile Python’da MQTT: Publish ve Subscribe Eğitimi

paho-mqtt, Python için standart MQTT client kütüphanesidir. MQTT 3.1, 3.1.1 ve 5’i, düz TCP’yi, TLS’i ve WebSocket’leri destekler; Raspberry Pi’dan bulut sunucusuna kadar her yerde çalışır. Bu eğitim güncel paho-mqtt 2.x API’sini kullanır — 1.x için yazılmış eski blog yazılarındaki kodu olduğu gibi kopyalarsanız hata alırsınız, bu yüzden farklar yol boyunca belirtiliyor.

paho-mqtt kurulumu

python -m pip install "paho-mqtt>=2.0"

Sürümünüzü python -c "import paho.mqtt; print(paho.mqtt.__version__)" ile kontrol edin. Aşağıdaki tüm örnekler 2.x’i ve broker.hivemq.com adresindeki public HiveMQ sandbox’ını hedefler; böylece bir broker kurmadan çalıştırabilirsiniz. Kendi broker’ınızı çalıştırmayı tercih ederseniz Mosquitto kurulumu yazısına bakın.

paho-mqtt 2.x’te neler değişti

En büyük değişiklik callback API sürümüdür. mqtt.Client()’ın ilk argümanı artık bir CallbackAPIVersion’dır ve VERSION2 ile tüm callback’ler MQTT 5 tarzı reason code’lar ve property’ler alır — MQTT 3.1.1 ile bağlandığınızda bile.

reason_code bir ReasonCode nesnesidir: rc == 0 karşılaştırması yerine reason_code.is_failure’ı kontrol edin. MQTT reason code sorgulama aracımız her değerin ne anlama geldiğini açıklar.

Eksiksiz bir subscriber

Aşağıdaki subscriber bağlanır, on_connect içinde abone olur ve gelen her mesajı yazdırır. connect()’ten sonra değil de on_connect içinde abone olmak önemlidir: bağlantı koptuğunda ve paho yeniden bağlandığında callback tekrar çalışır ve abonelikleriniz geri yüklenir.

import paho.mqtt.client as mqtt

BROKER = "broker.hivemq.com"
TOPIC = "testmqtt/python-demo/#"


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"Connect failed: {reason_code}")
        return
    print("Connected, subscribing to", TOPIC)
    client.subscribe(TOPIC, qos=1)


def on_subscribe(client, userdata, mid, reason_code_list, properties):
    for rc in reason_code_list:
        if rc.is_failure:
            print(f"Subscription rejected: {rc}")
        else:
            print(f"Subscribed with granted QoS {rc.value}")


def on_message(client, userdata, msg):
    print(f"{msg.topic} [QoS {msg.qos}, retain={msg.retain}]: {msg.payload.decode()}")


def on_disconnect(client, userdata, disconnect_flags, reason_code, properties):
    print(f"Disconnected: {reason_code}")


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="")
client.on_connect = on_connect
client.on_subscribe = on_subscribe
client.on_message = on_message
client.on_disconnect = on_disconnect

client.connect(BROKER, 1883, keepalive=60)
client.loop_forever()

Boş bir client_id, paho’nun rastgele bir ID üretmesini sağlar. Kendi ID’nizi belirlerseniz benzersiz olmasına dikkat edin — aynı ID’ye sahip iki client birbirini sürekli broker’dan düşürür. Client ID oluşturucu sizin için güvenli ID’ler üretebilir.

Eksiksiz bir publisher

Bir publisher’ın genellikle yapacak başka işleri de vardır — bir sensörü okumak, bir API’yi sorgulamak — bu yüzden network loop’u loop_start() ile arka plandaki bir thread’de çalıştırır:

import json
import random
import time

import paho.mqtt.client as mqtt

BROKER = "broker.hivemq.com"
TOPIC = "testmqtt/python-demo/sensor"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.connect(BROKER, 1883, keepalive=60)
client.loop_start()

try:
    while True:
        payload = json.dumps({
            "temperature": round(random.uniform(18, 26), 1),
            "ts": int(time.time()),
        })
        info = client.publish(TOPIC, payload, qos=1)
        info.wait_for_publish(timeout=5)
        print("Published", payload, "mid", info.mid)
        time.sleep(2)
except KeyboardInterrupt:
    pass
finally:
    client.disconnect()
    client.loop_stop()

Subscriber’ı bir terminalde, publisher’ı başka bir terminalde çalıştırın. Script’inizin publish ettiği mesajları tarayıcıdan da izleyebilirsiniz:

Python mesajlarınızı canlı izleyin

loop_forever, loop_start ve loop karşılaştırması

• loop_forever() — mevcut thread’i bloklar, yeniden bağlanmaları yönetir ve yalnızca disconnect()’ten sonra geri döner. Yalnızca abone olan script’ler ve servisler için en iyisidir.
• loop_start() / loop_stop() — loop’u arka plandaki bir thread’de çalıştırır. Ana thread’iniz başka işler yapıyorsa en iyi seçenektir. Callback’ler bu arka plan thread’inde çalışır; paylaşılan state’i buna göre koruyun.
• loop(timeout) — network trafiğini bir kez işler. Yalnızca paho’yu kendi event loop’unuza entegre ediyorsanız gerekir.

QoS, retained mesajlar ve Last Will

Hem publish()’a hem de subscribe()’a qos=0, 1 veya 2 verin; geçerli QoS ikisinden düşük olanıdır. retain=True, broker’ın son mesajı yeni subscriber’lar için saklamasını sağlar. Broker’ın beklenmedik bir bağlantı kopmasını duyurması için bağlanmadan önce bir Last Will ayarlayın:

client.will_set("testmqtt/python-demo/status", "offline", qos=1, retain=True)
client.connect(BROKER, 1883)
client.publish("testmqtt/python-demo/status", "online", qos=1, retain=True)

Daha fazlası için MQTT QoS seviyeleri, retained mesajlar ve Last Will and Testament yazılarını okuyun.

TLS ve kullanıcı adı/parola ile kimlik doğrulama

Gerçek her kurulum, 8883 portunda TLS ile birlikte kimlik bilgileri kullanmalıdır. İkisi de connect()’ten önce yapılandırılır:

import ssl
import paho.mqtt.client as mqtt

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="sensor-livingroom-01")
client.username_pw_set("my-username", "my-password")

# Uses the system CA store; fine for publicly trusted certificates.
client.tls_set()

# For a private CA or client-certificate auth instead:
# client.tls_set(ca_certs="ca.crt", certfile="client.crt", keyfile="client.key",
#                tls_version=ssl.PROTOCOL_TLS_CLIENT)

client.connect("broker.example.com", 8883, keepalive=60)
client.loop_forever()

Hızlı yerel testler dışında tls_insecure_set(True) kullanmaktan kaçının — hostname doğrulamasını devre dışı bırakır. MQTT 5 için client’ı protocol=mqtt.MQTTv5 ile oluşturun. WebSocket’ler için transport="websockets" verin ve client.ws_set_options(path="/mqtt") çağırın.

Güvenilir şekilde yeniden bağlanma

Hem loop_forever() hem de loop_start(), bağlantı kaybından sonra otomatik olarak yeniden bağlanır. Bekleme süresini (backoff) reconnect_delay_set() ile ayarlayın:

client.reconnect_delay_set(min_delay=1, max_delay=60)

İlk connect() çağrısı farklıdır: broker’a ulaşılamıyorsa bir exception fırlatır (örneğin ConnectionRefusedError veya bir socket zaman aşımı). Onu bir retry döngüsüne sarın ya da arka planda yeniden deneyen connect_async() ve ardından loop_start() kullanın. Bağlantılar sürekli başarısız oluyorsa MQTT bağlantı hataları rehberi olağan şüphelileri ele alır: yanlış port, engelleyen güvenlik duvarı, hatalı kimlik bilgileri veya tekrarlanan client ID’ler.

MQTT 5 özelliklerini kullanma

MQTT 5’in kilidini açmak için protocol=mqtt.MQTTv5 verin: session expiry, user property’ler, message expiry ve ayrıntılı reason code’lar. Bağlantı seçenekleri clean_session’dan (3.1.1 için bir constructor argümanı) connect() üzerindeki clean_start’a taşınır; mesaj başına metadata ise bir Properties nesnesine girer:

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)

connect_props = Properties(PacketTypes.CONNECT)
connect_props.SessionExpiryInterval = 3600  # keep the session for 1 hour

client.connect("broker.hivemq.com", 1883, keepalive=60,
               clean_start=False, properties=connect_props)
client.loop_start()

publish_props = Properties(PacketTypes.PUBLISH)
publish_props.MessageExpiryInterval = 30          # drop if undelivered after 30 s
publish_props.ContentType = "application/json"
publish_props.UserProperty = [("source", "python-demo")]

info = client.publish("testmqtt/python-demo/v5", '{"ok": true}', qos=1,
                      properties=publish_props)
info.wait_for_publish(timeout=5)

client.disconnect()
client.loop_stop()

on_message içinde alınan property’lere msg.properties üzerinden erişebilirsiniz. Sürüm 5’e ihtiyacınız olup olmadığından emin değil misiniz? MQTT 5 ve 3.1.1 karşılaştırmamız pratikte önemli olan farkları listeler.

Sık yapılan hatalar

• 1.x kodunu kopyalamak. mqtt.Client("my-id"), 2.x’in callback API sürümünü beklediği yere client ID’yi geçirir. client_id= keyword argümanını kullanın.
• Loop’u unutmak. loop_forever() veya loop_start() olmadan hiçbir şey gönderilmez ya da alınmaz ve callback’ler hiç tetiklenmez.
• Çok erken çıkmak. publish() çağırıp hemen sonlanan bir script, mesaj socket’ten çıkmadan kapanabilir. wait_for_publish() çağırın veya publish.single() kullanın.
• Körü körüne decode etmek. msg.payload bir bytes nesnesidir. Metin için msg.payload.decode("utf-8"), JSON için json.loads() kullanın ve decode hatalarını yakalayın — aynı topic’teki diğer client’lar her şeyi gönderebilir.
• Public bir broker’da #’e abone olmak. Diğer tüm kullanıcıların trafiğini alırsınız. Kendi topic prefix’inizin içinde kalın; MQTT topic’leri ve wildcard’lar (joker karakterler) yazısına bakın.

Tek seferlik publish

Yalnızca tek bir mesaj gönderen cron job’ları ve script’ler için paho.mqtt.publish yardımcısı bağlanma, publish ve bağlantıyı kesme işlemlerini tek bir çağrıda halleder:

import paho.mqtt.publish as publish

publish.single("testmqtt/python-demo/cron", "job finished", qos=1,
               hostname="broker.hivemq.com", port=1883)

Bu kodu kendi host’unuz, port’unuz, kimlik bilgileriniz ve topic’inizle önceden doldurulmuş halde mi istiyorsunuz? Broker ayarlarınıza göre oluşturmak için MQTT kod oluşturucuyu kullanın.

Sıkça sorulan sorular