前期准备

本篇文章采用的Mqtt 2.8.5类库    Visual Studio2019

引用mqtt类库

对着新建的项目右键，管理Nuget程序包，搜索MQTTNet选择版本2.8.5点击安装

 了解MQTT常用事件以及属性

Connected     连接成功事件
Disconnected     断开连接事件
ApplicationMessageReceived    订阅事件
ClientId    客户端ID（唯一）
Topic    订阅主题（一个客户端可以订阅一个到多个主题，接收到的主题消息由订阅接收消息事件处理）

了解Qos 消息等级

mqtt消息等级称Qos分三个等级
MqttQualityOfServiceLevel.AtMostOnce    最多一次 MqttQualityOfServiceLevel.AtLeastOnce    最少一次 MqttQualityOfServiceLevel.ExactlyOnce    只有一次
注意：当qos=MqttQualityOfServiceLevel.AtLeastOnce或MqttQualityOfServiceLevel.ExactlyOnce的时候,就遇到问题了. 想让设备重新连接,获取到mqtt保存的消息,需要设置cleansession=false,就是设备每次登陆,都是用同一个session,而没有发出去的消息,是放在内存里面的,跟session关联,这里问题就大了.消息堆积,内存就挂了.所以有些 消息,需要自己业务逻辑处理或定期清理防止内存爆满

下面展示一些 内联代码片。

// mqtt客户端
private static MqttClient mqttClient;
//ClientId，同SessionId性质一样，是客户端的身份识别唯一标识
public string ClientId="";
static void Main(string[] args)
{
	Console.WriteLine("===启动客户端MQTT===");
	Console.Write("请输入ClientId:");
    ClientId = Console.ReadLine().ToLower().Trim();
    ConnectMqttService();
    while (true)
    {
    	if (mqttClient == null)
		{
			ConnectMqttService();
		}
	}
}
public static void ConnectMqttService()
{
	try
	{
		if (mqttClient == null || !mqttClient.IsConnected)
		{
			mqttClient = new MqttFactory().CreateMqttClient() as MqttClient;
			//连接成功
            mqttClient.Connected += MqttClient_Connected;
            //断开连接
            mqttClient.Disconnected += MqttClient_Disconnected;
            //订阅客户端或服务端发布的消息
            mqttClient.ApplicationMessageReceived += MqttClient_ApplicationMessageReceived;
			var options = new MqttClientOptions
			{
				ClientId = ClientId,
				CleanSession = true,
				ChannelOptions = new MqttClientTcpOptions
				{
					Server = "mqtt服务端ip地址",
					Port = Convert.ToInt32(端口号)
				}
             };
             Task.WaitAll(mqttClient.ConnectAsync(options));
             MqttSubscribe();
    	}
    }
	catch (Exception ex)
	{
		Console.WriteLine($"客户端ID标识为Client100001的消息连接MQTT服务器异常:" + ex.ToString());
	}
}

//MQTT发布消息 content消息内容
public static void MqttClientPublish(String content)
{
	var msg = new MqttApplicationMessage()
	{
		Topic = "$Topic" + ClientId,
		Payload = Encoding.UTF8.GetBytes(content),
		//QOS 消息等级
		QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce,
		Retain = false
	};
	mqttClient.PublishAsync(msg);
}

//订阅MQTT消息
public static void MqttSubscribe()
{
	mqttClient.SubscribeAsync(new List<TopicFilter> {
		new TopicFilter("$Topic"+ClientId,MqttQualityOfServiceLevel.AtMostOnce)
	});
}

//订阅接收消息事件
private static void MqttClient_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
{
	//订阅内容
	String Payload = System.Text.Encoding.Default.GetString(e.ApplicationMessage.Payload);
	Console.WriteLine($"===订阅消息：{e.ClientId}内容:{Payload}===");
}

//mqtt断开连接两种形式 长时间订阅未接收到消息和服务端断开连接
private static void MqttClient_Disconnected(object sender, MqttClientDisconnectedEventArgs e)
{
	Console.WriteLine($"====已断开MQTT服务器，断开时间{DateTime.Now.ToString()}====");
}

private static void MqttClient_Connected(object sender, MqttClientConnectedEventArgs e)
{
	Console.WriteLine($"=====连接MQTT服务器成功,连接时间{DateTime.Now.ToString()}====");
}