介绍：
在当今互联互通的世界里，设备之间高效可靠的通信至关重要。MQTT（消息队列遥测传输）就是为此而设计的轻量级消息传递协议。本文将探讨 MQTT 是什么、它的优势以及如何在 .NET 框架中设置和实现它。最后，您将对 MQTT 有一个清晰的了解，并通过一个实际示例来帮助您入门。
1.什么是MQTT？
定义和概述：MQTT 代表消息队列遥测传输 (Message Queuing Telemetry Transport)。它是一种轻量级的发布-订阅网络协议，用于在设备之间传输消息。对于需要较少代码占用空间或网络带宽有限的远程位置的连接，MQTT 非常有用。
MQTT的优点：
• 低带宽使用率：旨在最大限度地减少网络带宽使用率，使其成为资源受限环境的理想选择。
• 高效的消息传递：确保以不同的服务质量 (QoS) 级别可靠地传递消息。
• 可扩展性：适用于小型到大型实施，无缝处理数千台设备。
• 易于使用：易于实施并与各种平台集成。
与传统方法的比较：
• 没有 MQTT：HTTP 等传统通信方法更繁重且效率更低，尤其是对于物联网应用而言。
• 使用 MQTT：提供轻量、高效、可靠的消息传递机制，提高性能并减少延迟。
2. 在 .NET Framework 中设置 MQTT
分步指南：
安装 MQTT 包：
1、在 Visual Studio 中打开您的 .NET 项目。
2、在解决方案资源管理器中右键单击您的项目并选择“管理 NuGet 包”。
3、搜索并安装 M2Mqtt 包。
编写 MQTT 客户端代码：创建一个新类来处理 MQTT 客户端操作。初始化 MQTT 客户端，配置连接选项，并连接到代理。
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Text;
using System.Threading.Tasks;
using uPLibrary.Networking.M2Mqtt;
using uPLibrary.Networking.M2Mqtt.Exceptions;
using uPLibrary.Networking.M2Mqtt.Messages;
namespace MQTT
{
public class MqttManager
{
#region Observer Pattern
private HashSet<string> topics = new HashSet<string>();
private HashSet<IMqttListener> listeners = new HashSet<IMqttListener>();
private readonly ushort keepAlivePeriod = 30;
private readonly string BrokerIp;
private readonly int? BrokerPort = null;
private string clientId;
private string Username = null;
private string Password = null;
private MqttClient mqttClient;
public MqttManager(string brokerIp, int? brokerPort = null, string username = null, string password = null)
{
BrokerIp = brokerIp;
BrokerPort = brokerPort;
Username = username;
Password = password;
clientId = RandomClientId();
Debug.WriteLine($"mqttClient clientId: {clientId}");
InitMQTT();
}
public void Register(string topic, IMqttListener listener)
{
if (!topics.Contains(topic))
{
Subscribe(topic);
}
listeners.Add(listener);
}
public void UnRegister(IMqttListener listener)
{
listeners.Remove(listener);
}
public void NotifyListeners(string topic, string message)
{
foreach (IMqttListener listener in listeners)
{
listener.OnMqttMessage(topic, message);
}
}
#endregion
#region Init Mqtt
async private void InitMQTT()
{
Debug.WriteLine($"mqttClient.Try connecting to brokerIp: {BrokerIp}");
mqttClient = new MqttClient(BrokerIp);
mqttClient.MqttMsgPublishReceived += (sender, e) =>
{
string topic = e.Topic;
string message = Encoding.UTF8.GetString(e.Message);
Debug.WriteLine($"mqttClient.MessageReceived. {topic} -> {message}");
NotifyListeners(topic, message);
};
var connectToBroker = new Func<Task>(() =>
{
return Task.Run(() =>
{
byte resultCode = mqttClient.Connect(clientId, Username, Password, true, keepAlivePeriod);
if (resultCode == 0)
{
foreach (var topic in topics)
{
Subscribe(topic);
}
}
});
});
mqttClient.ConnectionClosed += async (s, e) =>
{
await Task.Delay(TimeSpan.FromSeconds(5));
while (!mqttClient.IsConnected)
{
try
{
await connectToBroker();
}
catch (Exception ex)
{
Console.WriteLine($"An error occurred: {ex.Message}");
await Task.Delay(TimeSpan.FromSeconds(5));
}
}
};
while (!mqttClient.IsConnected)
{
try
{
await connectToBroker();
}
catch (Exception ex)
{
Console.WriteLine($"An error occurred: {ex.Message}");
await Task.Delay(TimeSpan.FromSeconds(5));
}
}
Console.WriteLine("Connected to Broker Successfully!");
}
async private void Subscribe(string topic)
{
Debug.WriteLine($"mqttClient.TryToSubscribe -> {topic}");
mqttClient.Subscribe(new string[] { topic }, new byte[] { MqttMsgBase.QOS_LEVEL_AT_MOST_ONCE });
topics.Add(topic);
}
private string RandomClientId()
{
return $"Server-{Guid.NewGuid()}";
}
}
public interface IMqttListener
{
void OnMqttMessage(string topic, string message);
}
}
3.示例实现：处理消息
场景：实现 MQTT 客户端来订阅主题并处理传入的消息。
详细步骤：
• 订阅主题：修改客户端代码以订阅特定主题。
• 处理传入消息：使用事件处理程序来处理传入消息。
public class BoostMqtt : Singleton<BoostMqtt>, IMqttListener
{
public readonly MqttManager MqttManager;
private const string TOPIC = "example_topic";
public static ConcurrentQueue<MqttMessage> MqttMessageQueue = new ConcurrentQueue<MqttMessage>();
private BoostMqtt()
{
string brokerIp = GetBrokerIp();
int? brokerPort = GetBrokerPort();
(string username, string password) = GetUsernamePassword();
MqttManager = new MqttManager(brokerIp, brokerPort, username, password);
MqttManager.Register(TOPIC, this);
StartProcessingMqttMessagesFromQueue();
}
~BoostMqtt()
{
MqttManager?.UnRegister(this);
}
private string GetBrokerIp()
{
try
{
return ConfigurationManager.AppSettings["MqttBrokerIpAddress"];
}
catch (Exception e)
{
Logging.Instance.WriteErrorToLog(e);
return string.Empty;
}
}
private int? GetBrokerPort()
{
try
{
if (int.TryParse(ConfigurationManager.AppSettings["MqttBrokerPort"], out int port))
{
return port;
}
}
catch (Exception e)
{
Logging.Instance.WriteErrorToLog(e);
}
return null;
}
private (string, string) GetUsernamePassword()
{
try
{
string username = ConfigurationManager.AppSettings["MqttUsername"];
string password = ConfigurationManager.AppSettings["MqttPassword"];
return (username, password);
}
catch (Exception e)
{
Logging.Instance.WriteErrorToLog(e);
return (null, null);
}
}
public void StartProcessingMqttMessagesFromQueue()
{
int numThreads = Environment.ProcessorCount;
for (int i = 0; i < numThreads; i++)
{
var thread = new Thread(() =>
{
while (true)
{
try
{
if (MqttMessageQueue.TryDequeue(out MqttMessage message))
{
ProcessMqttMessage(message.Topic, message.Message);
}
}
catch (Exception e)
{
Logging.Instance.WriteErrorToLog(e);
}
Thread.Sleep(1);
}
});
thread.Start();
}
}
public void ProcessMqttMessage(string topic, string original_message)
{
if (topic.Contains("example_topic"))
{
// Process the message here
}
}
}
4. 比较 HiveMQ、RabbitMQ、Kafka 和 MQTT
• HiveMQ：针对物联网进行了优化，具有广泛的监控和管理功能，非常适合大规模工业和商业物联网部署。
• RabbitMQ：提供多种协议和复杂路由功能的灵活性，适用于企业消息传递和微服务架构。
• Kafka：擅长高吞吐量、实时数据处理，适用于大数据和事件流应用。
• MQTT：轻量级且高效，适用于低带宽、高延迟环境，是物联网和移动应用的理想选择。
结论：
本文介绍了 MQTT 的基础知识、优势以及如何在 .NET 框架中设置它。我们还演示了一个使用 MQTT 处理消息的实际示例。利用 MQTT 轻量级且高效的协议，您可以显著改善应用程序中设备之间的通信。
如果您喜欢此文章，请收藏、点赞、评论，谢谢，祝您快乐每一天。
