MQTT协议测试工具及核心代码

这篇具有很好参考价值的文章主要介绍了MQTT协议测试工具及核心代码。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

基于MQTTnet做了个MQTT协议测试程序,本程序包括服务的和两个客户端,方便在不引入外部支撑的情况下测试MQTT协议。
测试软件界面如下
mqtt测试工具,MQTT,物联网,网络协议

1:启动MQTT服务

在Server区域,
启动服务可以选择MQTT服务使用的端口、用户名、密码。
如有有客户端需要连接服务,需要把这些信息告诉他们。
右边tab页有启动工程的详细日志。
mqtt测试工具,MQTT,物联网,网络协议

2:MQTT客户端

在该测试程序中实现了两个客户端,方便测试使用,两个客户端功能万千一致,以客户端1为例介绍配置过程中的相关参数。
mqtt测试工具,MQTT,物联网,网络协议
首先是需要连接的MQTT服务器IP和端口,这些信息一般有服务端提供,本程序直接连接本机的服务端,因此IP配置为127.0.0.1,端口和服务器端口一致12345。
其次,根据MQTT协议,连接服务器时需要提供用户名、密码用于鉴权认证。
最后,需要一个终端唯一标识,即客户标识,该标识在一个服务器下需要保证唯一。
到此和MQTT服务器连接相关参数就完备了,可以连接服务器了。
Qos是遗嘱消息服务质量等级,取值即含义如下
mqtt测试工具,MQTT,物联网,网络协议
RETAIN: 保留标志位,如果为1,服务器存储最新一条RETAIN消息,以便分发给新的订阅者

MQTT是发布订阅模式的,客户端如果要收到消息需要先订阅对应的主题,订阅主题数量是没有限制的。
例如:client1订阅主题client1/sub、client2/sub
client2订阅主题client2/sub
然后在已订主题的下拉框中就可以看到本客户端订阅的主题。
如何要取消对应的主题,在下拉列表中先择,然后点击取消,就会取消订阅。
发布主题需要设定主题和发布内容。
例如用client1发布:
主题:client2/sub
内容:新信息。
例如用client2发布:
主题:client1/sub
内容:test
根据刚才设置的订阅信息,两个客户端收到的数据如下:
mqtt测试工具,MQTT,物联网,网络协议
我们可以看到主题:client2/sub在两个客户端都收到了。
自此,测试功能基本就介绍完了。

3:程序功能实现

该程序是基于MQTTnet库最的。
因为要实现服务端和客户端。因此创建连个类:MQTTServer、MQTTClients,这部分代码也是在网上找的,但功能实现比较完备,所有我就直接使用了。
服务端MQTTServer

