C# uPLibrary.Networking.M2Mqtt实现MQTT重连后消息重复问题

问题现象 #

使用uPLibrary.Networking.M2Mqtt实现MQTT消息收发：

• 订阅端在线时，所有消息传输正常；
• 订阅端断开后重新连接，能接收断开期间的所有消息，但最后一条消息会重复接收。

原因分析 #

1. Retain标志滥用：发布端每次调用Publish时设置了retain=true（最后一个参数），MQTT服务器会保留最后一条带retain标志的消息。当订阅端重连时，服务器除了推送离线期间缓存的QoS1/2消息（因服务端开启了WithPersistentSessions()持久会话），还会额外推送这条保留的消息，而最后一条离线消息与保留消息是同一条，因此出现重复。
2. 订阅逻辑顺序问题：订阅端代码先执行Subscribe再Connect，虽然uPLibrary允许此操作，但在持久会话场景下，可能导致订阅会话的状态同步异常，加重重复消息的概率。

解决方案 #

方案1：关闭不必要的Retain标志 #

如果业务不需要保留最后一条消息，将发布端的Publish调用中retain参数改为false：

// 修改前
mqttClient.Publish("Application1/NEW_Message", Encoding.UTF8.GetBytes($"{Message}"), MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE, true);

// 修改后
mqttClient.Publish("Application1/NEW_Message", Encoding.UTF8.GetBytes($"{Message}"), MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE, false);

方案2：调整订阅端代码顺序 #

确保先建立连接，再执行订阅操作，避免会话状态同步问题：

// 修改前
mqttClient.Subscribe(new string[] { "Application1/NEW_Message" }, new byte[] { MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE });
mqttClient.Connect(clientName, null, null, false, 60);

// 修改后
mqttClient.Connect(clientName, null, null, false, 60);
mqttClient.Subscribe(new string[] { "Application1/NEW_Message" }, new byte[] { MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE });

方案3：客户端去重处理（可选） #

如果业务必须保留消息，可在订阅端添加去重逻辑，比如记录最后接收的消息内容，收到重复时忽略：

private static string _lastReceivedMessage;
private static void MqttClient_MqttMsgPublishReceived(object sender, uPLibrary.Networking.M2Mqtt.Messages.MqttMsgPublishEventArgs e)
{
    var message = Encoding.UTF8.GetString(e.Message);
    if (message == _lastReceivedMessage)
    {
        // 忽略重复消息
        return;
    }
    _lastReceivedMessage = message;
    Console.WriteLine(message);
}

补充说明 #

服务端开启的WithPersistentSessions()会保留客户端的会话状态，包括未确认的消息和订阅关系，这是离线消息能被接收的前提，但结合retain标志时需注意重复推送的问题。

内容的提问来源于stack exchange，提问作者M.Michael