aiomqtt ile Python'da asenkron MQTT

aiomqtt, Python asyncio kodundan MQTT kullanmanın en doğal yoludur. paho-mqtt'yi asenkron bir context manager içine sarar: async with aiomqtt.Client(...) bağlanır, await client.publish(...) gönderir, async for message in client.messages alır — callback yok, event loop'unuzla koordine etmeniz gereken bir arka plan thread'i yok.

Bu eğitim kurulumu, abone olmayı ve publish etmeyi, broker yeniden başlatmalarına dayanan bir yeniden bağlanma döngüsünü, yavaş mesajların eşzamanlı işlenmesini, TLS'i ve istemcinin bir FastAPI uygulamasına nasıl bağlanacağını anlatıyor.

Kurulum ve ilk bağlantı

aiomqtt, Python 3.8 veya üstünü gerektirir ve tek bağımlılık olarak paho-mqtt'yi getirir:

pip install aiomqtt

Çalışan en küçük program bir mesaj publish edip çıkar:

import asyncio
import aiomqtt

async def main():
    async with aiomqtt.Client("broker.connectmq.com", port=1883) as client:
        await client.publish("testmqtt/aiomqtt/hello", payload="asyncio'dan merhaba", qos=1)

asyncio.run(main())

async with bloğundan çıkıldığında temiz bir DISCONNECT gönderilir. Broker kimlik bilgisi istiyorsa username ve password verin; identifier ile sabit bir client ID belirleyin (diğer test istemcileriyle çakışmamak için client ID üreticisi ile benzersiz bir tane oluşturun):

import os

client = aiomqtt.Client(
    "broker.connectmq.com",
    port=1883,
    identifier="testmqtt-aiomqtt-demo-01",
    username=os.environ["MQTT_USER"],
    password=os.environ["MQTT_PASS"],
    keepalive=30,
    protocol=aiomqtt.ProtocolVersion.V5,  # varsayılan MQTT 3.1.1
)

async for ile abone olmak

aiomqtt 2.0'dan beri gelen tüm mesajlar tek bir asenkron iterator'a, client.messages'a düşer. Önce abone olun, sonra döngüye girin:

import asyncio
import aiomqtt

async def main():
    async with aiomqtt.Client("broker.connectmq.com") as client:
        await client.subscribe("testmqtt/aiomqtt/#", qos=1)
        async for message in client.messages:
            if message.topic.matches("testmqtt/aiomqtt/+/temperature"):
                print("temp", message.topic.value, message.payload.decode())
            elif message.topic.matches("testmqtt/aiomqtt/+/status"):
                print("status", message.topic.value, message.payload.decode())

asyncio.run(main())

message.topic bir Topic nesnesidir: .value string değeridir, .matches() ise MQTT wildcard kurallarını uygular; böylece birden fazla aboneliğin mesajlarını tek bir döngüde yönlendirebilirsiniz. Bir filtrenin eşleşip eşleşmediğinden emin değilseniz topic matcher ile kontrol edin. message.payload genellikle bytes'tır; decode edin ya da json.loads'a verin. Ayrıca message.qos, message.retain ve MQTT 5'te message.properties da mevcuttur.

Subscriber'ı test etmek için online MQTT istemcisini açın, broker.connectmq.com'a bağlanın ve testmqtt/aiomqtt/kitchen/temperature topic'ine publish edin.

Aynı anda publish etmek ve almak

Tek bir istemci, başka bir coroutine client.messages'ı tüketirken publish edebilir. İkisini bir TaskGroup içinde çalıştırın (Python 3.11+):

import asyncio
import json
import time
import aiomqtt

TOPIC = "testmqtt/aiomqtt/sensor-1/temperature"

async def produce(client: aiomqtt.Client):
    while True:
        payload = json.dumps({"t": round(time.time()), "value": 21.5})
        await client.publish(TOPIC, payload=payload, qos=1)
        await asyncio.sleep(2)

async def consume(client: aiomqtt.Client):
    await client.subscribe("testmqtt/aiomqtt/#")
    async for message in client.messages:
        print(message.topic.value, json.loads(message.payload))

async def main():
    async with aiomqtt.Client("broker.connectmq.com") as client:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(produce(client))
            tg.create_task(consume(client))

asyncio.run(main())

await client.publish(), mesaj istenen QoS ile teslim edildiğinde döner (QoS 1'de PUBACK'ten sonra). Başka bir thread'den çağırmayın — her şeyi event loop üzerinde tutun.

Gerçekten çalışan bir yeniden bağlanma kalıbı

aiomqtt kendi kendine yeniden bağlanmaz. Ağ koptuğunda veya broker yeniden başladığında async with bloğu aiomqtt.MqttError fırlatır; mesaj genellikle Disconnected during message iteration ya da [code:7] The connection was lost. gibidir. İstemci nesnesi yeniden kullanılabilir, bu yüzden standart kalıp context manager'ı saran bir döngüdür:

import asyncio
import random
import aiomqtt

