.net core 使用MQTTNET搭建MQTT 服务以及客户端例子

80 阅读 0 评论 53 点赞

我是靠谱客的博主老实服饰，最近开发中收集的这篇文章主要介绍.net core 使用MQTTNET搭建MQTT 服务以及客户端例子，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

最近项目可能用到MQTT协议故而稍作研究了一下，MQTT协议，基于TCP封装的发布订阅的消息传递机制，理论详情可查看MQTT--入门_liefyuan的博客-CSDN博客_mqtt 这位老兄的总结，废话不多说，先上效果图。

采用.NET体系用的较多的MQTTNET 3.0版本实现，服务端代码如下

try
{
var options = new MqttServerOptions
{
//连接验证
ConnectionValidator = new MqttServerConnectionValidatorDelegate(p =>
{
if (p.ClientId == "SpecialClient")
{
if (p.Username != "USER" || p.Password != "PASS")
{
p.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
}
}
}),

// Storage = new RetainedMessageHandler(),

//消息拦截发送验证
ApplicationMessageInterceptor = new MqttServerApplicationMessageInterceptorDelegate(context =>
{
if (MqttTopicFilterComparer.IsMatch(context.ApplicationMessage.Topic, "/myTopic/WithTimestamp/#"))
{
// Replace the payload with the timestamp. But also extending a JSON
// based payload with the timestamp is a suitable use case.
context.ApplicationMessage.Payload = Encoding.UTF8.GetBytes(DateTime.Now.ToString("O"));
}

if (context.ApplicationMessage.Topic == "not_allowed_topic")
{
context.AcceptPublish = false;
context.CloseConnection = true;
}
}),
///订阅拦截验证
SubscriptionInterceptor = new MqttServerSubscriptionInterceptorDelegate(context =>
{
if (context.TopicFilter.Topic.StartsWith("admin/foo/bar") && context.ClientId != "theAdmin")
{
context.AcceptSubscription = false;
}

if (context.TopicFilter.Topic.StartsWith("the/secret/stuff") && context.ClientId != "Imperator")
{
context.AcceptSubscription = false;
context.CloseConnection = true;
}
})
};

// Extend the timestamp for all messages from clients.
// Protect several topics from being subscribed from every client.

//var certificate = new X509Certificate(@"C:certstesttest.cer", "");
//options.TlsEndpointOptions.Certificate = certificate.Export(X509ContentType.Cert);
//options.ConnectionBacklog = 5;
//options.DefaultEndpointOptions.IsEnabled = true;
//options.TlsEndpointOptions.IsEnabled = false;

var mqttServer = new MqttFactory().CreateMqttServer();

mqttServer.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(e =>
{
Console.WriteLine(
$"'{e.ClientId}' reported '{e.ApplicationMessage.Topic}' > '{Encoding.UTF8.GetString(e.ApplicationMessage.Payload ?? new byte[0])}'",
ConsoleColor.Magenta);
});

//options.ApplicationMessageInterceptor = c =>
//{
// if (c.ApplicationMessage.Payload == null || c.ApplicationMessage.Payload.Length == 0)
// {
// return;
// }

// try
// {
// var content = JObject.Parse(Encoding.UTF8.GetString(c.ApplicationMessage.Payload));
// var timestampProperty = content.Property("timestamp");
// if (timestampProperty != null && timestampProperty.Value.Type == JTokenType.Null)
// {
// timestampProperty.Value = DateTime.Now.ToString("O");
// c.ApplicationMessage.Payload = Encoding.UTF8.GetBytes(content.ToString());
// }
// }
// catch (Exception)
// {
// }
//};

//开启订阅以及取消订阅
mqttServer.ClientSubscribedTopicHandler = new MqttServerClientSubscribedHandlerDelegate(MqttServer_SubscribedTopic);
mqttServer.ClientUnsubscribedTopicHandler = new MqttServerClientUnsubscribedTopicHandlerDelegate(MqttServer_UnSubscribedTopic);

//客户端连接事件
mqttServer.UseClientConnectedHandler( MqttServer_ClientConnected);

//客户端断开事件
mqttServer.UseClientDisconnectedHandler(MqttServer_ClientDisConnected);
await mqttServer.StartAsync(options);

Console.WriteLine("服务启动成功,输入任意内容并回车停止服务");
Console.ReadLine();

await mqttServer.StopAsync();
}
catch (Exception e)
{
Console.WriteLine(e);
}

