如何使用Paho MQTT实现客户端与MQTT Broker的自动重连

你当前用自定义try/except递归的方式实现重连虽然可行，但Paho MQTT确实提供了原生的自动重连机制，无需手动处理异常和递归。你之前尝试的loop_forever(retry_first_connection=True)未生效，是因为调用顺序和连接方式不对。

核心问题原因 #

你先调用了阻塞式的connect()方法，一旦首次连接失败会直接抛出异常，程序根本没进入loop_forever()的事件循环，导致retry_first_connection=True参数完全没机会生效。

原生实现方案 #

要实现首次连接失败也自动重连，需要调整两个关键步骤：

• 使用非阻塞式连接：调用connect_async()替代connect()，它只会发起连接请求，不会阻塞等待结果，让后续的事件循环处理连接状态。
• 启用首次连接重试：在loop_forever()中设置retry_first_connection=True，开启首次连接失败后的自动重试逻辑。

Paho会默认使用指数退避策略进行重连（间隔从1秒开始，每次翻倍，直到最大间隔30秒），也可以通过reconnect_delay_set()自定义间隔范围。

修改后的完整代码 #

import paho.mqtt.client as mqtt

def main():
    # 连接成功回调
    def on_connect(client, userdata, flags, reason_code, properties):
        print(f"Connected with result code {reason_code}")
        client.subscribe("sensors")

    # 消息接收回调
    def on_message(client, userdata, msg):
        print(f"{msg.topic} {str(msg.payload)}")

    # 连接失败回调（可选，用于日志输出）
    def on_connect_fail(client, userdata, flags, reason_code, properties):
        print(f"Connection failed with result code {reason_code}")

    # 初始化客户端（使用V2回调API）
    mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv311)
    mqttc.on_connect = on_connect
    mqttc.on_message = on_message
    mqttc.on_connect_fail = on_connect_fail

    # 设置认证与TLS
    mqttc.username_pw_set("username", "password")
    mqttc.tls_set(ca_certs="ca.crt", certfile="client.crt", keyfile="client.key")

    # 自定义重连间隔（可选，默认是1-30秒指数退避）
    # mqttc.reconnect_delay_set(min_delay=5, max_delay=15)

    # 非阻塞式发起连接请求
    mqttc.connect_async("mqtt.client.server.com", port=8883, keepalive=60)

    # 启动事件循环，开启首次连接重试
    mqttc.loop_forever(retry_first_connection=True)

if __name__ == "__main__":
    main()

工作原理说明 #

• connect_async()：仅向客户端发送连接指令，立即返回，不等待Broker响应，后续的连接状态由事件循环处理。
• loop_forever(retry_first_connection=True)：进入阻塞式事件循环，负责处理网络通信、回调触发。当检测到首次连接失败时，会按照预设的重连策略自动重试，直到连接成功。
• 连接成功后，on_connect回调会触发，自动重新订阅主题（符合你原代码的需求）。

内容的提问来源于stack exchange，提问作者Igor L.