使用MQTTnet库接收保留消息不全问题求助

问题背景 #

使用MQTTnet库基于MQTT V5协议开发，配置如下：

• QoS等级：qos=2
• 客户端会话配置：WithCleanStart(false).WithSessionExpiryInterval(3600)
• 服务器：默认配置（max_queued_messages=1000）的Mosquitto

出现的异常现象：

• 无保留消息场景：订阅端断开后，发布端发送2000条消息，订阅端重启后接收1000条，符合服务器队列限制预期。
• 第三方工具（MQTT Explorer）场景：发布2000个不同主题的单条保留消息，订阅端重启后可接收全部2000条。
• 自研C#程序场景：同样发布2000个不同主题的保留消息，订阅端重启后仅接收1000-1040条；但订阅端在线时可正常接收全部消息。

已排除服务器配置问题，聚焦MQTTnet库参数排查。

核心排查方向与解决方案 #

1. 订阅端本地消息队列限制 #

MQTTnet客户端默认的MaxPendingMessages参数值为1000，当服务器推送大量保留消息时，若客户端本地队列容量不足，超出部分会被自动丢弃，这是最可能的原因。

解决方案：
初始化客户端时，显式设置MaxPendingMessages为大于2000的值，确保能容纳所有待接收的保留消息：

var clientOptions = new MqttClientOptionsBuilder()
    .WithTcpServer("your-mqtt-server")
    .WithClientId("fixed-subscriber-client-id")
    .WithCleanStart(false)
    .WithSessionExpiryInterval(3600)
    .WithMaxPendingMessages(2500) // 设置为大于2000的数值
    .Build();

2. 发布端保留消息配置验证 #

确认自研程序发布消息时，是否正确设置了保留标志，且QoS2的发布流程完整（等待服务器ACK）。

验证代码示例：

// 确保每条消息都开启Retain标志
for (int i = 0; i < 2000; i++)
{
    var message = new MqttApplicationMessageBuilder()
        .WithTopic($"unique-topic/{i}")
        .WithPayload("test-content")
        .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce)
        .WithRetainFlag(true) // 必须开启此标志
        .Build();
    
    // QoS2需确保发布完成后再发送下一条，避免服务器未处理完成
    await mqttClient.PublishAsync(message, CancellationToken.None);
}

3. 订阅端RetainHandling参数配置 #

MQTT V5的RetainHandling参数决定会话恢复时如何处理保留消息，若配置错误可能导致部分消息不被推送。

正确配置示例：

var subscribeOptions = new MqttClientSubscribeOptionsBuilder()
    .WithTopicFilter(filter => filter
        .WithTopic("unique-topic/#")
        .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce)
        .WithRetainHandling(MqttRetainHandling.SendAtSubscribe)) // 确保订阅时发送保留消息
    .Build();

await mqttClient.SubscribeAsync(subscribeOptions);

4. 客户端会话一致性验证 #

确保订阅端重启前后使用相同的客户端ID，且CleanStart=false配置未被意外覆盖，否则会创建新会话，无法恢复原有的保留消息推送逻辑。

额外验证步骤 #

• 查看Mosquitto服务器日志，确认2000条保留消息已被成功存储，且订阅端重启时服务器确实推送了全部消息。
• 检查订阅端消息接收回调是否存在阻塞逻辑，若回调处理过慢会导致本地队列堆积，触发消息丢弃机制。

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