Paho MQTT Python Eğitimi: 10 Dakikada Publish ve Subscribe

Paho MQTT Python, ister bir Raspberry Pi sensörü için betik yazıyor, ister bir backend servisi geliştiriyor, ister bir cihazı test ediyor olun, Python'dan MQTT konuşmanın standart yoludur. Bu eğitimde paho-mqtt 2.x'i kuracak, broker.connectmq.com üzerinden konuşan bir subscriber ve bir publisher yazacak ve yaygın tuzaklardan kaçınacak kadar network loop'u anlayacaksınız. Sonunda aynı kodu kimlik bilgileri ve TLS kullanan özel bir broker'a geçireceksiniz.

Gerekenler

• Python 3.8 veya daha yenisi (python --version)
• 1883 portuna, TLS için 8883 portuna dışarı erişimi olan bir terminal
• On dakika

İlk bölüm için hesap gerekmez. broker.connectmq.com adresindeki ücretsiz herkese açık MQTT broker, boş olmayan ve benzersiz herhangi bir client ID ile anonim bağlantıları kabul eder. Herkese açık alanı herkes okuyabilir; bu yüzden rastgele bir topic öneki kullanın ve oraya asla gerçek veri göndermeyin.

paho-mqtt Kurulumu

Bir sanal ortam oluşturun ve kütüphaneyi kurun:

python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install "paho-mqtt>=2.0"
python -c "import paho.mqtt; print(paho.mqtt.__version__)"

2.x sürümünü görmelisiniz. Sürüm 2.0, istemcinin oluşturulma ve callback'lerin çağrılma biçimini değiştirdi. İnternetteki pek çok eğitim hâlâ 1.x tarzını gösteriyor; bu tarz artık hata veriyor veya kullanımdan kaldırma uyarıları yazdırıyor. Buradaki her şey güncel tarzı kullanır; mevcut kodu taşıyorsanız paho-mqtt 2.0 callback API sürümü rehberine bakın.

Adım 1: Bir Subscriber

Bunu subscriber.py olarak kaydedin. k7f3q2 değerini kendi rastgele dizginizle değiştirin:

import uuid

import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PORT = 1883
TOPIC = "connectmq/demo/k7f3q2/#"


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"Connection failed: {reason_code}")
        return
    print(f"Connected to {BROKER}")
    # Aboneliğin her yeniden bağlantıdan sonra geri yüklenmesi için burada subscribe olun.
    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 QoS {rc.value}")


def on_message(client, userdata, msg):
    retained = " (retained)" if msg.retain else ""
    print(f"{msg.topic}{retained}: {msg.payload.decode('utf-8', errors='replace')}")


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


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"sub-{uuid.uuid4().hex[:8]}")
client.on_connect = on_connect
client.on_subscribe = on_subscribe
client.on_message = on_message
client.on_disconnect = on_disconnect

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

python subscriber.py ile çalıştırın. "Connected" ve "Subscribed with QoS 1" yazdırır, ardından bekler.

Adım 2: Bir Publisher

İkinci bir terminalde bunu publisher.py olarak kaydedin:

import json
import random
import time
import uuid

import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PORT = 1883
PREFIX = "connectmq/demo/k7f3q2"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"pub-{uuid.uuid4().hex[:8]}")

# Last Will: bu istemci bağlantıyı kesmeden kaybolursa broker tarafından publish edilir.
client.will_set(f"{PREFIX}/status/publisher", payload="offline", qos=1, retain=True)

client.connect(BROKER, PORT, keepalive=60)
client.loop_start()  # arka plan thread'inde network loop

# Retained durum: yeni subscriber'lar bunu hemen alır.
client.publish(f"{PREFIX}/status/publisher", "online", qos=1, retain=True)

try:
    for i in range(10):
        payload = json.dumps({"seq": i, "temperature": round(random.uniform(20, 25), 2)})
        info = client.publish(f"{PREFIX}/sensors/room1", payload, qos=1)
        info.wait_for_publish(timeout=5)
        print(f"Sent #{i} (mid={info.mid}, published={info.is_published()})")
        time.sleep(1)
finally:
    client.publish(f"{PREFIX}/status/publisher", "offline", qos=1, retain=True).wait_for_publish(5)
    client.disconnect()
    client.loop_stop()

