MQTT mit Django: Worker als Management Command + Channels

Der zuverlässige Weg, MQTT mit Django zu nutzen: Halte die MQTT-Verbindung aus dem Webprozess heraus. Lass einen lang laufenden Subscriber als eigenes Management Command laufen, der mit dem normalen ORM in die Datenbank schreibt, publiziere aus Views mit einem kurzlebigen Client und nutze Django Channels, wenn Browser die Nachrichten live brauchen.

Dieses Tutorial baut genau das mit paho-mqtt 2.x auf und zeigt danach, wie du den Worker deployst und den kompletten Weg Ende-zu-Ende testest.

Warum nicht in apps.py oder einer View verbinden?

Die erste Idee der meisten ist, einen paho-Client in AppConfig.ready() zu starten. Das klappt auf dem Laptop und bricht dann auf subtile Weise:

Wo du den Client startest	Was schiefgeht
`AppConfig.ready()`	Läuft in jedem Prozess, der Django lädt: `runserver` (zweimal, wegen des Autoreloaders), jeder Gunicorn-Worker, `migrate`, `shell`, Celery-Worker. Jeder öffnet eine Verbindung.
In einer View	Eine neue MQTT-Session pro HTTP-Request: langsam, flutet den Broker mit CONNECTs und scheitert, sobald zwei Requests dieselbe Client-ID nutzen.
Ein Management Command	Genau ein Prozess, den du bewusst startest – leicht zu überwachen und neu zu starten.

Mehrere Prozesse mit derselben Subscription bedeuten, dass jede Nachricht mehrfach verarbeitet wird, und teilen sie sich eine Client-ID, wirft der Broker sie im Kreis hinaus (siehe doppelte Client-ID). Also: Subscribe im Worker, publizieren, wo immer du es brauchst.

Projekt aufsetzen

Installation

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

Füge telemetry zu INSTALLED_APPS hinzu und lege ein Model für eingehende Messwerte an:

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"]

Die Verbindungseinstellungen gehören in settings.py und kommen aus der Umgebung:

mysite/settings.py (Auszug)

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"

Führe python manage.py makemigrations telemetry && python manage.py migrate aus.

Der MQTT-Worker als Management Command

Lege telemetry/management/commands/mqtt_listen.py an (management/ und commands/ brauchen jeweils eine leere __init__.py):

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()  # System-CA-Bundle, prüft den Hostnamen
        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()  # blockiert, verbindet automatisch neu
        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
        # Hier abonnieren, damit die Subscription nach jedem Reconnect wiederhergestellt wird
        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()  # DB-Verbindungen älter als CONN_MAX_AGE verwerfen
        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)

Ein paar Details sind entscheidend:

Abonniere in on_connect, nicht nach connect(). paho verbindet sich selbst neu, und das Subscribe im Callback stellt die Subscription jedes Mal wieder her.
loop_forever() führt die Callbacks im Haupt-Thread aus, synchrone ORM-Aufrufe sind also sicher. In diesem Prozess brauchst du kein sync_to_async.
close_old_connections() ruft Django am Anfang und Ende jedes Requests auf. Ein Worker sieht nie einen Request – ruf es also selbst auf, sonst scheitert der Prozess irgendwann nach einem Datenbank-Neustart an einer geschlossenen Verbindung.
Eine eindeutige Client-ID pro Prozess (Hostname + PID). Ein anderes Schema erzeugst du mit dem Client-ID-Generator.
Stürze nie wegen eines kaputten Payloads ab. Eine unbehandelte Exception im Callback stoppt den Loop. Mehr zu Parsen und Validieren findest du unter JSON über MQTT in Python senden.

Starte ihn mit python manage.py mqtt_listen.

Aus Views und Tasks publizieren

Publizieren ist ein anderes Problem: kurz, gelegentlich und durch Requests ausgelöst. Für niedrige Raten (Kommandos, Benachrichtigungen) ist paho.mqtt.publish.single die einfachste korrekte Lösung – es verbindet, publiziert, wartet den QoS-Handshake ab und trennt die Verbindung:

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={},  # TLS mit dem System-CA-Bundle
    )
    return JsonResponse({"queued": True})

publish.single öffnet jedes Mal eine neue Verbindung, also einen TCP- und TLS-Handshake pro Aufruf. Für ein paar Nachrichten pro Minute ist das in Ordnung. Publizierst du viele Nachrichten pro Sekunde, halte stattdessen einen Client pro Prozess: lazy erzeugen, einmal loop_start() aufrufen und eine Client-ID pro Prozess vergeben. Derselbe publish.single-Aufruf funktioniert auch in einem Celery-Task – ein guter Ort für Publishes, die die HTTP-Antwort nicht blockieren sollen.

