aiomqtt ve asyncio ile Python'da Asenkron MQTT: Görevler, Worker'lar ve Yeniden Bağlanma

aiomqtt ve asyncio, Python'da callback'siz MQTT sunar: bir publish işlemini await ile beklersiniz, gelen mesajlar üzerinde async for ile dolaşırsınız ve tek bir thread üzerinde ihtiyacınız kadar eş zamanlı görev çalıştırırsınız. Bu eğitimde broker.connectmq.com adresine bağlanan, sensör ölçümlerini publish eden, mesajları bir worker kuyruğu üzerinden tüketen ve ağ koptuğunda düzgün şekilde yeniden bağlanan asenkron bir istemci oluşturacaksınız. Ayrıca neredeyse herkesi tökezleten Windows event loop ayrıntısını da göreceksiniz.

Neden Düz paho-mqtt Yerine aiomqtt?

paho-mqtt bir callback kütüphanesidir. Network loop'u kendi thread'inde çalışır ve bir şey olduğunda fonksiyonlarınızı çağırır. Bu küçük betikler için gayet iyidir, ancak uygulamanız zaten asyncio üzerine kuruluysa, örneğin bir FastAPI servisi, HTTP ve Modbus ile de asenkron konuşan bir gateway veya bir bot ise, işler hantallaşır. Thread'leri ve bir event loop'u karıştırmak run_coroutine_threadsafe, lock'lar ve sinsi kapanış hataları demektir.

aiomqtt, paho-mqtt'yi sarmalar ve bunun yerine asyncio'ya uygun bir API sunar:

Arka planda hâlâ paho çalışır, dolayısıyla protokol davranışı aynıdır. Callback'lerden geliyorsanız paho mqtt python eğitimi kavramlar için faydalı bir başvuru kaynağıdır.

aiomqtt Kurulumu

python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install "aiomqtt>=2.0"

Örnekler, Python 3.11 veya daha yenisini gerektiren asyncio.TaskGroup ve except* kullanır. Testlerinizin diğer kullanıcılardan ayrı kalması için herkese açık MQTT test broker'ı için rastgele bir topic öneki seçin; burada connectmq/demo/k7f3q2 kullanıyoruz.

Windows event loop tuzağı

aiomqtt, event loop'un add_reader ve add_writer metotlarına dayanır. Windows'ta varsayılan event loop, bunları uygulamayan Proactor loop'tur; bu yüzden bağlandığınız anda bir NotImplementedError alırsınız. Aşağıdaki örneklerde olduğu gibi programınızı bunun yerine bir selector event loop üzerinde çalıştırın. Linux ve macOS varsayılan olarak zaten bir selector loop kullanır.

Minimal Bir aiomqtt asyncio Örneği

Tek bir dosyada çalışan en küçük subscriber ve publisher ile başlayın:

import asyncio
import sys
import uuid

import aiomqtt

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


async def main():
    async with aiomqtt.Client(BROKER, 1883, identifier=f"aio-{uuid.uuid4().hex[:8]}") as client:
        await client.subscribe(f"{PREFIX}/hello", qos=1)
        await client.publish(f"{PREFIX}/hello", payload="hello from asyncio", qos=1)
        async for message in client.messages:
            print(message.topic, message.payload.decode())
            break  # ilk mesajdan sonra dur


if sys.platform == "win32":
    asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop)
else:
    asyncio.run(main())

async with bloğu girişte bağlanır ve bir istisna fırlatılsa bile çıkışta bağlantıyı keser. client.messages, tüm aboneliklerden gelen her mesajı geliş sırasına göre veren bir async iterator'dır.

Gerçekçi Bir İstemci: Eş Zamanlı Görevler ve Bir Worker Kuyruğu

Gerçek uygulamalar aynı anda birkaç iş yapar: ölçümleri bir zamanlamaya göre publish eder, komutlara tepki verir ve veritabanı yazma ya da HTTP çağrısı gibi yavaş G/Ç içerebilen gelen verileri işler. Püf noktası, mesajları okuyan döngüyü asla bloklamamaktır. Bunu aio_client.py olarak kaydedin:

import asyncio
import json
import random
import sys
import time
import uuid

import aiomqtt

BROKER = "broker.connectmq.com"
PORT = 1883
PREFIX = "connectmq/demo/k7f3q2"
CLIENT_ID = f"aio-{uuid.uuid4().hex[:8]}"
RECONNECT_INTERVAL = 5
WORKERS = 3


async def publish_readings(client: aiomqtt.Client) -> None:
    seq = 0
    while True:
        reading = {"seq": seq, "temperature": round(random.uniform(20, 25), 2), "ts": time.time()}
        await client.publish(f"{PREFIX}/sensors/room1", json.dumps(reading), qos=1)
        seq += 1
        await asyncio.sleep(2)


async def consume(client: aiomqtt.Client, queue: asyncio.Queue) -> None:
    await client.subscribe(f"{PREFIX}/sensors/+", qos=1)
    await client.subscribe(f"{PREFIX}/commands/#", qos=1)
    async for message in client.messages:
        await queue.put(message)  # hızlıca devret, okumaya devam et


