如何在Django视图中通过paho.mqtt向MQTT broker发布消息

• 你的__init__.py代码存在变量未定义问题：仅导入了mqtt模块，没有实例化client对象就直接调用client.loop_start()，运行必然报错
• run()方法中的client.loop_forever()是阻塞方法，直接在Django主线程调用会卡住整个服务，无法响应其他HTTP请求
• 没有复用MQTT客户端实例，每次调用发布都新建连接会导致资源浪费、连接异常等问题

1. 调整mqtt.py代码，实现全局单例客户端 #

from paho.mqtt import client as mqtt_client

MQTT_TOPIC = "#topic#"
MQTT_BROKER_HOST = "####the host##"
MQTT_BROKER_PORT = 1883
MQTT_USERNAME = "#username#"
MQTT_PASSWORD = "#psswd#"

# 全局单例客户端
_mqtt_client = None

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("Connected to MQTT Broker!")
    else:
        print(f"Failed to connect, return code {rc}")

def init_mqtt_client():
    global _mqtt_client
    if _mqtt_client is not None:
        return _mqtt_client
    _mqtt_client = mqtt_client.Client(MQTT_TOPIC)
    _mqtt_client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
    _mqtt_client.on_connect = on_connect
    _mqtt_client.connect(MQTT_BROKER_HOST, MQTT_BROKER_PORT)
    # 启动非阻塞的后台循环，不要用loop_forever
    _mqtt_client.loop_start()
    return _mqtt_client

def publish_msg(msg: str = "test test"):
    global _mqtt_client
    if _mqtt_client is None:
        raise RuntimeError("MQTT client not initialized, call init_mqtt_client first")
    result = _mqtt_client.publish(MQTT_TOPIC, msg)
    # 可选：检查发布结果
    status = result[0]
    if status != 0:
        print(f"Failed to send message to topic {MQTT_TOPIC}")

2. 配置Django启动时自动初始化MQTT客户端 #

不要把初始化逻辑写在__init__.py里，Django官方推荐启动时执行的初始化逻辑放在AppConfig的ready()方法中，避免dev模式下自动重载导致重复初始化：
找到你对应Django应用的apps.py文件，修改配置如下：

from django.apps import AppConfig
import os

class YourAppNameConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'your_app_name' # 替换成你的应用名

    def ready(self):
        # 避免Django dev模式下reloader进程重复初始化MQTT客户端
        if os.environ.get('RUN_MAIN') == 'true':
            from .mqtt import init_mqtt_client
            init_mqtt_client()

然后确认你的应用在settings.py的INSTALLED_APPS中注册的是带AppConfig的路径，比如：

INSTALLED_APPS = [
    # ... 其他应用
    'your_app_name.apps.YourAppNameConfig',
]

3. 视图中调用发布逻辑 #

直接导入mqtt模块的publish_msg方法即可调用，不需要执行run()：

from django.http import HttpResponse
from .mqtt import publish_msg

def test_mqtt_publish(request):
    publish_msg("你要发布的消息内容")
    return HttpResponse("消息发布成功")

• 生产环境用uWSGI/Gunicorn部署时，需要开启线程支持，否则MQTT的后台循环会无法运行。以Gunicorn为例，启动命令添加--enable-threads参数即可
• 如果你的部署配置了多个worker进程，每个worker都会初始化一个独立的MQTT连接，如果不需要高并发发布，可以将worker数设置为1，或者把MQTT发布逻辑拆成独立的异步任务用Celery执行

内容的提问来源于stack exchange，提问作者Jackson