Paho-MQTT客户端断开代码16后无法重连问题求助

问题描述 #

使用Python Paho-MQTT客户端（已简化脚本逻辑与数据库访问代码），客户端运行数小时后会以错误码16（MQTT_ERR_KEEPALIVE）断开连接，之后进入无限重连循环但无法成功（未收到CONNACK）。

客户端代码 #

import json
import logging
import time

from paho.mqtt import client as mqtt

logging.basicConfig(level=logging.INFO)
mqtt_logger = logging.getLogger("MqttCon")


def on_connect(client, userdata, flags, rc):
    mqtt_logger.info("Device connected with result code: " + str(rc))
    client.subscribe("some_topic/#", qos=0)


def on_disconnect(client, userdata, rc):
    mqtt_logger.info("Device disconnected with result code: " + str(rc))


def pub_values(client, base_topic, values):
    msg = json.dumps(values)
    target_topic = base_topic
    mqtt_logger.info("Sending message to wheel...")
    client.publish(target_topic, msg, qos=0)


def on_message(client, userdata, msg):
    """
    Input messages format (can be one or list):
    {
        "tag_id": "string",
        "value": "int or float",
        "timestamp": "string format YYYY-MM-DDTHH:MM:SS.F"
    }
    """
    try:
        info = json.loads(str(msg.payload.decode("utf-8")))
        mqtt_logger.info(msg.topic)
        pub_msg = {"tag_ids": [], "timestamp": ""}
        mqtt_logger.info(f"Received message topic is {msg.topic}")
        if isinstance(info, list):
            # some logic
        else:
            # some logic, writing values to InfluxDB
        pub_values(client=client, base_topic="some_topic", values=pub_msg)
    except Exception as e:
        mqtt_logger.error(str(e))


def on_log(client, userdata, level, buf):
    mqtt_logger.info(f"SYSTEM: {buf}")


if __name__ == "__main__":
    client = mqtt.Client(protocol=mqtt.MQTTv311)
    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.on_message = on_message
    client.on_log = on_log
    client.connect_async('<my_mqtt_url>', port=1883)

    client.loop_start()
    try:
        while True:
            time.sleep(0)
    except KeyboardInterrupt:
        pass
    finally:
        mqtt_logger.warning("Stopping client")
        client.loop_stop()
        client.disconnect()

断开时的日志信息 #

INFO:MqttCon:Received message topic is telemetry/MESPIV/batch
INFO:MqttCon:Sending message to wheel...
INFO:MqttCon:SYSTEM: Sending PUBLISH (d0, q0, r0, m4228), 'b'router/newdata'', ... (11336 bytes)
INFO:MqttCon:Sending message to wheel...
INFO:MqttCon:SYSTEM: Sending PUBLISH (d0, q0, r0, m4229), 'b'router/newdata'', ... (11336 bytes)
INFO:MqttCon:Device disconnected with result code: 16
INFO:MqttCon:Device disconnected with result code: 16
INFO:MqttCon:SYSTEM: Sending CONNECT (u0, p0, wr0, wq0, wf0, c1, k60) client_id=b''
INFO:MqttCon:SYSTEM: Sending CONNECT (u0, p0, wr0, wq0, wf0, c1, k60) client_id=b''
# more Sending CONNECT

最初版本的问题 #

最初使用常规connect方法的版本也存在相同问题：

if __name__ == "__main__":
    # setting mqtt client
    client.connect('<my_mqtt_url>', port=1883)
    client.loop_forever()
    client.disconnect()

补充信息 #

• 重启客户端容器可恢复，但数小时后仍会出现同样的断开重连失败问题；
• 故障发生前未挂起/恢复虚拟机，客户端与 broker 均运行在虚拟机上；
• broker日志中看不到客户端重连尝试，但在客户端断开时能看到连接中断记录；
• 未使用Wireshark抓包；
• 客户端日志中存在代码16断开后成功重连的情况，但部分场景下无法重连。

临时解决方案 #

修改on_disconnect回调，当rc=16时退出程序：

def on_disconnect(client, userdata, rc):
    mqtt_logger.info("Device disconnected with result code: " + str(rc))
    if rc == 16:
        exit(1)

同时给容器设置restart: always，断开后重启容器恢复，但重启期间会丢失broker数据。

需求 #

寻求该问题的根本解决办法。

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