使用paho-mqtt无法接收MQTT消息的排查求助

阿华AIGC实验室

2026-5-14

问题描述

我正在开发一款需接收MQTT消息的Python应用，采用paho-mqtt库。配置用户名/密码、on_message回调函数后，订阅主题并连接至Broker，随后启动loop_forever()。代码如下：

import paho.mqtt.client as mqtt
MQTT_TOPIC = "my_topic"
BROKER_ENDPOINT = "my_broker_url"
BROKER_PORT = my_port
BROKER_USERNAME = "my_username"
BROKER_PASSWORD = "my_password"
mqtt_client = mqtt.Client()
def on_message(client, userdata, message):
    print("Message Recieved from broker: " + message.payload)
def main():
    mqtt_client.username_pw_set(username=BROKER_USERNAME, password=BROKER_PASSWORD)
    mqtt_client.on_message = on_message
    mqtt_client.connect(BROKER_ENDPOINT, BROKER_PORT)
    mqtt_client.subscribe(MQTT_TOPIC)
    mqtt_client.loop_forever()
if __name__ == '__main__':
    main()

我使用mosquitto_pub命令测试：

mosquitto_pub -h my_broker_url -p my_port -u my_username -P my_password -t 'my_topic' -m "Hello World"

但应用无法接收任何消息，请问我哪里操作有误？

解决方案

你的代码存在几个关键问题，导致无法正常接收消息，我来逐一拆解并给出修正方案：

1. `message.payload` 类型错误，直接拼接会触发异常

message.payload返回的是字节对象，不是字符串，直接和字符串拼接会抛出TypeError——这会导致on_message回调崩溃，客户端可能静默停止处理消息（你看不到任何输出，因为异常被内部捕获了）。需要先将字节解码为字符串：

print("Message Recieved from broker: " + message.payload.decode("utf-8"))

2. 订阅操作应放在`on_connect`回调中

你在connect后直接调用subscribe，但如果客户端与Broker的连接未完全建立，或者后续出现重连，订阅会丢失。正确的做法是设置on_connect回调，在连接成功确认后再执行订阅，同时能直观看到连接状态：

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

记得在main函数中绑定这个回调：

mqtt_client.on_connect = on_connect

3. 缺少日志回调，无法排查潜在问题

你没有设置日志回调，无法知晓连接是否成功、是否有认证失败等隐藏问题。可以添加一个简单的日志函数辅助排查：

def on_log(client, userdata, level, buf):
    print(f"MQTT Log: {buf}")

修正后的完整代码

import paho.mqtt.client as mqtt

MQTT_TOPIC = "my_topic"
BROKER_ENDPOINT = "my_broker_url"
BROKER_PORT = 1883  # 替换为实际端口，比如1883（默认）或8883（加密）
BROKER_USERNAME = "my_username"
BROKER_PASSWORD = "my_password"

mqtt_client = mqtt.Client()

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("Connected to MQTT Broker successfully")
        # 连接确认后再订阅主题
        client.subscribe(MQTT_TOPIC)
    else:
        print(f"Failed to connect, return code: {rc}")
        # 返回码对应问题：rc=4是用户名/密码错误，rc=5是未授权，可针对性排查

def on_message(client, userdata, message):
    # 将字节payload解码为UTF-8字符串
    payload_str = message.payload.decode("utf-8")
    print(f"Message Received from broker: {payload_str}")

def on_log(client, userdata, level, buf):
    # 打印日志，辅助排查连接、订阅问题
    print(f"MQTT Log: {buf}")

def main():
    mqtt_client.username_pw_set(username=BROKER_USERNAME, password=BROKER_PASSWORD)
    # 绑定所有回调函数
    mqtt_client.on_connect = on_connect
    mqtt_client.on_message = on_message
    mqtt_client.on_log = on_log
    
    # 连接Broker
    mqtt_client.connect(BROKER_ENDPOINT, BROKER_PORT)
    # 启动永久消息循环
    mqtt_client.loop_forever()

if __name__ == '__main__':
    main()