MQTT con Django: worker como management command y Channels

La forma fiable de usar MQTT con Django es mantener la conexión MQTT fuera del proceso web. Ejecuta un suscriptor de larga duración como management command propio, deja que escriba en la base de datos con el ORM normal, publica desde las vistas con un cliente de vida corta y usa Django Channels cuando los navegadores necesiten los mensajes en vivo.

Este tutorial construye exactamente eso con paho-mqtt 2.x y después muestra cómo desplegar el worker y probar todo el recorrido de extremo a extremo.

¿Por qué no conectar en apps.py o en una vista?

La primera idea de casi todo el mundo es arrancar un cliente paho en AppConfig.ready(). Funciona en tu portátil y luego falla de formas sutiles:

Dónde arrancas el cliente	Qué sale mal
`AppConfig.ready()`	Se ejecuta en cada proceso que carga Django: `runserver` (dos veces, por el autoreloader), cada worker de Gunicorn, `migrate`, `shell`, los workers de Celery. Cada uno abre una conexión.
Dentro de una vista	Una sesión MQTT nueva por petición HTTP: lento, inunda el broker de CONNECT y falla en cuanto dos peticiones comparten client ID.
Un management command	Exactamente un proceso que arrancas a propósito, fácil de supervisar y reiniciar.

Varios procesos suscritos al mismo topic significa que cada mensaje se procesa varias veces, y si comparten client ID el broker los desconecta en bucle (consulta client ID duplicado). Así que: suscríbete en un worker y publica desde donde lo necesites.

Configuración del proyecto

Instalación

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

Añade telemetry a INSTALLED_APPS y un modelo para las lecturas entrantes:

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

Guarda los ajustes de conexión en settings.py, leídos del entorno:

mysite/settings.py (extracto)

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"

Ejecuta python manage.py makemigrations telemetry && python manage.py migrate.

El worker MQTT como management command

Crea telemetry/management/commands/mqtt_listen.py (tanto management/ como commands/ necesitan un __init__.py vacío):

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()  # CA del sistema, verifica el hostname
        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()  # bloquea y reconecta automáticamente
        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
        # Suscribirse aquí restaura la suscripción tras cada reconexión
        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()  # descarta conexiones de BD más antiguas que CONN_MAX_AGE
        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)

Aquí importan algunos detalles:

Suscríbete en on_connect, no después de connect(). paho reconecta por su cuenta, y suscribirse en el callback restaura la suscripción cada vez.
loop_forever() ejecuta los callbacks en el hilo principal, así que las llamadas síncronas al ORM son seguras. En este proceso no hace falta sync_to_async.
close_old_connections() es lo que Django llama al principio y al final de cada petición. Un worker nunca ve una petición, así que llámalo tú o el proceso acabará fallando con una conexión a la base de datos cerrada tras un reinicio de la BD.
Un client ID único por proceso (hostname + PID). Si prefieres otro patrón, genéralo con el generador de client ID.
No te caigas nunca por un payload malo. Una excepción no controlada en un callback detiene el bucle. Para más sobre parseo y validación, consulta enviar JSON por MQTT en Python.

Arráncalo con python manage.py mqtt_listen.

Publicar desde vistas y tareas

Publicar es otro problema: es breve, ocasional y lo disparan las peticiones. Para frecuencias bajas (comandos, notificaciones) la opción correcta más sencilla es paho.mqtt.publish.single, que conecta, publica, espera el handshake de QoS y desconecta:

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 con el bundle de CA del sistema
    )
    return JsonResponse({"queued": True})

publish.single abre una conexión nueva cada vez, lo que cuesta un handshake TCP + TLS por llamada. Eso está bien para unos pocos mensajes por minuto. Si publicas muchos mensajes por segundo, mantén un cliente por proceso: créalo de forma perezosa, llama a loop_start() una vez y dale un client ID por proceso. La misma llamada a publish.single funciona también dentro de una tarea de Celery, que es un buen lugar para publicaciones que no deben bloquear la respuesta HTTP.

Enviar mensajes al navegador con Channels

Para mostrar la telemetría en vivo, haz que el worker reenvíe cada mensaje a un grupo de Channels; cada consumer WebSocket de ese grupo lo recibe. Configura Channels con una capa Redis, porque el worker y el servidor ASGI son procesos diferentes:

mysite/settings.py (extracto)

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

El consumer se une al grupo y reenvía los eventos:

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

Por último, amplía on_message en el worker. El worker es síncrono, así que envuelve el group_send asíncrono con async_to_sync:

mqtt_listen.py (on_message ampliado)

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

channel_layer = get_channel_layer()

# dentro de on_message, después de Reading.objects.create(...)
async_to_sync(channel_layer.group_send)(
    "telemetry",
    {"type": "telemetry.message", "data": {"device": device, **data}},
)

En el navegador, new WebSocket("ws://localhost:8000/ws/telemetry/") recibe ahora cada lectura como JSON. Si añades autenticación, envuelve el router en AuthMiddlewareStack y comprueba self.scope["user"] en connect().

Ejecutarlo en producción

Trata el worker como cualquier otro daemon: un servicio para el servidor ASGI y otro para 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

Si necesitas más de un worker por redundancia o rendimiento, no arranques simplemente copias: cada copia guardaría todas las lecturas. Suscríbete en su lugar con una shared subscription ($share/django/testmqtt/django/+/telemetry) para que el broker reparta los mensajes entre el grupo. Shared subscriptions explica las reglas y qué brokers las soportan.

Probar todo el recorrido

Arranca Redis, python manage.py runserver y python manage.py mqtt_listen en terminales separadas.
Abre el cliente MQTT online, conéctate al mismo broker y publica {"temperature": 21.5} en testmqtt/django/sensor-1/telemetry, o usa la CLI:

Publicar una lectura de prueba

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

Comprueba la base de datos: python manage.py shell -c "from telemetry.models import Reading; print(Reading.objects.first().payload)".
Publica un payload inválido como hello y confirma que el worker registra un aviso en lugar de morir.
Reinicia la conexión con el broker (o corta brevemente tu red) y confirma que el worker registra la desconexión, reconecta y sigue guardando mensajes.

Observa los topics de Django en el cliente online

El patrón es el mismo para cualquier proyecto Django: un proceso supervisado es dueño de la suscripción, las vistas publican brevemente y Channels lleva los mensajes a los navegadores. Si estás comparando frameworks, las versiones de este tutorial para Flask y FastAPI siguen las mismas reglas con sus propios hooks de ciclo de vida.

Preguntas frecuentes

¿Dónde debo iniciar el cliente MQTT en un proyecto Django?

En su propio proceso, no en AppConfig.ready() ni en una vista. Un management command propio (python manage.py mqtt_listen) que llama a loop_forever() es la opción más sencilla: arranca una sola vez, es dueño de la suscripción y se puede supervisar con systemd, Docker o Kubernetes.

¿Puedo usar el ORM de Django dentro de los callbacks de paho-mqtt?

Sí. Con loop_forever() los callbacks se ejecutan en el hilo principal del comando, así que las llamadas síncronas al ORM funcionan. Llama a close_old_connections() con regularidad para que un worker de larga duración no conserve una conexión a la base de datos ya muerta.

¿Por qué Channels no entrega los mensajes de mi worker MQTT?

Casi siempre el proyecto usa InMemoryChannelLayer, que solo funciona dentro de un proceso. El worker MQTT y el servidor ASGI son procesos distintos, así que necesitan una capa compartida como channels_redis apuntando al mismo Redis.

MQTT con Django: workers en segundo plano y Channels