public class MQTTServer
    {
        public StringBuilder sb = new StringBuilder("");
        public MqttServer mqttServer = null;
        public bool MqttServer_runflag = false;

        public async void StartMqttServer(int port = 12345, string user = "123", string passwd = "123")
        {
            if (mqttServer == null)
            {
                var optionsBuilder = new MqttServerOptionsBuilder()
                .WithDefaultEndpoint().WithDefaultEndpointPort(port).WithConnectionValidator(
                    c =>
                    {
                        var currentUser = user;
                        var currentPWD = passwd;

                        if (currentUser == null || currentPWD == null)
                        {
                            c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
                            return;
                        }

                        if (c.Username != currentUser)
                        {
                            c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
                            return;
                        }

                        if (c.Password != currentPWD)
                        {
                            c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
                            return;
                        }

                        c.ReasonCode = MqttConnectReasonCode.Success;
                    }).WithSubscriptionInterceptor(
                    c =>
                    {
                        c.AcceptSubscription = true;
                    }).WithApplicationMessageInterceptor(
                    c =>
                    {
                        c.AcceptPublish = true;
                    });

                mqttServer = new MqttFactory().CreateMqttServer() as MqttServer;
                mqttServer.StartedHandler = new MqttServerStartedHandlerDelegate(OnMqttServerStarted);
                mqttServer.StoppedHandler = new MqttServerStoppedHandlerDelegate(OnMqttServerStopped);

                mqttServer.ClientConnectedHandler = new MqttServerClientConnectedHandlerDelegate(OnMqttServerClientConnected);
                mqttServer.ClientDisconnectedHandler = new MqttServerClientDisconnectedHandlerDelegate(OnMqttServerClientDisconnected);
                mqttServer.ClientSubscribedTopicHandler = new MqttServerClientSubscribedHandlerDelegate(OnMqttServerClientSubscribedTopic);
                mqttServer.ClientUnsubscribedTopicHandler = new MqttServerClientUnsubscribedTopicHandlerDelegate(OnMqttServerClientUnsubscribedTopic);
                mqttServer.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(OnMqttServer_ApplicationMessageReceived);


                await mqttServer.StartAsync(optionsBuilder.Build());
            }

        }

        public async void StopMqttServer()
        {
            await mqttServer?.StopAsync();
        }

        public async void PublishMqttTopic(string topic, string payload)
        {
            var message = new MqttApplicationMessage()
            {
                Topic = topic,
                Payload = Encoding.UTF8.GetBytes(payload)
            };
            await mqttServer.PublishAsync(message);
        }
       
        public void OnMqttServerStarted(EventArgs e)
        {
            sb.Append("MQTT服务启动完成!\r\n");
        }
        public void OnMqttServerStopped(EventArgs e)
        {
            sb.Append("MQTT服务停止完成!\r\n");
        }
        public void OnMqttServerClientConnected(MqttServerClientConnectedEventArgs e)
        {
            sb.Append($"客户端[{e.ClientId}]已连接\r\n");
        }

        public void OnMqttServerClientDisconnected(MqttServerClientDisconnectedEventArgs e)
        {
            sb.Append($"客户端[{e.ClientId}]已断开连接!\r\n");
            //PublishMqttTopic("client/Disconnected", $"客户端[{ e.ClientId}]已断开连接");
        }

        public void OnMqttServerClientSubscribedTopic(MqttServerClientSubscribedTopicEventArgs e)
        {
            //sb.Append($"客户端[{e.ClientId}]已成功订阅主题[{e.TopicFilter}]!\r\n");
            sb.Append($"客户端[{e.ClientId}]已成功订阅主题[{e.TopicFilter.Topic}]!\r\n");
        }
        public void OnMqttServerClientUnsubscribedTopic(MqttServerClientUnsubscribedTopicEventArgs e)
        {
            sb.Append($"客户端[{e.ClientId}]已成功取消订阅主题[{e.TopicFilter}]!\r\n");
        }

        public void OnMqttServer_ApplicationMessageReceived(MqttApplicationMessageReceivedEventArgs e)
        {
            sb.Append($"客户端[{e.ClientId}]>> 主题:{e.ApplicationMessage.Topic} \r\n");
            //Console.WriteLine($"客户端[{e.ClientId}]>> 主题:{e.ApplicationMessage.Topic} 负荷:{Encoding.UTF8.GetString(e.ApplicationMessage.Payload)} Qos:{e.ApplicationMessage.QualityOfServiceLevel} 保留:{e.ApplicationMessage.Retain}");
        }
    }

