Django ile MQTT: management command worker ve Channels

Django ile MQTT kullanmanın güvenilir yolu, MQTT bağlantısını web sürecinin dışında tutmaktır. Uzun ömürlü bir subscriber'ı özel bir management command olarak çalıştırın, normal ORM ile veritabanına yazmasına izin verin, view'lardan kısa ömürlü bir istemciyle publish edin ve tarayıcıların mesajlara canlı ihtiyaç duyduğu yerde Django Channels kullanın.

Bu eğitim tam olarak bunu paho-mqtt 2.x ile kuruyor, ardından worker'ın nasıl deploy edileceğini ve tüm yolun uçtan uca nasıl test edileceğini gösteriyor.

Neden apps.py veya bir view içinde bağlanmamalı?

Çoğu kişinin aklına ilk gelen, AppConfig.ready() içinde bir paho istemcisi başlatmaktır. Dizüstü bilgisayarınızda çalışır, sonra ince yollarla bozulur:

İstemciyi nerede başlatıyorsunuz	Ne ters gidiyor
`AppConfig.ready()`	Django'yu yükleyen her süreçte çalışır: `runserver` (autoreloader yüzünden iki kez), her Gunicorn worker'ı, `migrate`, `shell`, Celery worker'ları. Her biri bir bağlantı açar.
Bir view içinde	Her HTTP isteği için yeni bir MQTT oturumu: yavaştır, broker'ı CONNECT'lerle boğar ve iki istek aynı client ID'yi paylaştığı anda bozulur.
Bir management command	Bilerek başlattığınız tam olarak bir süreç; denetlemesi ve yeniden başlatması kolay.

Aynı topic'e abone olan birden fazla süreç, her mesajın birkaç kez işlenmesi demektir; client ID'yi paylaşıyorlarsa broker onları döngü halinde düşürür (bkz. yinelenen client ID). Yani: abone olmayı bir worker'da yapın, publish'i ihtiyaç duyduğunuz her yerden yapın.

Proje kurulumu

Kurulum

pip install django paho-mqtt channels channels-redis daphne
django-admin startproject mysite .
python manage.py startapp telemetry

telemetry'yi INSTALLED_APPS'e ekleyin ve gelen ölçümler için bir model oluşturun:

telemetry/models.py

from django.db import models


class Reading(models.Model):
    device = models.CharField(max_length=64, db_index=True)
    topic = models.CharField(max_length=255)
    payload = models.JSONField()
    received_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        ordering = ["-received_at"]

Bağlantı ayarlarını ortam değişkenlerinden okuyarak settings.py içinde tutun:

mysite/settings.py (alıntı)

import os

MQTT_HOST = os.environ.get("MQTT_HOST", "broker.connectmq.com")
MQTT_PORT = int(os.environ.get("MQTT_PORT", "8883"))
MQTT_USER = os.environ.get("MQTT_USER", "")
MQTT_PASS = os.environ.get("MQTT_PASS", "")
MQTT_TOPIC = "testmqtt/django/+/telemetry"

python manage.py makemigrations telemetry && python manage.py migrate komutunu çalıştırın.

Management command olarak MQTT worker'ı

telemetry/management/commands/mqtt_listen.py dosyasını oluşturun (management/ ve commands/ klasörlerinin her birinde boş bir __init__.py olmalı):

telemetry/management/commands/mqtt_listen.py

import json
import os
import socket

import paho.mqtt.client as mqtt
from django.conf import settings
from django.core.management.base import BaseCommand
from django.db import close_old_connections

from telemetry.models import Reading


class Command(BaseCommand):
    help = "Subscribe to MQTT telemetry and store it"

    def handle(self, *args, **options):
        client_id = f"django-worker-{socket.gethostname()}-{os.getpid()}"
        client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
        client.username_pw_set(settings.MQTT_USER, settings.MQTT_PASS)
        client.tls_set()  # sistem CA paketi, hostname'i doğrular
        client.reconnect_delay_set(min_delay=1, max_delay=60)

        client.on_connect = self.on_connect
        client.on_message = self.on_message
        client.on_disconnect = self.on_disconnect

        client.connect(settings.MQTT_HOST, settings.MQTT_PORT, keepalive=60)
        self.stdout.write(f"Connecting as {client_id}")
        try:
            client.loop_forever()  # bloklar, otomatik yeniden bağlanır
        except KeyboardInterrupt:
            client.disconnect()

    def on_connect(self, client, userdata, flags, reason_code, properties):
        if reason_code.is_failure:
            self.stderr.write(f"Connect failed: {reason_code}")
            return
        # Burada abone olmak, her yeniden bağlantıdan sonra aboneliği geri yükler
        client.subscribe(settings.MQTT_TOPIC, qos=1)
        self.stdout.write(f"Subscribed to {settings.MQTT_TOPIC}")

    def on_disconnect(self, client, userdata, flags, reason_code, properties):
        self.stderr.write(f"Disconnected: {reason_code}")

    def on_message(self, client, userdata, msg):
        close_old_connections()  # CONN_MAX_AGE'den eski DB bağlantılarını kapatır
        try:
            data = json.loads(msg.payload)
        except (UnicodeDecodeError, json.JSONDecodeError):
            self.stderr.write(f"Ignoring non-JSON payload on {msg.topic}")
            return
        device = msg.topic.split("/")[2]
        Reading.objects.create(device=device, topic=msg.topic, payload=data)

