使用MQTTnet搭建Mqtt服务器

全栈程序员站长

发布于 2022-09-12 20:29:32

2.9K0

发布于 2022-09-12 20:29:32

文章被收录于专栏：全栈程序员必看全栈程序员必看

大家好，又见面了，我是你们的朋友全栈君。

官方介绍：

MQTTnet

MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). The implementation is based on the documentation from http://mqtt.org/.

Features

General

Async support
TLS 1.2 support for client and server (but not UWP servers)
Extensible communication channels (i.e. In-Memory, TCP, TCP+TLS, WS)
Lightweight (only the low level implementation of MQTT, no overhead)
Performance optimized (processing ~70.000 messages / second)*
Interfaces included for mocking and testing
Access to internal trace messages
Unit tested (~90 tests)

* Tested on local machine (Intel i7 8700K) with MQTTnet client and server running in the same process using the TCP channel. The app for verification is part of this repository and stored in /Tests/MQTTnet.TestApp.NetCore.

Client

Communication via TCP (+TLS) or WS (WebSocket) supported
Included core MqttClient with low level functionality
Also included ManagedMqttClient which maintains the connection and subscriptions automatically. Also application messages are queued and re-scheduled for higher QoS levels automatically.
Rx support (via another project)
Compatible with Microsoft Azure IoT Hub

Server (broker)

List of connected clients available
Supports connected clients with different protocol versions at the same time
Able to publish its own messages (no loopback client required)
Able to receive every message (no loopback client required)
Extensible client credential validation
Retained messages are supported including persisting via interface methods (own implementation required)
WebSockets supported (via ASP.NET Core 2.0, separate nuget)
A custom message interceptor can be added which allows transforming or extending every received application message
Validate subscriptions and deny subscribing of certain topics depending on requesting clients

Supported frameworks

.NET Standard 1.3+
.NET Core 1.1+
.NET Core App 1.1+
.NET Framework 4.5.2+ (x86, x64, AnyCPU)
Mono 5.2+
Universal Windows Platform (UWP) 10.0.10240+ (x86, x64, ARM, AnyCPU, Windows 10 IoT Core)
Xamarin.Android 7.5+
Xamarin.iOS 10.14+

Supported MQTT versions

5.0.0 (planned)
3.1.1
3.1.0

Nuget

This library is available as a nuget package: https://www.nuget.org/packages/MQTTnet/

创建项目

使用vs创建mqtt项目，选择winform项目，方便创建界面，查看相关数据信息。项目包括两个，server和client。

服务器端界面结构如下：

Server在程序中添加本机IP：

var ips = Dns.GetHostAddressesAsync(Dns.GetHostName());

foreach (var ip in ips.Result)
{
    switch (ip.AddressFamily)
    {
        case AddressFamily.InterNetwork:
           TxbServer.Text = ip.ToString();
           break;
        case AddressFamily.InterNetworkV6:
           break;
     }
}

添加一个Action ，用来想listbox中添加数据：

private Action<string> _updateListBoxAction;


//在load方法中定义
//超过1000条数据时，自动删除第一个
//出现滚动条时，滚动条自动向下移动
_updateListBoxAction = new Action<string>((s) =>
{
    listBox1.Items.Add(s);
    if (listBox1.Items.Count > 1000)
    {
        listBox1.Items.RemoveAt(0);
    }
    var visibleItems = listBox1.ClientRectangle.Height/listBox1.ItemHeight;

    listBox1.TopIndex = listBox1.Items.Count - visibleItems + 1;
});


//添加按键事件，按c时清空listbox
listBox1.KeyPress += (o, args) =>
{
    if (args.KeyChar == 'c' || args.KeyChar=='C')
    {
         listBox1.Items.Clear();
    }
};

mqttserver

        private async void MqttServer()
        {
            if (null != _mqttServer)
            {
                return;
            }

            var optionBuilder =
                new MqttServerOptionsBuilder().WithConnectionBacklog(1000).WithDefaultEndpointPort(Convert.ToInt32(TxbPort.Text));

            if (!TxbServer.Text.IsNullOrEmpty())
            {
                optionBuilder.WithDefaultEndpointBoundIPAddress(IPAddress.Parse(TxbServer.Text));
            }
            
            var options = optionBuilder.Build();
            
            
            (options as MqttServerOptions).ConnectionValidator += context =>
            {
                if (context.ClientId.Length < 10)
                {
                    context.ReturnCode = MqttConnectReturnCode.ConnectionRefusedIdentifierRejected;
                    return;
                }
                if (!context.Username.Equals("admin"))
                {
                    context.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
                    return;
                }
                if (!context.Password.Equals("public"))
                {
                    context.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
                    return;
                }
                context.ReturnCode = MqttConnectReturnCode.ConnectionAccepted;
                
            };
            

            _mqttServer = new MqttFactory().CreateMqttServer();
            _mqttServer.ClientConnected += (sender, args) =>
            {
                listBox1.BeginInvoke(_updateListBoxAction, $">Client Connected:ClientId:{args.ClientId},ProtocalVersion:");

                var s = _mqttServer.GetClientSessionsStatusAsync();
                label3.BeginInvoke(new Action(() => { label3.Text = $"连接总数：{s.Result.Count}"; }));
            };

            _mqttServer.ClientDisconnected += (sender, args) =>
            {
                listBox1.BeginInvoke(_updateListBoxAction, $"<Client DisConnected:ClientId:{args.ClientId}");
                var s = _mqttServer.GetClientSessionsStatusAsync();
                label3.BeginInvoke(new Action(() => { label3.Text = $"连接总数：{s.Result.Count}"; }));
            };

            _mqttServer.ApplicationMessageReceived += (sender, args) =>
            {
                listBox1.BeginInvoke(_updateListBoxAction,
                    $"ClientId:{args.ClientId} Topic:{args.ApplicationMessage.Topic} Payload:{Encoding.UTF8.GetString(args.ApplicationMessage.Payload)} QualityOfServiceLevel:{args.ApplicationMessage.QualityOfServiceLevel}");

            };

            _mqttServer.ClientSubscribedTopic += (sender, args) =>
            {
                listBox1.BeginInvoke(_updateListBoxAction, $"@ClientSubscribedTopic ClientId:{args.ClientId} Topic:{args.TopicFilter.Topic} QualityOfServiceLevel:{args.TopicFilter.QualityOfServiceLevel}");
            };
            _mqttServer.ClientUnsubscribedTopic += (sender, args) =>
            {
                listBox1.BeginInvoke(_updateListBoxAction, $"%ClientUnsubscribedTopic ClientId:{args.ClientId} Topic:{args.TopicFilter.Length}");
            };

            _mqttServer.Started += (sender, args) =>
            {
                listBox1.BeginInvoke(_updateListBoxAction, "Mqtt Server Start...");
            };

            _mqttServer.Stopped += (sender, args) =>
            {
                listBox1.BeginInvoke(_updateListBoxAction, "Mqtt Server Stop...");
                
            };

            await _mqttServer.StartAsync(options);
        }

把mqttserver中定义的事件都进行了绑定

(options as MqttServerOptions).ConnectionValidator 为进行mqttclient连接时进行的验证工作

想要查看完整代码，请移步到gitee

https://gitee.com/sesametech-group/MqttNetSln

你感觉文章还可以，移步到gitee上给个star

发布者：全栈程序员栈长，转载请注明出处：https://javaforall.cn/152877.html原文链接：https://javaforall.cn

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

如有侵权请联系 cloudcommunity@tencent.com 删除

mqtt