客户端MQTTClients

 public class MQTTClients
    {
        private MqttClient mqttClient = null;
        public StringBuilder sb = new StringBuilder("");

        public async Task<MqttClientConnectResultCode> ClientStart(string tcpServer, int tcpPort, string mqttUser, string mqttPassword, 
            String ClientId = "01001")
        {
            try
            {
                var mqttFactory = new MqttFactory();
                //https://www.cnblogs.com/zhaoqm999/p/12960677.html
                var options = new MqttClientOptions
                {
                    ClientId = ClientId,
                    ProtocolVersion = MQTTnet.Formatter.MqttProtocolVersion.V311,
                    ChannelOptions = new MqttClientTcpOptions
                    {
                        Server = tcpServer,
                        Port = tcpPort
                    },
                    WillDelayInterval = 10,
                    WillMessage = new MqttApplicationMessage()
                    {
                        Topic = $"LastWill/121313",
                        Payload = Encoding.UTF8.GetBytes("I Lost the connection!" + ClientId),
                        QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce
                    }
                };
                if (options.ChannelOptions == null)
                {
                    throw new InvalidOperationException();
                }

                if (!string.IsNullOrEmpty(mqttUser))
                {
                    options.Credentials = new MqttClientCredentials
                    {
                        Username = mqttUser,
                        Password = Encoding.UTF8.GetBytes(mqttPassword)
                    };
                }

                options.CleanSession = true;
                options.KeepAlivePeriod = TimeSpan.FromSeconds(5);

                mqttClient = mqttFactory.CreateMqttClient() as MqttClient;
                mqttClient.ConnectedHandler = new MqttClientConnectedHandlerDelegate(OnMqttClientConnected);
                mqttClient.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(OnMqttClientDisConnected);
                mqttClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(OnSubscriberMessageReceived);
                MqttClientAuthenticateResult result =  await mqttClient.ConnectAsync(options);
                return result.ResultCode;
            }
            catch (Exception ex)
            {
                return MqttClientConnectResultCode.ServerBusy;
                //lbxMonitor.BeginInvoke(_updateMonitorAction,
                //        Logger.TraceLog(Logger.Level.Fatal, $"客户端尝试连接出错.>{ex.Message}"));
            }
        }

        public async Task<int> CLientConnect(string tcpServer, int tcpPort, string mqttUser, string mqttPassword,
            String ClientId = "01001")
        {
            MqttClientConnectResultCode resultCode = await ClientStart(tcpServer, tcpPort, mqttUser, mqttPassword, ClientId);
            if(resultCode == MqttClientConnectResultCode.Success)
            {
                return 0;
            }
            else
            {
                return 1;
            }
        }

        public async Task ClientStop()
        {
            try
            {
                if (mqttClient == null) return;
                await mqttClient.DisconnectAsync();
                mqttClient = null;
            }
            catch (Exception ex)
            {
            }
        }

        public void OnMqttClientConnected(MqttClientConnectedEventArgs e)
        {
            //Console.WriteLine($"客户端[{e.ClientId}]已断开连接!");
            //btnConnect.Text = "Connected"; ;
            //btnConnect.BackColor = Color.LightGreen;
            //btnConnect.Tag = 1;
        }

        public void OnMqttClientDisConnected(MqttClientDisconnectedEventArgs e)
        {
            //Console.WriteLine($"客户端[{e.ClientId}]已断开连接!");
        }

        public async Task ClientPublishMqttTopic(string topic, string payload, int qos = 1, bool retain = false)
        {
            try
            {
                var message = new MqttApplicationMessage()
                {
                    Topic = topic,
                    Payload = Encoding.UTF8.GetBytes(DateTime.Now.ToString("HH:mm:ss:ffff") + ":" + payload),
                    QualityOfServiceLevel = (MqttQualityOfServiceLevel)qos,// cmbQos.SelectedIndex,
                    Retain = retain//bool.Parse(cmbRetain.SelectedItem.ToString())
                };
                await mqttClient.PublishAsync(message);
                //lbxMonitor.BeginInvoke(_updateMonitorAction,
                //        Logger.TraceLog(Logger.Level.Info, string.Format("客户端[{0}]发布主题[{1}]成功!", mqttClient.Options.ClientId, topic)));
            }
            catch (Exception ex)
            {
                //lbxMonitor.BeginInvoke(_updateMonitorAction,
                //            Logger.TraceLog(Logger.Level.Fatal, string.Format("客户端[{0}]发布主题[{1}]异常!>{2}", mqttClient.Options.ClientId, topic, ex.Message)));
            }
        }

        public async Task ClientSubscribeTopic(string topic)
        {
            await mqttClient.SubscribeAsync(topic);
            //lbxMonitor.BeginInvoke(_updateMonitorAction,
            //        Logger.TraceLog(Logger.Level.Info, string.Format("客户端[{0}]订阅主题[{1}]成功!", mqttClient.Options.ClientId, topic)));
        }

        public async Task ClientUnSubscribeTopic(string topic)
        {
            await mqttClient.UnsubscribeAsync(topic);
            //lbxMonitor.BeginInvoke(_updateMonitorAction,
            //        Logger.TraceLog(Logger.Level.Info, string.Format("客户端[{0}]取消主题[{1}]成功!", mqttClient.Options.ClientId, topic)));
        }
        /// <summary>
        /// 当客户端接收到所订阅的主题消息时
        /// </summary>
        /// <param name="e"></param>
        private void OnSubscriberMessageReceived(MqttApplicationMessageReceivedEventArgs e)
        {
            string text = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
            string Topic = e.ApplicationMessage.Topic;
            string QoS = e.ApplicationMessage.QualityOfServiceLevel.ToString();
            string Retained = e.ApplicationMessage.Retain.ToString();
            Console.WriteLine(Topic +" " +text);
            sb.Append(DateTime.Now.ToString("HH:mm:ss:ffff") + ">>" + Topic + " " + text);
            //lbxMonitor.BeginInvoke(_updateMonitorAction,
            //        Logger.TraceLog(Logger.Level.Info, "MessageReceived >>Topic:" + Topic + "; QoS: " + QoS + "; Retained: " + Retained));
            //lbxMonitor.BeginInvoke(_updateMonitorAction,
            //        Logger.TraceLog(Logger.Level.Info, "MessageReceived >>Msg: " + text));
        }


    }

然后在Winform的界面程序中定义使用。文章来源地址https://www.toymoban.com/news/detail-529059.html

