MQTT消息可送达Broker但客户端接收异常，求排查

问题描述 #

通过ESP（MicroPython）向树莓派上的Mosquitto MQTT Broker发送消息，树莓派本地运行Paho MQTT客户端。使用mosquitto_sub可确认消息稳定送达Broker，但本地客户端的on_message回调仅偶尔触发，尝试设置QoS=1、2均无改善。

ESP端MicroPython代码 #

from umqtt.simple import MQTTClient

broker_ip = "[IP]"
client_name = "[client]"
user = "[user]"
password = "[password]"

def connect_publish(broker, client, topic, message, user, password):
    print("Creating client object...")
    client = MQTTClient(client_id=client_name,
                        server=broker_ip,
                        user=user,
                        password=password,
                        keepalive=60)
    print("Connecting to server...")
    client.connect()
    print("Publishing message")
    client.publish(topic = topic, msg = str(message), qos = 1)
    print("Published", message, "to", topic)
    print("Disconecting from server")
    client.disconnect()

[function to connect to wifi]
[initialize sensor]

while True:
    if [sensor_trigger]:
        connect_publish(broker = broker_ip,
                        client = client_name,
                        topic = b"sensor",
                        message = "on",
                        user = user,
                        password = password)

树莓派端Paho MQTT客户端代码 #

#!/usr/bin/env python3

import paho.mqtt.client as paho
import time

# The callback for when the client receives a CONNACK response from the server.
def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("Connected with result code " + str(rc))
    else: 
        print("Failed to subscribe, code:", rc)
    client.subscribe("sensor", qos = 1)

def on_message(client, userdata, msg):
    print(msg.topic+" "+ msg.payload.decode())
    if msg.payload.decode() == "on":
        if [some further conditions defined in variables below]:
            [do something]


#Initialize the variables for MQTT
BROKER = '[IP]'
#uname and password for mqtt client stored on pi: /etc/mosquitto/passwd
uname = '[user]'
pwd = '[password]'

#Initialize all the paho functions
client = paho.Client('[name]')
client.on_connect = on_connect
client.on_message = on_message
client.username_pw_set(password = pwd, username = uname)

client.connect(host = BROKER)
client.loop_start()

[initialize some more variables]

while True:
    [update some variables]
    time.sleep(0.1)

重点排查方向 #

• 客户端ID冲突：检查ESP端的client_name和树莓派客户端paho.Client('[name]')中的[name]是否一致。如果相同，ESP每次连接Broker都会踢掉树莓派客户端的现有连接，导致树莓派客户端断连期间无法接收消息，这是此类问题的常见原因。
• ESP发布后断开时机：ESP使用QoS=1发布后立即调用client.disconnect()，可能未等待Broker返回PUBACK确认就断开，虽然mosquitto_sub能收到消息，但可能导致Broker对消息的转发逻辑异常。可以在publish后添加短暂延迟（如time.sleep(0.5)）再断开，或改用umqtt.robust库处理重连和确认。
• 树莓派客户端回调阻塞：on_message中的[do something]如果包含耗时操作，会阻塞回调线程（loop_start()启动的是单线程循环），导致后续消息无法及时处理甚至丢失。建议将耗时操作放到单独线程执行，避免阻塞MQTT消息循环。
• Broker日志验证：在树莓派上执行journalctl -u mosquitto查看Broker日志，确认是否存在树莓派客户端断开、订阅失败、消息转发失败的记录，以此判断Broker是否确实将消息推送给了客户端。
• 客户端重连后的订阅逻辑：虽然on_connect中包含subscribe，但如果树莓派客户端意外断开后重连，需确认on_connect是否被正确触发。可以在on_connect中添加日志，验证每次重连都重新执行了订阅操作。
• Keepalive配置：树莓派客户端未显式设置keepalive参数，默认值为60秒。如果客户端的消息循环（loop_start()）因主循环阻塞（如time.sleep(0.1)之外的耗时操作）无法及时发送心跳，Broker可能判定客户端离线，停止转发消息。建议显式设置client.connect(host=BROKER, keepalive=60)，并确保主循环无长时间阻塞。

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