Console.ReadLine();

客户端代码如下：

   // Create a new MQTT client.
            var factory = new MqttFactory();
            mqttClient = factory.CreateMqttClient();
            // Create TCP based options using the builder.
            var options = new MqttClientOptionsBuilder()
                .WithClientId(clientId)
                .WithCredentials(user, pwd)//用户名 密码
                .WithCleanSession()
                  .WithTcpServer("127.0.0.1", 1883) // Port is optional TCP 服务
                                                    //      .WithTls(
                                                    //new MqttClientOptionsBuilderTlsParameters
                                                    //{
                                                    //    UseTls = true,
                                                    //    CertificateValidationCallback = (X509Certificate x, X509Chain y, SslPolicyErrors z, IMqttClientOptions o) =>
                                                    //    {
                                                    //        // TODO: Check conditions of certificate by using above parameters.
                                                    //        return true;
                                                    //    },

                //})//类型 TCPS
                .Build();

            //重连机制
            mqttClient.UseDisconnectedHandler(async e =>
                {
                    Console.WriteLine("### DISCONNECTED FROM SERVER ###");
                    await Task.Delay(TimeSpan.FromSeconds(3));

                    try
                    {
                        await mqttClient.ConnectAsync(options, CancellationToken.None); // Since 3.0.5 with CancellationToken
                }
                    catch
                    {
                        Console.WriteLine("### RECONNECTING FAILED ###");
                    }
                });

            //消费消息
            mqttClient.UseApplicationMessageReceivedHandler(e =>
            {
                Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###");
                Console.WriteLine($"+ Topic = {e.ApplicationMessage.Topic}");//主题
                Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");//页面信息
                Console.WriteLine($"+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}");//消息等级
                Console.WriteLine($"+ Retain = {e.ApplicationMessage.Retain}");//是否保留
                Console.WriteLine();

                // Task.Run(() => mqttClient.PublishAsync("hello/world"));
            });



            //连接成功触发订阅主题
            mqttClient.UseConnectedHandler(async e =>
            {
                Console.WriteLine($"### 成功连接 ###");

                // Subscribe to a topic
                //await mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic("my/topic").Build());


                //Console.WriteLine("### SUBSCRIBED ###");
            });

            try
            {

                await mqttClient.ConnectAsync(options, CancellationToken.None);
                bool isExit = false;
                while (!isExit)
                {
                    Console.WriteLine(@"请输入
                    1.订阅主题
                    2.取消订阅
                    3.发送消息
                    4.退出输入exit");
                    var input = Console.ReadLine();
                    switch (input)
                    {
                        case "1":
                            Console.WriteLine(@"请输入主题名称：");
                            var topicName = Console.ReadLine();
                            await SubscribedTopic(topicName);
                            break;
                        case "2":
                            Console.WriteLine(@"请输入需要取消订阅主题名称：");
                            topicName = Console.ReadLine();
                            await UnsubscribedTopic(topicName);
                            break;
                        case "3":
                            Console.WriteLine("请输入需要发送的主题名称");
                            topicName = Console.ReadLine();
                            Console.WriteLine("请输入需要发送的消息");
                            var message = Console.ReadLine();

                            await PublishMessageAsync(new MQTTMessageModel() { Payload = message, Topic = topicName, Retain = true });
                            break;
                        case "exit":
                            isExit = true;
                            break;
                        default:
                            Console.WriteLine("请输入正确指令！");
                            break;
                    }
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.ToString());
                Console.ReadLine();
            }

服务端以及客户端都是core 3.1需要的小伙伴可以下载完整例子参考。