到了这里,关于MQTT协议测试工具及核心代码的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 物联网协议之MQTT

    MQTT 可以简单看做一个网络协议,用于 机器对机器 的通信(与客户端到服务器的传输有点区别)。智能传感器、可穿戴设备和其他物联网(IoT)设备通常必须通过 带宽有限 的 资源受限 网络传输和接收数据。这些物联网设备使用 MQTT 进行数据传输,因为它易于实施,并且可

    2024年02月05日
    浏览(55)
  • 【物联网协议】MQTT

    MQTT是一种基于发布/订阅模式的协议,其中客户端(称为发布者)向主题发布消息,并允许其他客户端(称为订阅者)从这些主题订阅消息。MQTT支持多个QoS级别以及最后遗愿消息(LWT)等高级特性。 MQTT客户端通常需要与MQTT代理服务器进行通信以发布或订阅消息。它们可以运

    2024年02月02日
    浏览(52)
  • 基于mqtt协议的物联网项目之微信小程序mqtt连接(三)

    官方连接 https://www.emqx.com/zh/blog/how-to-use-mqtt-in-wechat-miniprogram 所以uniapp使用v4.1.0版本,试过v3.0.0也可以用 默认从阿里云下载的SSL证书为cert.pem和key.key格式,我们要使用转换工具把.key转换成.pem 在线转换地址为:https://www.myssl.cn/tools/merge-pem-cert.html 替换EMQX安装路径下etc/emqx/cer

    2024年02月10日
    浏览(53)
  • 穿越物联网的迷雾:深入理解MQTT协议

    在物联网(IoT)的海洋中,数据像水流一样流动,连接成千上万个设备,使它们能够“交流”并执行任务。要实现这种无缝的交流,需要一个高效、轻量级的通信协议,而MQTT(Message Queuing Telemetry Transport)正是这样一种协议。本篇博客将深入探索MQTT的世界,从它的基础原理到

    2024年04月25日
    浏览(41)
  • Java 物联网协议MQTT和 CoAP

    Survive by day and develop by night. talk for import biz , show your perfect code,full busy,skip hardness,make a better result,wait for change,challenge Survive. happy for hardess to solve denpendies. java IOT 协议是大家非常熟悉的一种的协议。 需求: 1.概述 2. 协议分类 1.概述 物联网核心组成就是物联设备、网关和云端

    2024年01月25日
    浏览(53)
  • 物联网通信协议-MQTT及使用python实现

    简述 MQTT(Message Queuing Telemetry Transport,消息队列遥测传输协议),是一种基于发布/订阅(publish/subscribe)模式的\\\"轻量 级\\\"通讯协议,该协议构建于TCP/IP协议上,由IBM在1999年发布。MQTT最大优点在于,可以以极少的代码和有限的带宽,为连接远程设备提供实时可靠的消息服务。作

    2024年02月10日
    浏览(48)
  • 基于OpenWrt使用MQTT物联网协议详解和应用

    大家好,这篇文章给大家介绍MQTT协议以及如何在OpenWrt系统中使用MQTT客户端和开发,并给出相关实例代码。 MQTT(Message Queuing Telemetry Transport)是一种轻量级的通信协议,设计用于在低带宽和不稳定的网络环境中传输消息。它最初由IBM开发,用于连接远程设备和传感器到网络,

    2024年02月02日
    浏览(60)
  • Java实践-物联网loT入门-MQTT传输协议

    MQTT 是一个极其 轻量级 的 发布/订阅 消息 传输协议,适用于网络带宽较低的场合 . 它 通过 一个 代理服务器 (broker),任何一个 客户端 (client)都可以订阅或者发布某个主题的消息,然后订阅了该主题的客户端则会收到该消息 硬件采集的数据传入EMQX平台(采用MQTT协议),j

    2024年02月09日
    浏览(53)
  • 揭秘MQTT:为何它是物联网的首选协议?

    概览 MQTT 是一种基于 发布/订阅模式 的轻量级消息传输协议,专门针对低带宽和不稳定网络环境的物联网应用而设计,可以用极少的代码为联网设备提供实时可靠的消息服务。MQTT 协议广泛应用于物联网、移动互联网、智能硬件、车联网、智慧城市、远程医疗、电力、石油与

    2024年02月05日
    浏览(57)
  • 使用4G通信模块和MQTT协议,完成物联网设备开发。

    (1)安装并使用4G模块通信模块,建立microPython开发环境; (2)使用提供的Demo开发例程,使用MQTT传输协议连接阿里或腾讯网站,完成物联网设备开发。 (3)将温湿度信息上传到网站; (4)手机APP查看数 这是第一步,在阿里云平台创建产品和设备,用来将实际的设备数据

    2024年02月04日
    浏览(63)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包