Django中MQTT订阅onMessage回调执行两次问题求助

问题描述 #

搭建了Mosquitto MQTT Broker，并创建了一个简单的Django应用订阅$SYS/broker/uptime主题，但on_message回调中的print语句会执行两次，控制台输出重复内容。

代码如下：

from django.apps import AppConfig
from threading import Thread
import paho.mqtt.client as mqtt


class MqttClient(Thread):
    def __init__(self, broker, port, timeout, topics):
        super(MqttClient, self).__init__()
        self.client = mqtt.Client()
        self.broker = broker
        self.port = port
        self.timeout = timeout
        self.topics = topics
        self.total_messages = 0

    #  run method override from Thread class

    def run(self):
        self.connect_to_broker()

    def connect_to_broker(self):
        self.client.on_connect = self.on_connect
        self.client.on_message = self.on_message
        self.client.connect(self.broker, self.port, self.timeout)
        self.client.loop_forever()

    # The callback for when a PUBLISH message is received from the server.

    def on_message(self, client, userdata, msg):
        self.total_messages = self.total_messages + 1
        print(str(msg.payload) + "Total: {}".format(self.total_messages))


    # The callback for when the client receives a CONNACK response from the server.

    def on_connect(self, client, userdata, flags, rc):
        #  Subscribe to a list of topics using a lock to guarantee that a topic is only subscribed once
        for topic in self.topics:
            client.subscribe(topic)


class AppMqtteConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'app_mqtt'

    def ready(self):
        MqttClient("localhost", 1883, 60, ["$SYS/broker/uptime"]).start()

原因 #

这是Django开发模式的特性导致的：当用python manage.py runserver启动项目时，Django会启动两个进程——一个主进程，一个负责代码变更自动重载的辅助进程。两个进程都会执行AppConfig.ready()方法，因此创建了两个独立的MQTT客户端实例，它们都订阅了$SYS/broker/uptime主题，每当Broker发布消息时，两个客户端的on_message回调都会被触发，导致打印两次。

解决方案 #

方案1：仅在主进程中启动MQTT客户端 #

修改ready()方法，通过检查环境变量RUN_MAIN判断是否为主进程，只有主进程才启动客户端：

import os
from django.apps import AppConfig
from threading import Thread
import paho.mqtt.client as mqtt

# ... 其他代码保持不变 ...

class AppMqtteConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'app_mqtt'

    def ready(self):
        # 仅在主进程中启动MQTT客户端
        if os.environ.get('RUN_MAIN') == 'true':
            MqttClient("localhost", 1883, 60, ["$SYS/broker/uptime"]).start()

方案2：禁用Django的自动重载 #

启动项目时添加--noreload参数，Django只会启动一个进程：

python manage.py runserver --noreload

两种方法都能避免创建重复的MQTT客户端，解决on_message回调执行两次的问题。

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