Burada birkaç ayrıntı önemlidir:

connect()'ten sonra değil, on_connect içinde abone olun. paho kendi kendine yeniden bağlanır; callback içinde abone olmak aboneliği her seferinde geri yükler.
loop_forever() callback'leri ana thread'de çalıştırır, bu yüzden senkron ORM çağrıları güvenlidir. Bu süreçte sync_to_async gerekmez.
close_old_connections(), Django'nun her isteğin başında ve sonunda çağırdığı fonksiyondur. Bir worker hiçbir zaman istek görmez; bu yüzden onu kendiniz çağırın, yoksa süreç bir veritabanı yeniden başlatmasından sonra kapanmış bir bağlantı yüzünden eninde sonunda hata verir.
Süreç başına benzersiz bir client ID (hostname + PID). Farklı bir kalıp isterseniz client ID üreticisi ile oluşturun.
Bozuk bir payload yüzünden asla çökmeyin. Callback içinde yakalanmayan bir exception döngüyü durdurur. Parse etme ve doğrulama için Python'da MQTT üzerinden JSON gönderme yazısına bakın.

python manage.py mqtt_listen ile başlatın.

View'lardan ve task'lerden publish

Publish etmek farklı bir problemdir: kısa, ara sıra ve isteklerle tetiklenir. Düşük hızlar için (komutlar, bildirimler) en basit doğru seçenek paho.mqtt.publish.single'dır; bağlanır, publish eder, QoS el sıkışmasını bekler ve bağlantıyı kapatır:

telemetry/views.py

import json

from django.conf import settings
from django.http import JsonResponse
from django.views.decorators.http import require_POST
from paho.mqtt import publish


@require_POST
def send_command(request, device):
    command = json.loads(request.body)
    publish.single(
        f"testmqtt/django/{device}/commands",
        json.dumps(command),
        qos=1,
        hostname=settings.MQTT_HOST,
        port=settings.MQTT_PORT,
        auth={"username": settings.MQTT_USER, "password": settings.MQTT_PASS},
        tls={},  # sistem CA paketiyle TLS
    )
    return JsonResponse({"queued": True})

publish.single her çağrıda yeni bir bağlantı açar; yani her seferinde bir TCP + TLS el sıkışması maliyeti vardır. Dakikada birkaç mesaj için bu sorun değildir. Saniyede çok sayıda mesaj publish ediyorsanız bunun yerine süreç başına bir istemci tutun: tembel (lazy) şekilde oluşturun, loop_start()'ı bir kez çağırın ve süreç başına bir client ID verin. Aynı publish.single çağrısı bir Celery task'i içinde de çalışır; HTTP yanıtını bloklamaması gereken publish'ler için iyi bir yerdir.

Channels ile tarayıcıya mesaj gönderme

Telemetriyi canlı göstermek için worker'ın her mesajı bir Channels grubuna iletmesini sağlayın; o gruptaki her WebSocket consumer'ı mesajı alır. Worker ve ASGI sunucusu farklı süreçler olduğu için Channels'ı bir Redis katmanıyla yapılandırın:

mysite/settings.py (alıntı)

INSTALLED_APPS = ["daphne", *INSTALLED_APPS]  # daphne first, so runserver is ASGI
ASGI_APPLICATION = "mysite.asgi.application"
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {"hosts": [("127.0.0.1", 6379)]},
    }
}

Consumer gruba katılır ve olayları iletir:

telemetry/consumers.py

from channels.generic.websocket import AsyncJsonWebsocketConsumer


class TelemetryConsumer(AsyncJsonWebsocketConsumer):
    async def connect(self):
        await self.channel_layer.group_add("telemetry", self.channel_name)
        await self.accept()

    async def disconnect(self, code):
        await self.channel_layer.group_discard("telemetry", self.channel_name)

    async def telemetry_message(self, event):  # handles type "telemetry.message"
        await self.send_json(event["data"])

mysite/asgi.py

import os

from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")
django_asgi_app = get_asgi_application()