async def main():
    client = aiomqtt.Client("broker.connectmq.com", identifier="testmqtt-aiomqtt-worker")
    delay = 1
    while True:
        try:
            async with client:
                delay = 1                                  # bağlandı: backoff'u sıfırla
                await client.subscribe("testmqtt/aiomqtt/#", qos=1)
                async for message in client.messages:
                    print(message.topic.value, message.payload)
        except aiomqtt.MqttError as err:
            wait = delay + random.uniform(0, delay)        # jitter
            print(f"bağlantı koptu ({err}); {wait:.1f} sn sonra yeniden bağlanılıyor")
            await asyncio.sleep(wait)
            delay = min(delay * 2, 60)                     # üstel backoff

asyncio.run(main())

Üç ayrıntı önemlidir. Abonelikler bloğun içinde yapılır, böylece her yeniden bağlanmadan sonra geri yüklenir. Bekleme süresi jitter ile üstel olarak artar; böylece bir worker filosu, broker geri geldiğinde onu aynı anda bombardımana tutmaz — bkz. yeniden bağlanma için en iyi uygulamalar. Sabit bir identifier ise bunu kalıcı bir oturumla birleştirmenizi sağlar (MQTT 3.1.1'de clean_session=False, MQTT 5'te clean_start ve session expiry interval); böylece siz çevrimdışıyken publish edilen QoS 1 mesajları yine teslim edilir.

Test etmek için betiği çalıştırın, yerel broker'ınızı yeniden başlatın (ya da ağ kablosunu çekin) ve log'u izleyin: bir bağlantı koptu satırı, ardından yeniden akan mesajlar görmelisiniz.

Mesaj döngüsünü bloklamayın

async for döngüsü mesajları teker teker işler. Bir handler yavaş bir veritabanı yazımını ya da HTTP çağrısını beklerse diğer tüm mesajlar onun arkasında bekler. Yavaş işleri task'lere devredin ve eşzamanlılığı bir semaphore ile sınırlayın:

sem = asyncio.Semaphore(20)

async def handle(message: aiomqtt.Message):
    async with sem:
        await save_to_db(message)          # yavaş I/O işiniz

async with asyncio.TaskGroup() as tg:
    async for message in client.messages:
        tg.create_task(handle(message))

CPU yoğun işler (görüntü çözme, ML çıkarımı) loop'u yine bloklar; bunları asyncio.to_thread()'e ya da bir process pool'a taşıyın. Gelen mesajlar dahili bir kuyrukta tamponlanır; yavaş bir tüketici belleği sınırsız büyütecekse max_queued_incoming_messages kuyruğun boyutunu sınırlar.

TLS ve WebSocket

Üretim trafiği için 8883 portunda TLS kullanın. Standart bir ssl.SSLContext verin:

import ssl

client = aiomqtt.Client(
    "broker.connectmq.com",
    port=8883,
    tls_context=ssl.create_default_context(),
    username=os.environ["MQTT_USER"],
    password=os.environ["MQTT_PASS"],
)

Yalnızca HTTPS'e izin veren ağlarda taşıma katmanını güvenli WebSocket'e çevirin: aynı tls_context ile birlikte port=8084, transport="websockets", websocket_path="/mqtt". İstemci sertifikaları ve özel CA'lar Python'da MQTT TLS sertifikaları yazısında anlatılıyor.

aiomqtt'yi FastAPI içinde kullanmak

FastAPI zaten bir event loop üzerinde çalıştığı için aiomqtt doğal olarak oturur. Bağlantıyı lifespan handler'ında açın, listener'ı arka plan task'i olarak başlatın ve istemciyi app.state aracılığıyla istek handler'larına sunun:

import asyncio
from contextlib import asynccontextmanager
import aiomqtt
from fastapi import FastAPI

latest: dict[str, str] = {}

async def listen(client: aiomqtt.Client):
    await client.subscribe("testmqtt/aiomqtt/+/temperature")
    async for message in client.messages:
        latest[message.topic.value] = message.payload.decode()

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with aiomqtt.Client("broker.connectmq.com") as client:
        app.state.mqtt = client
        task = asyncio.create_task(listen(client))
        yield
        task.cancel()

app = FastAPI(lifespan=lifespan)

@app.post("/publish/{device}")
async def publish(device: str, value: float):
    await app.state.mqtt.publish(f"testmqtt/aiomqtt/{device}/temperature", payload=str(value), qos=1)
    return {"ok": True}

@app.get("/latest")
async def get_latest():
    return latest

uvicorn app:app ile — tek bir worker ile — çalıştırın. Her worker süreci kendi MQTT bağlantısını açar; aynı identifier'a sahip birden fazla worker birbirini broker'dan düşürür. Tarayıcılara WebSocket ile push ve yeniden bağlanma yönetimi içeren tam sürüm FastAPI ile MQTT yazısında.

aiomqtt mi, düz paho mu?

Seçeneklere daha geniş bir bakış için Python MQTT istemci kütüphaneleri karşılaştırmasına göz atın.

aiomqtt mesajlarınızı canlı izleyin

aiomqtt, MQTT kodunu okunur tutar: bağlantı için bir context manager, mesajlar için bir döngü, hatalar için sade bir try/except. Yeniden bağlanma döngüsünü ilk günden ekleyin, yavaş işleri mesaj döngüsünden uzak tutun ve deploy etmeden önce her şeyi ikinci bir istemciyle doğrulayın.

Sıkça sorulan sorular