MqttNet - MQTT 客户端无法接收 MQTT 服务器发布的所有消息

Question

当客户端订阅主题时，我从 MQTT 服务器发布了 1000 条消息。 MQTT 客户端未收到某些消息。这是一个错误还是我执行的有误？

这里是服务器和客户端配置示例。

服务器配置

using MQTTnet;
using MQTTnet.Server;
using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace MqttNetServerExample
{
    class Program
    {
        private static IMqttServer _mqttServer;

        static void Main(string[] args)
        {

            // Configure MQTT server.
            var optionsBuilder = new MqttServerOptionsBuilder().WithConnectionBacklog(100)
                                                               .WithDefaultEndpointPort(1884);

            // Define a mqttServer
            _mqttServer = new MqttFactory().CreateMqttServer();

            // Message arrived configuration
            _mqttServer.UseApplicationMessageReceivedHandler(async e =>
            {
                if (Encoding.UTF8.GetString(e.ApplicationMessage.Payload)== "Test Message")
                {
                    Console.WriteLine("subscription message received");
                    Console.WriteLine("Simulating messages...");
                    await Simulate();
                }
            });

            // When a new client connected
            _mqttServer.UseClientConnectedHandler(e =>
            {
                Console.WriteLine("***** CLIENT CONNECTED : " + e.ClientId + " *******");
            });

            // Start the mqtt server
            _mqttServer.StartAsync(optionsBuilder.Build());

            Console.ReadLine();
        }

        private static async Task PublishMessage(string message)
        {
            // Create mqttMessage
            var mqttMessage = new MqttApplicationMessageBuilder()
                                .WithTopic("mqttServerTopic")
                                .WithPayload(message)
                                .WithAtLeastOnceQoS()
                                .WithRetainFlag(false)
                                .WithDupFlag(false)
                                .Build();

            // Publish the message asynchronously
            var result = await _mqttServer.PublishAsync(mqttMessage, CancellationToken.None);

            if(result.ReasonCode == MQTTnet.Client.Publishing.MqttClientPublishReasonCode.Success)
                Console.WriteLine("Message published : " + message);
        }

        private static async Task Simulate()
        {
            for (int i = 0; i < 1000; i++)
            {
                var message = "This is a message from server " + i.ToString();
                await PublishMessage(message);
            }
        }
    }
}

客户端配置

using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Client.Options;
using MQTTnet.Client.Subscribing;
using System;
using System.Text;
using System.Threading;

namespace MqttNetClientExample
{
    class Program
    {
        private static IMqttClient _mqttClient;

        static void Main(string[] args)
        {
            // Create client
            _mqttClient = new MqttFactory().CreateMqttClient();
            var options = new MqttClientOptionsBuilder().WithClientId("MqttClient")
                                                        .WithTcpServer("localhost", 1884)
                                                        .Build();
            // When client connected to the server
            _mqttClient.UseConnectedHandler(async e =>
            {
                // Subscribe to a topic
                MqttClientSubscribeResult subResult = await _mqttClient.SubscribeAsync(new MqttClientSubscribeOptionsBuilder()
                                                                       .WithTopicFilter("mqttServerTopic")
                                                                       .Build());
                // Sen a test message to the server
                PublishMessage("Test Message");
            });

            // When client received a message from server
            _mqttClient.UseApplicationMessageReceivedHandler(e =>
            {
                Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
            });

            // Connect ot server
            _mqttClient.ConnectAsync(options, CancellationToken.None);

            Console.Read();
        }

        private static async void PublishMessage(string message)
        {
            // Create mqttMessage
            var mqttMessage = new MqttApplicationMessageBuilder()
                                .WithTopic("mqttServerTopic")
                                .WithPayload(message)
                                .WithExactlyOnceQoS()
                                .Build();

            // Publish the message asynchronously
            await _mqttClient.PublishAsync(mqttMessage, CancellationToken.None);
        }
    }
}

结果输出样本

这是客户端应用程序的示例输出：

+ Payload = Test Message
+ Payload = This is a message from server 0
+ Payload = This is a message from server 1
+ Payload = This is a message from server 2
+ Payload = This is a message from server 3
+ Payload = This is a message from server 4
+ Payload = This is a message from server 5
+ Payload = This is a message from server 6
+ Payload = This is a message from server 7
+ Payload = This is a message from server 8
+ Payload = This is a message from server 9
+ Payload = This is a message from server 10
+ Payload = This is a message from server 11
+ Payload = This is a message from server 13
+ Payload = This is a message from server 14
+ Payload = This is a message from server 23
+ Payload = This is a message from server 53
+ Payload = This is a message from server 54
+ Payload = This is a message from server 55
+ Payload = This is a message from server 56
+ Payload = This is a message from server 57
+ Payload = This is a message from server 83
+ Payload = This is a message from server 105
+ Payload = This is a message from server 120
+ Payload = This is a message from server 138
+ Payload = This is a message from server 139
+ Payload = This is a message from server 140
+ Payload = This is a message from server 141
+ Payload = This is a message from server 172
+ Payload = This is a message from server 192
+ Payload = This is a message from server 207
+ Payload = This is a message from server 218
+ Payload = This is a message from server 236
+ Payload = This is a message from server 258
+ Payload = This is a message from server 278
+ Payload = This is a message from server 302
     .
     .
     .
     .
+ Payload = This is a message from server 999

如您所见，一些消息丢失了。我怎样才能毫无损失地获取所有消息？谢谢

Answer 1

根据评论，问题出在 MqttNet Broker 的配置上。

设置MaxPendingMessagesPerClient默认为250，这意味着当特定客户端有250条消息等待传递时，当另一条消息到达时，队列中的第一条消息将被丢弃。

您的应用程序发送消息的速度超过了接收方处理消息的速度，因此队列建立得很快并且消息丢失了。最后 250 条消息毫无问题地到达，因为没有发送任何新消息（因此代理不需要丢弃更多消息）。

要解决此问题，您可以使用 WithMaxPendingMessagesPerClient 例如

var optionsBuilder = new MqttServerOptionsBuilder()                
   .WithMaxPendingMessagesPerClient(1000);  

注意：在生产中用于此设置的值将取决于您的具体要求。您需要确保订阅者能够以消息产生的速度消费消息（否则代理队列将不断增长）。