mqttnet+emqx实现消息订阅与发布_mqttnet emqx

mqttnet+emqx实现消息订阅与发布

1 在Windows操作系统下使用EMQX

1.1 部署前准备

• 1. 首先进入官网https://www.emqx.io/zh/downloads?os=Windows直接下载EMQX的Windows版本。
• 2. 准备一个连接EMQX服务器的客户单，官方下载地址https://www.emqx.com/zh/products/mqttx，直接下载即可。本次学习准备自己写一个MQTT客户端，所以采用C#+.net3.1+mqttnett写了个简单的客户端进行使用。
     做好上述准备，则进人主题。

1.2 部署EMQX

   将下载下来的压缩包emqx-5.0.9-windows-amd64.tar.gz进行解压，以解压到D:\emqx-5.0.9-windows-amd64为例

• 1. 首先进入路径搜索栏，输入CMD回车进入当前目录下的命令窗口，如下图：
     
     
• 2. 做完1步骤后那么我们就可以启动EMQX啦，输入命令： ./bin/emqx start弹出以下界面的内容就说明启动成功啦!
     

如果想确定是否启动成功，也可以通过.\bin\emqx ping进行测试，如果显示下面图中内容，那就说明确实是启动成功啦，不要怀疑！

开发模式下可以使用 console 命令在控制台启动 EMQX，该模式可以实时查看 EMQX 启动和运行输出日志信息：./bin/emqx console。

• 3. 进行完第2步后，则可以打开浏览器输入127.0.0.1:18083地址后，可以进入EMQX的网页版管理界面：用户名：admin 密码：public（默认可修改）
     

登录后界面如下：

如果觉得右侧图标看起来不方便，可以点击最下面的切换按钮，展示实际的菜单名称，如下图：

可以通过Connections进行当前连接的客户端查看，如下图

如果觉得英文看的不舒服，可以修改显示的语言，步骤如下：

其他功能页签不做具体介绍，可以去官网自己学习，再次附使用说明的官网地址https://www.emqx.io/docs/zh/v5.0/#产品优势。
完成以上步骤，则一个完整的本地EMQX就可以使用啦！

1.3 开发MQTT客户端

• 1. 首先导入nuget包，安装最新稳定版就可以，如下图：
     
• 2. 由于只是简单的测试测试一下mqttNet的使用流程，所有没有做复杂的界面，具体界面如下：
     

前端代码比较简单，这里就不做分享啦，仅简单介绍下服务器的连接，主题发布，消息发送，订阅主题，获取发布者的消息等。
2.1 服务器连接与断开

        private async void btnConnect_Click(object sender, RoutedEventArgs e)
        {
            PublisherClient = mqttFactory.CreateMqttClient() as MqttClient;
            var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer(this.txtServiceAddress.Text.Trim(),Convert.ToInt32(this.txtServicePort.Text.Trim())).Build();
            mqttClientOptions.ClientId = "HITE_publisher";
            mqttClientOptions.Credentials = new MqttClientCredentials("admin", Encoding.ASCII.GetBytes("wang12520"));
            mqttClientOptions.CleanSession = true;
            //保持期
            mqttClientOptions.KeepAlivePeriod = TimeSpan.FromSeconds(10000.5);
            try
            {
                await PublisherClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
                this.InfoList.Items.Insert(0, $"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}：与服务器建立连接成功");
            }
            catch(Exception ex)
            {
                this.InfoList.Items.Insert(0, $"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}：与服务器建立连接失败，失败原因：{ex.Message}");
            }
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

        private void btnDisConnect_Click(object sender, RoutedEventArgs e)
        {
            try
            {
                PublisherClient.DisconnectAsync();
                this.InfoList.Items.Insert(0, $"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}：与服务器断开连接成功");
            }
            catch (Exception ex)
            {

                this.InfoList.Items.Insert(0, $"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}：与服务器断开连接失败，失败原因：{ex.Message}");
            }

        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2.2 主题发布

        private async void btnPublish_Click(object sender, RoutedEventArgs e)
        {
            var applicationMessage = new MqttApplicationMessageBuilder()
                        .WithTopic(this.txtToptcName.Text.Trim())
                        .WithPayload(this.txtToptcInfo.Text.Trim())
                        .Build();
            try
            {

                await PublisherClient.PublishAsync(applicationMessage);
                this.InfoList.Items.Insert(0, $"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}：发布主题{applicationMessage.Topic}成功");
            }
            catch (Exception ex)
            {
                this.InfoList.Items.Insert(0, $"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}：发布主题{applicationMessage.Topic}失败，失败原因：{ex.Message}");
            }
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

2.3 消息发送

       private async  void btnSendInfo_Click(object sender, RoutedEventArgs e)
       {
           try
           {
               await PublisherClient.PublishStringAsync(this.txtToptcName.Text.Trim(), this.txtSendInfo.Text);
               this.InfoList.Items.Insert(0, $"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}：主题{this.txtToptcName.Text.Trim()}发布消息“{this.txtSendInfo.Text}”成功");
           }
           catch(Exception ex)
           {
               this.InfoList.Items.Insert(0, $"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}：主题{this.txtToptcName.Text.Trim()}发布消息“{this.txtSendInfo.Text}”失败，失败原因：{ex.Message}");
           }
       }
1
2
3
4
5
6
7
8
9
10
11
12

2.4 主题订阅

        private async void btnSubscriber_Click(object sender, RoutedEventArgs e)
        {
            try
            {
                SubscriberClient = mqttFactory.CreateMqttClient() as MqttClient;


                var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", 1883).Build();

                SubscriberClient.ApplicationMessageReceivedAsync += e =>
                {
                    Application.Current.Dispatcher.Invoke(() =>
                    {
                        this.InfoList.Items.Insert(0, $"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}：收到主题“{this.txtToptcName.Text.Trim()}”的消息“{UTF8Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}”");
                    });
                    return Task.CompletedTask;
                }; 

                await SubscriberClient.ConnectAsync(mqttClientOptions, CancellationToken.None);

                var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
                    .WithTopicFilter(
                        f =>
                        {
                            f.WithTopic(this.txtToptcName.Text.Trim());
                        })
                    .Build();

                await SubscriberClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
                this.InfoList.Items.Insert(0, $"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}：订阅主题{this.txtToptcName.Text.Trim()}成功");
            }
            catch (Exception ex)
            {
                this.InfoList.Items.Insert(0, $"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}：订阅主题{this.txtToptcName.Text.Trim()}失败，失败原因：{ex.Message}");
            }
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

以上就是全部的代码，EMQX在windows基础使用篇，仅供初学者参考，大佬可绕，如果需要完整代码可以，可以留下邮箱，看到后我会发给你的。后续会将EMQX部署到linux进行记录。