async def handle(message: aiomqtt.Message) -> None:
    if message.topic.matches(f"{PREFIX}/commands/#"):
        print(f"Command on {message.topic}: {message.payload.decode()}")
        return
    data = json.loads(message.payload)
    await asyncio.sleep(0.5)  # yavaş bir veritabanı veya HTTP çağrısını temsil eder
    print(f"Processed reading {data['seq']}: {data['temperature']} C")


async def worker(name: int, queue: asyncio.Queue) -> None:
    while True:
        message = await queue.get()
        try:
            await handle(message)
        except Exception as exc:  # tek bir hatalı mesaj worker'ı öldürmemeli
            print(f"Worker {name} failed on {message.topic}: {exc}")
        finally:
            queue.task_done()


async def main() -> None:
    queue: asyncio.Queue = asyncio.Queue(maxsize=1000)
    workers = [asyncio.create_task(worker(i, queue)) for i in range(WORKERS)]

    client = aiomqtt.Client(
        BROKER,
        PORT,
        identifier=CLIENT_ID,
        protocol=aiomqtt.ProtocolVersion.V5,
        keepalive=30,
        will=aiomqtt.Will(f"{PREFIX}/status/{CLIENT_ID}", payload="offline", qos=1, retain=True),
        # Özel ConnectMQ broker:
        # username="YOUR_USERNAME", password="YOUR_PASSWORD",
        # tls_context=ssl.create_default_context(), ve 8883 portu
    )

    try:
        while True:
            try:
                async with client:
                    print("Connected")
                    await client.publish(f"{PREFIX}/status/{CLIENT_ID}", "online", qos=1, retain=True)
                    async with asyncio.TaskGroup() as tg:
                        tg.create_task(publish_readings(client))
                        tg.create_task(consume(client, queue))
            except* aiomqtt.MqttError as group:
                print(f"Connection lost: {group.exceptions[0]}. Reconnecting in {RECONNECT_INTERVAL}s")
            await asyncio.sleep(RECONNECT_INTERVAL)
    finally:
        for w in workers:
            w.cancel()


if __name__ == "__main__":
    try:
        if sys.platform == "win32":
            asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop)
        else:
            asyncio.run(main())
    except KeyboardInterrupt:
        print("Stopped")

python aio_client.py komutunu çalıştırın. Her iki saniyede bir ölçüm publish edilir, abonelik üzerinden geri alınır ve üç worker'dan biri tarafından işlenir. Başka bir terminalden bir komut gönderin:

mosquitto_pub -h broker.connectmq.com -p 1883 -i aio-cmd-test -t 'connectmq/demo/k7f3q2/commands/led' -m 'on'

Nasıl Çalışır

TaskGroup ile yapılandırılmış eş zamanlılık

Publisher ve consumer, bir TaskGroup içinde kardeş görevler olarak çalışır. Biri hata fırlatırsa, örneğin bağlantı koptuğu ve publish() başarısız olduğu için, grup diğer görevi iptal eder ve hatayı bir exception group içine sararak yeniden fırlatır. except* aiomqtt.MqttError, hangi görev başarısız olursa olsun onu yakalar. Bir task group olmadan, bir görevdeki hata fark edilmeyebilir ve diğer görev ölü bir bağlantı üzerinde çalışmaya devam edebilir.

Mesaj döngüsünü hızlı tutmak

Consumer, mesajları bir asyncio.Queue içine taşımaktan başka bir şey yapmaz. Yavaş kısmı worker'lar halleder. Bu önemlidir, çünkü aiomqtt mesajları tek bir iterator üzerinden sırayla iletir; async for döngüsü içinde yarım saniyelik bir veritabanı yazmasını bekleseydiniz, diğer her mesaj onun arkasında beklerdi. Üç worker ile üç mesaj eş zamanlı işlenir.

Kuyruğun maxsize değeri geri basınç (backpressure) sağlar. Worker'lar yetişemezse queue.put() bekler; bu da belleği sınırsızca büyütmek yerine broker'dan okumayı yavaşlatır. Topic başına sıralama önemliyse, mesajları paylaşılan bir kuyruk yerine topic hash'ine göre worker'lara yönlendirin.

Düzgün yeniden bağlanmalar

Bir aiomqtt istemcisi yeniden kullanılabilir ancak reentrant değildir: çıktıktan sonra async with client bloğuna tekrar girebilirsiniz, ama aynı anda iki kez giremezsiniz. Dıştaki while True döngüsü tam olarak bunu yapar. Bir hatadan sonra birkaç saniye bekler, yeniden bağlanır, online durumunu yeniden publish eder ve abonelikler bağlantıya ait olduğu için yeniden subscribe olur. İşlem düzgün bir bağlantı kesme olmadan ölürse Last Will "offline" publish eder.