from channels.routing import ProtocolTypeRouter, URLRouter  # noqa: E402
from django.urls import path  # noqa: E402

from telemetry.consumers import TelemetryConsumer  # noqa: E402

application = ProtocolTypeRouter({
    "http": django_asgi_app,
    "websocket": URLRouter([path("ws/telemetry/", TelemetryConsumer.as_asgi())]),
})

Son olarak worker'daki on_message'ı genişletin. Worker senkron olduğu için asenkron group_send'i async_to_sync ile sarın:

mqtt_listen.py (genişletilmiş on_message)

from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer

channel_layer = get_channel_layer()

# on_message içinde, Reading.objects.create(...) sonrasında
async_to_sync(channel_layer.group_send)(
    "telemetry",
    {"type": "telemetry.message", "data": {"device": device, **data}},
)

Tarayıcıda new WebSocket("ws://localhost:8000/ws/telemetry/") artık her ölçümü JSON olarak alır. Kimlik doğrulama eklerseniz router'ı AuthMiddlewareStack ile sarın ve connect() içinde self.scope["user"]'ı kontrol edin.

Production'da çalıştırma

Worker'a diğer daemon'lar gibi davranın: ASGI sunucusu için bir servis, mqtt_listen için bir servis.

/etc/systemd/system/mqtt-listen.service

[Unit]
Description=Django MQTT worker
After=network-online.target

[Service]
WorkingDirectory=/srv/mysite
EnvironmentFile=/srv/mysite/.env
ExecStart=/srv/mysite/.venv/bin/python manage.py mqtt_listen
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

Yedeklilik veya verim için birden fazla worker'a ihtiyacınız varsa, sadece kopyalar başlatmayın: her kopya her ölçümü kaydeder. Bunun yerine bir shared subscription ($share/django/testmqtt/django/+/telemetry) ile abone olun; böylece broker mesajları grup içinde dağıtır. Shared subscription'lar yazısı kuralları ve hangi broker'ların desteklediğini açıklar.

Tüm yolu test etme

Redis'i, python manage.py runserver'ı ve python manage.py mqtt_listen'i ayrı terminallerde başlatın.
Online MQTT istemcisini açın, aynı broker'a bağlanın ve testmqtt/django/sensor-1/telemetry topic'ine {"temperature": 21.5} publish edin ya da CLI kullanın:

Test ölçümü publish etme

mosquitto_pub -h broker.connectmq.com -p 8883 \
  -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t testmqtt/django/sensor-1/telemetry -q 1 \
  -m '{"temperature": 21.5}'

Veritabanını kontrol edin: python manage.py shell -c "from telemetry.models import Reading; print(Reading.objects.first().payload)".
hello gibi geçersiz bir payload publish edin ve worker'ın çökmek yerine bir uyarı logladığını doğrulayın.
Broker bağlantısını kesin (veya ağınızı kısa süreliğine kapatın) ve worker'ın kopmayı logladığını, yeniden bağlandığını ve mesajları kaydetmeye devam ettiğini doğrulayın.

Django topic'lerini online istemcide izleyin

Desen her Django projesi için aynıdır: denetlenen tek bir süreç aboneliğin sahibidir, view'lar kısa süreliğine publish eder ve Channels mesajları tarayıcılara taşır. Framework'leri karşılaştırıyorsanız, bu eğitimin Flask ve FastAPI sürümleri aynı kuralları kendi yaşam döngüsü hook'larıyla uygular.

Sıkça sorulan sorular

Bir Django projesinde MQTT istemcisini nerede başlatmalıyım?

Kendi sürecinde; AppConfig.ready() veya bir view içinde değil. loop_forever() çağıran özel bir management command (python manage.py mqtt_listen) en basit seçenektir: bir kez başlar, aboneliğin sahibidir ve systemd, Docker veya Kubernetes ile denetlenebilir.

paho-mqtt callback'lerinde Django ORM'ini kullanabilir miyim?

Evet. loop_forever() ile callback'ler komutun ana thread'inde çalışır, bu yüzden senkron ORM çağrıları sorunsuzdur. Uzun süre çalışan bir worker'ın ölü bir veritabanı bağlantısını tutmaması için close_old_connections()'ı düzenli olarak çağırın.

Channels neden MQTT worker'ımdan gelen mesajları iletmiyor?

Çoğu zaman proje, yalnızca tek bir süreç içinde çalışan InMemoryChannelLayer kullanıyordur. MQTT worker'ı ve ASGI sunucusu ayrı süreçlerdir; aynı Redis'e işaret eden channels_redis gibi paylaşılan bir katmana ihtiyaç duyarlar.

Django ile MQTT: arka plan worker'ları ve Channels