MQTT com Django: worker como management command e Channels

O jeito confiável de usar MQTT com Django é manter a conexão MQTT fora do processo web. Rode um subscriber de longa duração como management command próprio, deixe-o gravar no banco com o ORM normal, publique a partir das views com um cliente de vida curta e use Django Channels quando os navegadores precisarem das mensagens ao vivo.

Este tutorial constrói exatamente isso com paho-mqtt 2.x e depois mostra como fazer o deploy do worker e testar todo o caminho de ponta a ponta.

Por que não conectar no apps.py ou em uma view?

A primeira ideia da maioria das pessoas é iniciar um cliente paho em AppConfig.ready(). Funciona no seu notebook e depois quebra de formas sutis:

Onde você inicia o cliente	O que dá errado
`AppConfig.ready()`	Roda em todo processo que carrega o Django: `runserver` (duas vezes, por causa do autoreloader), cada worker do Gunicorn, `migrate`, `shell`, workers do Celery. Cada um abre uma conexão.
Dentro de uma view	Uma nova sessão MQTT por requisição HTTP: lento, inunda o broker de CONNECTs e falha assim que duas requisições compartilham o client ID.
Um management command	Exatamente um processo que você inicia de propósito, fácil de supervisionar e reiniciar.

Vários processos assinando o mesmo topic significa que cada mensagem é processada várias vezes, e se eles compartilham o client ID o broker os derruba em loop (veja client ID duplicado). Então: assine em um worker e publique de onde precisar.

Configuração do projeto

Instalação

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

Adicione telemetry ao INSTALLED_APPS e um model para as leituras recebidas:

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

Mantenha as configurações de conexão no settings.py, lidas do ambiente:

mysite/settings.py (trecho)

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"

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

O worker MQTT como management command

Crie telemetry/management/commands/mqtt_listen.py (tanto management/ quanto commands/ precisam de um __init__.py vazio):

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()  # CAs do sistema, verifica o 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()  # bloqueia e reconecta automaticamente
        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
        # Assinar aqui restaura a assinatura após cada reconexão
        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 conexões de BD mais antigas 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)

Alguns detalhes importam aqui:

Assine em on_connect, não depois de connect(). O paho reconecta sozinho, e assinar no callback restaura a assinatura todas as vezes.
loop_forever() roda os callbacks na thread principal, então chamadas síncronas ao ORM são seguras. Neste processo não é preciso sync_to_async.
close_old_connections() é o que o Django chama no início e no fim de cada requisição. Um worker nunca vê uma requisição, então chame você mesmo, ou o processo acabará falhando com uma conexão de banco fechada depois de um restart do banco.
Um client ID único por processo (hostname + PID). Se preferir outro padrão, gere com o gerador de client ID.
Nunca caia por causa de um payload ruim. Uma exceção não tratada em um callback para o loop. Para mais sobre parsing e validação, veja enviar JSON via MQTT em Python.

Inicie com python manage.py mqtt_listen.

Publicando a partir de views e tasks

Publicar é outro problema: é curto, ocasional e disparado por requisições. Para taxas baixas (comandos, notificações) a opção correta mais simples é paho.mqtt.publish.single, que conecta, publica, espera o handshake de QoS e 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 com o bundle de CAs do sistema
    )
    return JsonResponse({"queued": True})

publish.single abre uma conexão nova a cada chamada, o que custa um handshake TCP + TLS por vez. Isso é aceitável para algumas mensagens por minuto. Se você publica muitas mensagens por segundo, mantenha um cliente por processo: crie-o de forma lazy, chame loop_start() uma vez e dê a ele um client ID por processo. A mesma chamada publish.single também funciona dentro de uma task do Celery, um bom lugar para publicações que não devem bloquear a resposta HTTP.

Enviando mensagens ao navegador com Channels

Para mostrar a telemetria ao vivo, faça o worker encaminhar cada mensagem para um grupo do Channels; todo consumer WebSocket desse grupo a recebe. Configure o Channels com uma camada Redis, porque o worker e o servidor ASGI são processos diferentes:

mysite/settings.py (trecho)

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

O consumer entra no grupo e encaminha os 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 fim, estenda o on_message do worker. O worker é síncrono, então envolva o group_send assíncrono com async_to_sync:

mqtt_listen.py (on_message estendido)

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

channel_layer = get_channel_layer()

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

No navegador, new WebSocket("ws://localhost:8000/ws/telemetry/") agora recebe cada leitura como JSON. Se você adicionar autenticação, envolva o router em AuthMiddlewareStack e verifique self.scope["user"] em connect().

Rodando em produção

Trate o worker como qualquer outro daemon: um serviço para o servidor ASGI e outro para o 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

Se você precisar de mais de um worker por redundância ou vazão, não inicie simplesmente cópias: cada cópia gravaria todas as leituras. Em vez disso, assine com uma shared subscription ($share/django/testmqtt/django/+/telemetry) para que o broker distribua as mensagens entre o grupo. Shared subscriptions explica as regras e quais brokers as suportam.

Testando todo o caminho

Inicie o Redis, python manage.py runserver e python manage.py mqtt_listen em terminais separados.
Abra o cliente MQTT online, conecte ao mesmo broker e publique {"temperature": 21.5} em testmqtt/django/sensor-1/telemetry, ou use a CLI:

Publicar uma leitura de teste

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

Verifique o banco: python manage.py shell -c "from telemetry.models import Reading; print(Reading.objects.first().payload)".
Publique um payload inválido como hello e confirme que o worker registra um aviso em vez de morrer.
Reinicie a conexão com o broker (ou derrube sua rede por um instante) e confirme que o worker registra a desconexão, reconecta e continua gravando mensagens.

Acompanhe os topics do Django no cliente online

O padrão é o mesmo para qualquer projeto Django: um processo supervisionado é dono da assinatura, as views publicam rapidamente e o Channels leva as mensagens aos navegadores. Se você está comparando frameworks, as versões deste tutorial para Flask e FastAPI seguem as mesmas regras com seus próprios hooks de ciclo de vida.

Perguntas frequentes

Onde devo iniciar o cliente MQTT em um projeto Django?

Em um processo próprio, não em AppConfig.ready() nem em uma view. Um management command próprio (python manage.py mqtt_listen) que chama loop_forever() é a opção mais simples: inicia uma única vez, é dono da assinatura e pode ser supervisionado por systemd, Docker ou Kubernetes.

Posso usar o ORM do Django dentro dos callbacks do paho-mqtt?

Sim. Com loop_forever() os callbacks rodam na thread principal do comando, então chamadas síncronas ao ORM funcionam. Chame close_old_connections() regularmente para que um worker de longa duração não mantenha uma conexão de banco de dados já morta.

Por que o Channels não entrega as mensagens do meu worker MQTT?

Na maioria das vezes o projeto usa InMemoryChannelLayer, que só funciona dentro de um processo. O worker MQTT e o servidor ASGI são processos separados, então precisam de uma camada compartilhada como channels_redis apontando para o mesmo Redis.

MQTT com Django: workers em segundo plano e Channels