MQTT代理宕机期间的数据丢失问题及解决方案咨询

核心问题分析 #

• 默认会话不持久：你的发布者和订阅者都使用了默认的clean_session=True，MQTT代理不会保留客户端的会话状态。当代理宕机重启后，之前未完成传递的消息会被直接丢弃，订阅者也无法获取离线期间的消息。
• Retain消息的局限：Retain机制仅能存储指定主题的最后一条消息，无法保存多条历史数据，这就是你只能收到最后一条消息的原因。

解决方案 #

要完整避免消息丢失，需要从会话持久化、客户端配置两方面调整：

1. 启用持久化会话 #

发布者和订阅者创建客户端时都要设置clean_session=False，让代理为客户端维护持久化会话：

• 对发布者：代理会缓存未确认的QoS1/QoS2消息，恢复后自动重新传递。
• 对订阅者：代理会保存订阅关系，并缓存订阅者离线期间到达的QoS1/QoS2消息，待订阅者重连后全部推送。

2. 优化客户端逻辑 #

• 发布者：确保客户端在连接断开后能自动重连，Paho客户端默认会处理QoS1/QoS2的消息缓存。
• 订阅者：重连后重新确认订阅（即使代理保留了订阅，再次订阅能确保QoS级别匹配）。

3. 代理配置（以Mosquitto为例） #

确保代理开启持久化功能，重启后能恢复保存的会话数据：
在Mosquitto配置文件中添加/确认：

persistence true
persistence_location /var/lib/mosquitto/

修改后的代码 #

发布者代码 #

import paho.mqtt.client as mqtt

broker_address = "localhost"
port = 1883

# 创建持久化会话客户端
client = mqtt.Client(clean_session=False)

def on_connect(client, userdata, flags, rc):
    print(f"连接成功，返回码: {rc}")

client.on_connect = on_connect

client.connect(broker_address, port, 60)
client.loop_start()

try:
    while True:
        message = input("输入要发布的消息（输入'q'退出）: ")
        if message == 'q':
            break
        # 发布QoS2消息，无需retain（除非需要保留最后一条）
        publish_result = client.publish("mytestingtopic", message, qos=2)
        # 等待客户端确认消息已缓存（可选，确保消息不会丢失）
        publish_result.wait_for_publish()
        print(f"消息 '{message}' 已提交缓存")
finally:
    client.loop_stop()
    client.disconnect()

订阅者代码 #

import paho.mqtt.client as mqtt
import time

broker_address = "localhost"
port = 1883

# 创建持久化会话客户端
client = mqtt.Client(clean_session=False)

def on_connect(client, userdata, flags, rc):
    print(f"连接成功，返回码: {rc}")
    # 重连后重新订阅，确保QoS级别匹配
    client.subscribe("mytestingtopic", qos=2)

def on_message(client, userdata, msg):
    print(f"收到消息: {msg.topic} -> {msg.payload.decode('utf-8')}")

client.on_connect = on_connect
client.on_message = on_message

while True:
    try:
        client.connect(broker_address, port, 60)
        client.loop_start()
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        client.loop_stop()
        client.disconnect()
        break
    except Exception as e:
        print(f"错误: {str(e)}")
        client.loop_stop()
        client.disconnect()
        time.sleep(5)

关键注意事项 #

• QoS级别：必须使用QoS1或QoS2，QoS0无法保证离线消息的传递。
• clean_session=False：这是实现消息持久化的核心，不要遗漏。
• Retain的使用：如果不需要保留主题的最后一条消息，可以去掉retain=True，持久化会话已经能覆盖离线消息的推送需求。

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