MqttNet 通讯

MqttNet 通讯

MQTT,IBM发明的物联网通讯协议基于tcp ip , 收集传感器上的数据。

下图理解：  broker 这里有很多消息，根据主题不同来进行区分，它这里可以保管所有连过来的客户端的数据，然后客户端，通过订阅broker它有的主题进行获取数据。

学习网址：https://github.com/chkr1011/MQTTnet/wiki/Client

broker网址 代理：http://www.mqtt-dashboard.com/

开发(只需客户端)：

vs2015

1.添加引用本文MQTTnet2.8.4（管理NuGet程序包）

2.界面

3.代码

using MQTTnet;

using MQTTnet.Client;

using System;

using System.Collections.Generic;

using System.ComponentModel;

using System.Data;

using System.Drawing;

using System.Linq;

using System.Text;

using System.Threading.Tasks;

using System.Windows.Forms;

namespace server

{

    public partial class Form1 : Form

    {

        public Form1()

        {

            InitializeComponent();

        }

        public static IMqttClient mqttClient = null;

        public static IMqttClientOptions options = null;

        private void Form1_Load(object sender, EventArgs e)

        {

            //实例化对象

            var factory = new MqttFactory();

            mqttClient = factory.CreateMqttClient();

            mqttClient.Connected += SubscribeTopic;

            mqttClient.Disconnected += MqttClient_Disconnected;

            mqttClient.ApplicationMessageReceived += Receive;

            //配置参数

            //options = new MqttClientOptionsBuilder()

            //    .WithClientId(Guid.NewGuid().ToString().Substring(0, 5))

            //    .WithTcpServer("broker.hivemq.com")

            //    .WithCredentials("bud", "%spencer%")

            //    .WithTls()

            //    .WithCleanSession()

            //    .Build();

            options = new MqttClientOptionsBuilder().WithWebSocketServer("broker.hivemq.com:8000/mqtt").Build();

            //连接

            Task.Run(async () => { await ConnectMqtt(); });

        }

        public async Task ConnectMqtt()

        {

            try

            {

                MqttClientConnectResult x = await mqttClient.ConnectAsync(options);

            }

            catch (Exception ex)

            {

                Invoke((new Action(() =>

                {

                    textBox2.Text = $"连接到MQTT服务器失败！" + Environment.NewLine + ex.Message + Environment.NewLine;

                })));

            }

        }

        //接收消息

        public async void Receive(object sender, MqttApplicationMessageReceivedEventArgs e)

        {

            try

            {

                Invoke((new Action(() =>

                {

                    //textBox2.AppendText("### RECEIVED APPLICATION MESSAGE 接收消息 ###");

                    textBox2.AppendText($"Topica(主题) = {e.ApplicationMessage.Topic}" + "\t\n");

                    textBox2.AppendText($"Payloada(内容) = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}" + "\t\n");

                    //textBox2.AppendText($"+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}");

                    //textBox2.AppendText($"+ Retain = {e.ApplicationMessage.Retain}");

                })));

            }

            catch (Exception ex)

            {

                Invoke((new Action(() =>

                {

                    textBox2.Text = ex.Message;

                })));

            }

        }

        //连接成功

        public async void SubscribeTopic(object sender, EventArgs e)

        {

            Invoke((new Action(() => { label2.Text = "连接成功"; })));

        }

        /// <summary>

        /// 连接失败

        /// </summary>

        /// <param name="sender"></param>

        /// <param name="e"></param>

        public async void MqttClient_Disconnected(object sender, EventArgs e)

        {

            Invoke((new Action(() =>

            {

                textBox2.AppendText("连接失败！" + Environment.NewLine);

            })));

            //重新连接

            await Task.Delay(TimeSpan.FromSeconds());

            try

            {

                await mqttClient.ConnectAsync(options);

                Invoke((new Action(() =>

                {

                    textBox2.AppendText("连接成功！");

                })));

            }

            catch

            {

                Invoke((new Action(() =>

                {

                    textBox2.AppendText("连接失败！");

                })));

            }

        }

        /// <summary>

        /// 订阅消息

        /// </summary>

        /// <param name="sender"></param>

        /// <param name="e"></param>

        private void button1_Click(object sender, EventArgs e)

        {

            string topic = textBox1.Text;

            mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).Build());

            textBox2.AppendText("### 订阅" + topic + "成功 ###\t\n");

        }

        /// <summary>

        /// 释放资源

        /// </summary>

        /// <param name="sender"></param>

        /// <param name="e"></param>

        private void button2_Click(object sender, EventArgs e)

        {

            mqttClient.Dispose();

            textBox2.AppendText("### 断开连接###\t\n");

        }

        /// <summary>

        /// 发布一个主题内容

        /// </summary>

        /// <param name="sender"></param>

        /// <param name="e"></param>

        private void button3_Click(object sender, EventArgs e)

        {

            var message = new MqttApplicationMessageBuilder()

                .WithTopic(textBox3.Text)

                .WithPayload(textBox4.Text)

                .WithExactlyOnceQoS()

                .WithRetainFlag()

                .Build();

             mqttClient.PublishAsync(message);

        }

    }

}

4.运行效果

这些数据就是根据你的主题从http://www.mqtt-dashboard.com/   代理取来，你也可以发布主题的内容，别人也可以订阅你的主题，取数据。