Nachrichten mit Channels in den Browser pushen

Für Live-Telemetrie leitet der Worker jede Nachricht an eine Channels-Gruppe weiter; jeder WebSocket-Consumer in dieser Gruppe empfängt sie. Konfiguriere Channels mit einem Redis-Layer, denn Worker und ASGI-Server sind verschiedene Prozesse:

mysite/settings.py (Auszug)

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)]},
    }
}

Der Consumer tritt der Gruppe bei und leitet Events weiter:

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())]),
})

Zum Schluss erweiterst du on_message im Worker. Der Worker ist synchron, also wickelst du das asynchrone group_send in async_to_sync:

mqtt_listen.py (on_message, erweitert)

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

channel_layer = get_channel_layer()

# in on_message, nach Reading.objects.create(...)
async_to_sync(channel_layer.group_send)(
    "telemetry",
    {"type": "telemetry.message", "data": {"device": device, **data}},
)

Im Browser empfängt new WebSocket("ws://localhost:8000/ws/telemetry/") jetzt jeden Messwert als JSON. Wenn du Authentifizierung ergänzt, wickle den Router in AuthMiddlewareStack und prüfe self.scope["user"] in connect().

Betrieb in Produktion

Behandle den Worker wie jeden anderen Daemon: ein Dienst für den ASGI-Server, einer für mqtt_listen.

/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

Brauchst du mehr als einen Worker für Redundanz oder Durchsatz, starte nicht einfach Kopien: Jede Kopie würde jeden Messwert speichern. Abonniere stattdessen mit einer Shared Subscription ($share/django/testmqtt/django/+/telemetry), damit der Broker die Nachrichten auf die Gruppe verteilt. Shared Subscriptions erklärt die Regeln und welche Broker sie unterstützen.

Den kompletten Weg testen

Starte Redis, python manage.py runserver und python manage.py mqtt_listen in getrennten Terminals.
Öffne den Online-MQTT-Client, verbinde dich mit demselben Broker und publiziere {"temperature": 21.5} auf testmqtt/django/sensor-1/telemetry – oder nutze die CLI:

Testmesswert publizieren

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}'

Prüfe die Datenbank: python manage.py shell -c "from telemetry.models import Reading; print(Reading.objects.first().payload)".
Publiziere einen ungültigen Payload wie hello und prüfe, dass der Worker eine Warnung loggt, statt abzustürzen.
Unterbrich die Broker-Verbindung (oder kurz dein Netzwerk) und prüfe, dass der Worker den Disconnect loggt, sich neu verbindet und weiter Nachrichten speichert.

Django-Topics im Online-Client beobachten

Das Muster gilt für jedes Django-Projekt: Ein überwachter Prozess besitzt die Subscription, Views publizieren kurz, und Channels bringt die Nachrichten zum Browser. Wenn du Frameworks vergleichst: Die Flask- und FastAPI-Varianten dieses Tutorials folgen denselben Regeln mit ihren eigenen Lifecycle-Hooks.

Häufig gestellte Fragen

Wo starte ich den MQTT-Client in einem Django-Projekt?

In einem eigenen Prozess, nicht in AppConfig.ready() oder einer View. Ein eigenes Management Command (python manage.py mqtt_listen), das loop_forever() aufruft, ist die einfachste Lösung: Es startet genau einmal, besitzt die Subscription und lässt sich mit systemd, Docker oder Kubernetes überwachen.

Kann ich das Django-ORM in paho-mqtt-Callbacks verwenden?

Ja. Mit loop_forever() laufen die Callbacks im Haupt-Thread des Commands, synchrone ORM-Aufrufe funktionieren also. Ruf regelmäßig close_old_connections() auf, damit ein lang laufender Worker keine tote Datenbankverbindung behält.

Warum liefert Channels die Nachrichten meines MQTT-Workers nicht aus?

Meistens nutzt das Projekt InMemoryChannelLayer, der nur innerhalb eines Prozesses funktioniert. MQTT-Worker und ASGI-Server sind getrennte Prozesse und brauchen einen gemeinsamen Layer wie channels_redis, der auf dasselbe Redis zeigt.

MQTT mit Django: Hintergrund-Worker und Channels