Python Paho MQTT客户端非确定性崩溃问题排查

问题描述 #

按照Paho官方文档编写基础MQTT订阅客户端，与Mosquitto服务端配合时会随机崩溃，无法正常处理消息。但使用mosquitto_sub -u simon -P password -t "home/sensors/dht22" -i spy命令可正常接收消息，说明服务端和消息源无异常。

原始代码 #

from paho.mqtt import client as mqtt_client

broker = '127.0.0.1'
port = 1883
topic = "home/sensors/dht22"
client_id = f'python-mqtt-dht22'

PASSWORD = "password"

def connect_mqtt():
    def on_connect(client, userdata, flags, rc):
        if rc == 0:
            print("Connected to MQTT Broker!")
        else:
            print("Failed to connect, return code %d\n", rc)

    client = mqtt_client.Client(mqtt_client.CallbackAPIVersion.VERSION1, client_id)

    client.username_pw_set("simon", PASSWORD)
    client.on_connect = on_connect
    client.connect(broker, port)
    return client

def subscribe(client: mqtt_client):
    def on_message(client, userdata, msg):
        payload = msg.payload.decode()
        print(f"Received `{payload}` from `{msg.topic}` topic")
        handle_msg(payload)

    client.subscribe(topic)
    client.on_message = on_message

def run():
    client = connect_mqtt()
    client.loop_start()
    subscribe(client)
    client.loop_forever()

if __name__ == '__main__':
    run()

触发的随机错误 #

错误1：struct格式解析异常 #

Exception in thread paho-mqtt-client-python-mqtt-dht22:
Traceback (most recent call last):
  File "/usr/lib/python3.12/threading.py", line 1073, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.12/threading.py", line 1010, in run
    self._target(*self._args, **self._kwargs)
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 4523, in _thread_main
    self.loop_forever(retry_first_connection=True)
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 2297, in loop_forever
    rc = self._loop(timeout)
         ^^^^^^^^^^^^^^^^^^^
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 1686, in _loop
    rc = self.loop_read()
         ^^^^^^^^^^^^^^^^
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 2100, in loop_read
    rc = self._packet_read()
         ^^^^^^^^^^^^^^^^^^^
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 3142, in _packet_read
    rc = self._packet_handle()
         ^^^^^^^^^^^^^^^^^^^^^
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 3808, in _packet_handle
    return self._handle_publish()
           ^^^^^^^^^^^^^^^^^^^^^^
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 4099, in _handle_publish
    (slen, packet) = struct.unpack(pack_format, self._in_packet['packet'])
struct.error: bad char in struct format

错误2：ReasonCode KeyError异常 #

Exception in thread paho-mqtt-client-python-mqtt-dht22:
Traceback (most recent call last):
  File "/usr/lib/python3.12/threading.py", line 1073, in _bootstrap_inner
Traceback (most recent call last):
  File "/home/simon/code/mqtt_sink.py", line 76, in <module>
    self.run()
  File "/usr/lib/python3.12/threading.py", line 1010, in run
    self._target(*self._args, **self._kwargs)
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 4523, in _thread_main
    run()
    self.loop_forever(retry_first_connection=True)
  File "/home/simon/code/mqtt_sink.py", line 71, in run
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 2297, in loop_forever
    client.loop_forever()
    rc = self._loop(timeout)
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 2297, in loop_forever
         ^^^^^^^^^^^^^^^^^^^
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 1686, in _loop
    rc = self._loop(timeout)
         ^^^^^^^^^^^^^^^^^^^
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 1686, in _loop
    rc = self.loop_read()
    rc = self.loop_read()
         ^^^^^^^^^^^^^^^^
         ^^^^^^^^^^^^^^^^
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 2100, in loop_read
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 2100, in loop_read
    rc = self._packet_read()
    rc = self._packet_read()
         ^^^^^^^^^^^^^^^^^^^
         ^^^^^^^^^^^^^^^^^^^
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 3142, in _packet_read
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 3142, in _packet_read
    rc = self._packet_handle()
    rc = self._packet_handle()
         ^^^^^^^^^^^^^^^^^^^^^
         ^^^^^^^^^^^^^^^^^^^^^
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 3816, in _packet_handle
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 3816, in _packet_handle
    self._handle_suback()
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 4053, in _handle_suback
    self._handle_suback()
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/client.py", line 4053, in _handle_suback
    reasoncodes = [ReasonCode(SUBACK >> 4, identifier=c) for c in granted_qos]
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/reasoncodes.py", line 129, in __init__
    reasoncodes = [ReasonCode(SUBACK >> 4, identifier=c) for c in granted_qos]
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/reasoncodes.py", line 129, in __init__
    self.getName()  # check it's good
    self.getName()  # check it's good
    ^^^^^^^^^^^^^^
    ^^^^^^^^^^^^^^
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/reasoncodes.py", line 172, in getName
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/reasoncodes.py", line 172, in getName
    return self.__getName__(self.packetType, self.value)
    return self.__getName__(self.packetType, self.value)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/reasoncodes.py", line 140, in __getName__
  File "/home/simon/.avenv/mqtt/lib/python3.12/site-packages/paho/mqtt/reasoncodes.py", line 140, in __getName__
    raise KeyError(identifier)
    raise KeyError(identifier)
KeyError: 48

问题根源 #

代码中同时调用了client.loop_start()和client.loop_forever()，导致两个独立的MQTT事件循环同时运行：

• loop_start()会启动后台线程运行loop_forever()
• 主线程再次调用loop_forever()，两个循环同时处理MQTT数据包，引发数据包解析混乱，进而出现struct格式错误或ReasonCode异常

修复方案 #

移除client.loop_start()，仅保留主线程的loop_forever()，确保只有一个事件循环处理MQTT消息。

修复后的代码 #

from paho.mqtt import client as mqtt_client

broker = '127.0.0.1'
port = 1883
topic = "home/sensors/dht22"
client_id = f'python-mqtt-dht22'

PASSWORD = "password"

def connect_mqtt():
    def on_connect(client, userdata, flags, rc):
        if rc == 0:
            print("Connected to MQTT Broker!")
        else:
            print("Failed to connect, return code %d\n", rc)

    client = mqtt_client.Client(mqtt_client.CallbackAPIVersion.VERSION1, client_id)

    client.username_pw_set("simon", PASSWORD)
    client.on_connect = on_connect
    client.connect(broker, port)
    return client

def subscribe(client: mqtt_client):
    def on_message(client, userdata, msg):
        payload = msg.payload.decode()
        print(f"Received `{payload}` from `{msg.topic}` topic")
        handle_msg(payload)

    client.subscribe(topic)
    client.on_message = on_message

def run():
    client = connect_mqtt()
    subscribe(client)
    client.loop_forever()

if __name__ == '__main__':
    run()

验证结果 #

修复后客户端可稳定运行，正常接收并处理MQTT消息，无崩溃现象。

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