Üretim için sabit aralığı bir veya iki dakikayla sınırlandırılmış üstel backoff ile değiştirin; böylece bir kesintiden sonra istemci filosu broker'a aynı anda yüklenmez. paho-mqtt yeniden bağlanma rehberi backoff ve oturum stratejisini daha derinlemesine ele alır ve aynı fikirler burada da geçerlidir.

aiomqtt ile MQTT 5 Properties Kullanmak

Yukarıdaki istemci ProtocolVersion.V5 ile bağlandığı için tek tek mesajlara MQTT 5 properties ekleyebilirsiniz. aiomqtt, paho'nun Properties sınıfını yeniden kullanır; dolayısıyla kod, callback tabanlı bir istemcidekiyle aynıdır:

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

props = Properties(PacketTypes.PUBLISH)
props.MessageExpiryInterval = 60          # 60 sn içinde teslim edilmezse at
props.ContentType = "application/json"
await client.publish(f"{PREFIX}/sensors/room1", payload, qos=1, properties=props)

Mesaj süresi dolumu (message expiry) komutlar için kullanışlıdır: on dakika geç gelen bir "aç" komutu, çoğu zaman hiç gelmeyen birinden daha kötüdür. Alıcı tarafta message.properties aynı özellikleri sunar. MQTT 5 özellikleri rehberi hangi properties'in kullanılmaya değer olduğunu açıklar.

Yaygın Sorunlar

1. Windows'ta NotImplementedError: Proactor loop üzerindesiniz. Gösterildiği gibi selector loop kullanın.
2. MqttError: Could not connect: 1883 portu engellenmiş veya host adı yanlış. Bir tls_context ile 8883 üzerinden TLS deneyin.
3. Bağlantı birkaç saniyede bir kopuyor: ikinci bir istemci aynı identifier'ı kullanıyor. Her işlem için benzersiz bir tane üretin.
4. Mesajlar geç veya toplu halde geliyor: async for içinde yavaş bir şey çalışıyor. Onu worker'lara taşıyın.
5. Herkese açık broker'da hız sınırına takılmak: istemci başına saniyede 10 mesaja ve istemci başına 10 aboneliğe izin verir. Publish aralığını makul tutun.
6. Bloklayan çağrılar her şeyi donduruyor: bir coroutine içindeki senkron requests.get() veya time.sleep() tüm loop'u bloklar. Asenkron kütüphaneler veya asyncio.to_thread() kullanın.

Üretime Geçiş

Özel bir ConnectMQ broker'ında üretilen username, password değerlerini ve client ID'yi identifier olarak, ayrıca 8883 portunda tls_context=ssl.create_default_context() geçersiniz. Python MQTT TLS rehberi sertifika doğrulamasını açıklar, Python MQTT JSON eğitimi ise yukarıdaki ölçümler gibi payload'ların nasıl sürümleneceğini gösterir. Worker'ların işlediği verileri saklamak için Python ile MQTT'den SQLite'a logger'ı uyarlayın; diğer dillerdeki istemcilerle karşılaştırma için MQTT istemci kütüphanelerine genel bakış yazısına bakın. GitHub'daki aiomqtt projesi her istemci seçeneğini belgeler.

Asenkron Python Servislerinizi ConnectMQ Üzerinde Çalıştırın

ConnectMQ; TCP, TLS ve WebSocket üzerinde MQTT 3.1.1 ve 5.0'ı destekleyen ve bu istemcinin dayandığı session expiry, retained mesajlar ve Last Will özelliklerini sunan yönetilen bir MQTT broker hizmetidir. broker.connectmq.com üzerinde geliştirin, ardından bağlantı ayarları dışında kodunuzu değiştirmeden özel bir bulut MQTT broker'ına dağıtın. Serverless planında ücretsiz kaydolun, kredi kartı gerekmez.

Sık Sorulan Sorular

aiomqtt, paho-mqtt'den daha mı hızlı?

Dahili olarak paho-mqtt kullanır, bu yüzden ham protokol verimi benzerdir. Avantajı yapıdadır: asenkron kod, tek bir thread'in callback'ler veya lock'lar olmadan birçok eş zamanlı işlemi yönetmesini sağlar.

aiomqtt Windows'ta neden başarısız oluyor?

Varsayılan Windows Proactor event loop'unda aiomqtt'nin ihtiyaç duyduğu add_reader ve add_writer metotları yoktur. Programınızı asyncio.SelectorEventLoop üzerinde çalıştırdığınızda sorun çözülür.

aiomqtt otomatik olarak yeniden bağlanır mı?

Hayır. Kopan bir bağlantı aiomqtt.MqttError fırlatır ve siz bir döngü içinde async with client bloğuna yeniden girerek bağlanırsınız; bu da backoff ve yeniden subscribe işlemlerini kontrol etmenizi sağlar.

aiomqtt'yi FastAPI ile kullanabilir miyim?

Evet. İstemciyi uygulamanın lifespan handler'ında başlatın ve consumer'ı aynı event loop üzerinde bir arka plan görevi olarak çalıştırın.