如何检查Python Paho MQTT连接是否成功建立？附代码示例

首先纠正原代码的一个关键顺序问题：username_pw_set 必须在 connect 之前调用，否则认证信息不会被生效。

方法1：利用paho的回调函数捕获连接状态 #

paho MQTT客户端提供了on_connect回调函数，当客户端与broker完成连接（或连接失败）时会自动触发，通过回调参数里的rc可以直接判断连接结果：

• 0: 连接成功
• 1: 协议版本不兼容
• 2: 客户端ID无效
• 3: Broker服务不可用
• 4: 用户名/密码错误
• 5: 无连接权限

修改后的代码（添加日志与回调验证）：

import paho.mqtt.client as paho
import logging
import time

# 配置日志格式与级别
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        logging.info("MQTT连接成功")
    else:
        logging.error(f"MQTT连接失败，错误码: {rc}")

# 初始化客户端
client = paho.Client()
# 设置认证信息（必须在connect之前）
client.username_pw_set(username=USERNAME_BROKER, password=PASSWORD_BROKER)
# 绑定连接状态回调
client.on_connect = on_connect

# 发起连接请求
client.connect(MQTT_BROKER, 1883, 60)
# 启动网络循环，处理回调事件（非阻塞）
client.loop_start()

# 短暂等待，确保连接回调被触发
time.sleep(1)

# 仅在连接成功时发布消息（可选）
if client.is_connected():
    client.publish(id, value, 0)
    logging.info("消息发布完成")

# 断开连接前停止循环
client.loop_stop()
client.disconnect()

方法2：同步等待连接结果 #

如果需要同步确认连接状态，可以用connect_async异步发起连接，结合loop_start和is_connected()方法判断：

import paho.mqtt.client as paho
import logging
import time

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

client = paho.Client()
client.username_pw_set(username=USERNAME_BROKER, password=PASSWORD_BROKER)

# 异步发起连接
client.connect_async(MQTT_BROKER, 1883, 60)
# 启动网络循环
client.loop_start()

# 最多等待5秒，检查连接状态
time.sleep(5)
if client.is_connected():
    logging.info("MQTT连接成功")
    client.publish(id, value, 0)
else:
    logging.error("MQTT连接超时，未成功建立连接")

client.loop_stop()
client.disconnect()

注意事项 #

• connect是阻塞调用，但仅负责发送连接请求，不会等待broker的确认响应，必须通过loop_start/loop_forever启动网络循环才能处理回调事件。
• 日志记录可以快速定位问题，比如认证失败、Broker地址错误、网络不通等场景。

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