python publisher.py komutunu çalıştırın. Subscriber durum mesajını ve saniyede bir olmak üzere on sensör ölçümünü yazdırır.

Paho MQTT Python Network Loop'unu Anlamak

Network loop, paho'nun soketten okuyan ve sokete yazan, keepalive ping'leri gönderen ve callback'lerinizi çağıran kısmıdır. O olmadan hiçbir şey olmaz. Onu çalıştırmanın üç yolu vardır:

Yeni başlayanların en yaygın hatası, hiçbir loop çalışmazken publish() çağırıp neden hiçbir şeyin ulaşmadığını veya QoS 1 mesajlarının neden hiç onaylanmadığını merak etmektir. Bir diğeri, loop_start() kullandığınızda callback'lerin loop thread'inde çalıştığını unutmaktır; bir callback ana thread'inizin de kullandığı verilere dokunuyorsa, onu bir lock veya queue.Queue ile koruyun.

loop_forever() ve loop_start(), kopan bir bağlantıdan sonra otomatik olarak yeniden bağlanır. Subscriber'ın on_connect içinde subscribe olmasının nedeni budur: clean session ile abonelikler yeniden bağlanmada kaybolur ve tekrar yapılmaları gerekir. paho-mqtt yeniden bağlanma rehberi backoff ve çevrimdışı tamponlamayı daha derinlemesine ele alır.

Pratikte QoS, Retained Mesajlar ve Last Will

Publisher, anlamaya değer üç MQTT özelliği kullandı:

QoS 1, broker'ın her mesajı onayladığı ve onay gelmezse paho'nun mesajı yeniden gönderdiği anlamına gelir. wait_for_publish() bu onay gelene kadar bloklar; bu da publish ettikten hemen sonra çıkan betiklerde işe yarar. QoS 0 "gönder ve unut" mantığıyla çalışır ve biraz daha ucuzdur; QoS 2 ek bir gidiş-dönüş pahasına tam olarak bir kez teslimatı garanti eder. MQTT QoS seviyeleri açıklaması yazımız bunları ayrıntılı olarak karşılaştırır.

Retained mesajlar broker tarafından topic başına bir tane olmak üzere saklanır ve daha sonra subscribe olan herkese hemen gönderilir. "online" gibi durum bilgileri veya bir cihazın mevcut yapılandırması için idealdirler. Subscriber'ı yeniden başlatın ve retained durumu "(retained)" işaretiyle anında aldığını görün.

Last Will bağlantı anında kaydedilir. Publisher çökerse veya ağını kaybederse, keepalive süresi dolduktan sonra broker onun adına "offline" publish eder. Döngü sırasında Ctrl+C'ye basarsanız finally bloğundan gelen düzgün "offline" mesajını görürsünüz; bunun yerine işlemi kill -9 ile öldürürseniz broker will mesajını gönderir.

Komut Satırı Araçlarıyla Test

Python kodunuzu diğer taraftan Mosquitto istemcileriyle doğrulayabilirsiniz:

mosquitto_sub -h broker.connectmq.com -p 1883 -i cli-k7f3q2 -t 'connectmq/demo/k7f3q2/#' -v

ConnectMQ portalı hiçbir şey kurmadan aynı görünümü sunar: tarayıcı içi MQTT istemcisi ve canlı topic gezgini, mesajlar geldikçe önekinizin altındaki her topic'i gösterir; bu, Mosquitto kuramadığınız bir makinede hata ayıklarken işe yarar. Portalın özel broker'ınızla çalıştığını, yukarıdaki CLI komutunun ise herkese açık olanı hedeflediğini unutmayın.

CLI mesajlarınızı alıyor ama Python subscriber'ınız almıyorsa sorun subscriber kodunuzdadır; hiçbiri almıyorsa publisher'a veya ağa bakın.

Yaygın Hatalar ve Çözümleri

1. ValueError: Unsupported callback API version veya callback_api_version ile ilgili bir TypeError: paho-mqtt 2.x ile 1.x tarzı oluşturma kullanıyorsunuz. İlk argüman olarak mqtt.CallbackAPIVersion.VERSION2 geçin.
2. TypeError: on_connect() takes 4 positional arguments but 5 were given: callback'iniz eski imzaya sahip. Yukarıda gösterilen sürüm 2 imzalarını kullanın.
3. ConnectionRefusedError veya zaman aşımları: 1883 portu ağınızda engellenmiş. 8883 üzerinden TLS deneyin.
4. Bağlanıyor, sonra döngü halinde bağlantı kopuyor: başka bir istemci aynı client ID'yi kullanıyor. Örneklerde olduğu gibi benzersiz bir tane üretin.
5. Hiçbir şey alınmıyor: topic'lerin büyük/küçük harf ve önek dahil tam olarak eşleştiğini ve bir loop'un çalıştığını kontrol edin.
6. Herkese açık broker'da mesajlar düşüyor: istemci başına saniyede 10 mesaja ve mesaj başına 64 KB'a izin verir. Yavaşlayın veya ölçümleri toplu gönderin.

Kimlik Bilgileri ve TLS ile Özel Bir Broker'a Geçmek

Gerçek veriler için kendi broker'ınızı oluşturun. app.connectmq.com adresinde kaydolun, ücretsiz Serverless planını seçin, bir broker oluşturun ve üretilen kullanıcı adı, parola ve client ID'yi kopyalayın. Ardından bağlantı kodunu değiştirin:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # sistem CA deposunu kullanır ve host adını doğrular
client.connect("broker.connectmq.com", 8883, keepalive=60)

Portalda broker'ınız için gösterilen host adını kullanın. ConnectMQ broker'ları, çoğu sistemdeki varsayılan CA paketinin zaten güvendiği bir Let's Encrypt sertifikası kullanır. Sertifika hatalarıyla karşılaşırsanız Python MQTT TLS rehberi her birini adım adım ele alır. Kimlik bilgilerini kaynak dosyalarda değil ortam değişkenlerinde tutun.

Sırada Ne Var

Publish ve subscribe çalıştığında, bu rehberler aynı istemcinin üzerine inşa eder:

• Python MQTT JSON eğitimi ile payload'ları düzgün yapılandırın.
• aiomqtt ve asyncio ile çok sayıda eş zamanlı görev çalıştırın.
• Python ile MQTT'den SQLite'a eğitimi ile mesajları kalıcı hale getirin.
• MQTT istemci kütüphanelerine genel bakış ile Python'u diğer dillerle karşılaştırın.

Resmi Eclipse Paho Python dokümantasyonu her yöntem ve seçenek için başvuru kaynağıdır.

ConnectMQ ile Başlayın

ConnectMQ; TCP, TLS ve WebSocket üzerinde MQTT 3.1.1 ve 5.0'ı destekleyen yönetilen bir MQTT broker hizmetidir ve portalında Python kodunuzun ne publish ettiğini kontrol etmek için tarayıcı içi bir MQTT istemcisi ve canlı bir topic gezgini bulunur. Herkese açık broker üzerinde deneyin, ardından üretim trafiğini kendi bulut MQTT broker'ınızda çalıştırın. Serverless planında ücretsiz hesap oluşturun, kredi kartı gerekmez.

Sık Sorulan Sorular

Hangi paho-mqtt sürümünü kullanmalıyım?

Güncel 2.x sürümünü kullanın. İstemciyi oluştururken bir callback_api_version argümanı gerektirir ve yeni kod için önerilen seçim sürüm 2 callback'leridir.

loop_start ile loop_forever arasındaki fark nedir?

loop_forever() çağıran thread'i bloklar ve siz bağlantıyı kesene kadar network loop'u çalıştırır. loop_start() aynı loop'u bir arka plan thread'inde çalıştırır, böylece ana kodunuz çalışmaya devam eder.

Paho MQTT Python, MQTT 5'i destekliyor mu?

Evet. Properties, reason code'lar ve session expiry gibi MQTT 5 özelliklerini kullanmak için istemciyi oluştururken protocol=mqtt.MQTTv5 geçin.

paho-mqtt'yi ConnectMQ ile denemek için hesap gerekir mi?

Hayır. broker.connectmq.com, test için 1883 portunda anonim istemcileri kabul eder. Kimlik bilgileri olan özel bir broker'a ihtiyaç duyduğunuzda ücretsiz bir hesap oluşturun.