使用Surging Mqtt 開發基於WS的MqttClient客戶端

時間 2019-11-18

標籤使用 surging mqtt 開發基於 mqttclient 客戶端简体版

原文原文鏈接

最近一段時間因爲要作一套智能設備系統，而有幸瞭解到Surging中的Mqtt broker，學習了不少東西本篇文章基於Surging開源的.netcore項目有興趣的朋友可點擊此處進行了解。話很少說咱們來基於Surging 中的WS與MqttClient結合來開發服務端MqttClient的使用。html

準備工做

開發環境： Visual Studio 2017 15.9.5git

.netCore版本：2.2.102（目前Surging已經升級至netcore 2.2版本）github

surging項目下載地址 https://github.com/dotnetcore/surging json

開始工做

接口部分

新建類庫Surging.IModuleServices.MqttWithWSapi

添加引用Surging.Core.Protocol.WS緩存

新建文件夾Model 並建立類MqttClientOption.cs 此類爲讀取surgingSettings.json中配置的MqttClient的相關參數MqttClientOption的代碼以下app

public class MqttClientOption
    {
        public string ClientID { get; set; }
        public string MqttClientConnection { get; set; } = "";
        public string MqttClientUserName { get; set; }
        public string MqttClientPassword { get; set; }
        public int Port { get; set; }
        public int KeepAlivedTime { get; set; }
        public List<string> Topics { get; set; } = new List<string>();
        public bool CleanSession { get; set; }
    }

surgingSettings.json的MqttClient配置代碼參見底部Surging.MqttClientWithWsServices.Server中的配置async

根據Surging做者建立的例子來建立一個接口IChatService.cside

using Surging.Core.CPlatform.Ioc;
using Surging.Core.CPlatform.Runtime.Client.Address.Resolvers.Implementation.Selectors.Implementation;
using Surging.Core.CPlatform.Runtime.Server.Implementation.ServiceDiscovery.Attributes;
using Surging.Core.CPlatform.Support.Attributes;
using Surging.Core.Protocol.WS.Attributes;
using Surging.IModuleServices.MqttWithWS.Models;
using System;
using System.Threading.Tasks;

namespace Surging.IModuleServices.MqttWithWS
{
    [ServiceBundle("Api/{Service}")]
    [BehaviorContract(IgnoreExtensions = true)]
    public interface IChatService: IServiceKey
    {
        [Command(ShuntStrategy = AddressSelectorMode.HashAlgorithm)]
        Task RunMqttClient(MqttClientOption mqttClientOption);
    }
}

結構如圖所示學習

接口實現部分

新建類庫Surging.Modules.WSWithMqtt

添加引用Surging.IModuleServices.MqttWithWS

添加依賴項：MQTTnet

因爲要獲取surgingSettings.json中的配置項，所以須要建立一個文件夾Configurations並在文件夾下建立SurgingMqttBuilderExtendsions.cs類代碼以下：

using Autofac;
using Microsoft.Extensions.Configuration;
using Surging.Core.CPlatform;
using Surging.Core.ServiceHosting.Internal;
using Surging.IModuleServices.MqttWithWS;
using Surging.IModuleServices.MqttWithWS.Models;
using System;
using System.Collections.Generic;
using System.Text;

namespace Surging.Modules.WSWithMqtt.Configurations
{
  public static class SurgingMqttBuilderExtendsions
    {
        public static IServiceHostBuilder UseMqttClient(this IServiceHostBuilder builder)
        {
            builder.MapServices(collection =>
            {
                MqttClientOption mqttClientOption = new MqttClientOption();
                var section = AppConfig.GetSection("MqttClient");
                if (section.Exists())
                    mqttClientOption = section.Get<MqttClientOption>();
                collection.Resolve<IChatService>().RunMqttClient(mqttClientOption);
            });
            return builder;
        }
    }
}

因爲此服務基於WSServiceBase 所以我建立了MqttWSServieBase繼承自WSServiceBase用於建立MqttClient客戶端，代碼以下：

using MQTTnet;
using MQTTnet.Client;
using Surging.Core.Protocol.WS;
using Surging.IModuleServices.MqttWithWS.Models;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;

namespace Surging.Modules.WSWithMqtt
{
    public abstract class MqttWSServieBase : WSServiceBase
    {
        protected IMqttClientOptions _mqttClientOptions;
        protected IMqttClient _mqttClient;
        protected IEnumerable<TopicFilter> _topicFilters;
        public Task RunMqttClientBase(MqttClientOption mqttClientOption)
        {
            var clientOptions = new MqttClientOptionsBuilder();
            if (mqttClientOption != null)
            {
                clientOptions.WithClientId(mqttClientOption.ClientID + Guid.NewGuid().ToString("N"))
                .WithCleanSession(mqttClientOption.CleanSession);
                clientOptions.WithTcpServer(mqttClientOption.MqttClientConnection, mqttClientOption.Port);
                if (!string.IsNullOrWhiteSpace(mqttClientOption.MqttClientUserName))
                    clientOptions.WithCredentials(mqttClientOption.MqttClientUserName, mqttClientOption.MqttClientPassword);
                clientOptions.WithKeepAlivePeriod(TimeSpan.FromSeconds(mqttClientOption.KeepAlivedTime));
            }
            _mqttClientOptions = clientOptions.Build();
            IList<TopicFilter> filters = new List<TopicFilter>();
            if (mqttClientOption != null)
            {
                foreach (var item in mqttClientOption.Topics)
                {
                    var topicFilerbuilder = new TopicFilterBuilder();
                    topicFilerbuilder.WithTopic(item);
                    filters.Add(topicFilerbuilder.Build());
                }
            }
            _topicFilters = filters;
            _mqttClient = new MqttFactory().CreateMqttClient();
            return Task.CompletedTask;
        }
    }
}

接口繼承類ChatService，此類用於鏈接WS，並經過此鏈接對Surging的 Mqtt Broker進行發佈訂閱。代碼以下：

using MQTTnet;
using MQTTnet.Client;
using Surging.IModuleServices.MqttWithWS;
using Surging.IModuleServices.MqttWithWS.Models;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading.Tasks;
using WebSocketCore;

namespace Surging.Modules.WSWithMqtt
{
    public class ChatService : MqttWSServieBase, IChatService
    {
        private static readonly ConcurrentDictionary<string, string> _users = new ConcurrentDictionary<string, string>();
        private static readonly ConcurrentDictionary<string, string> _clients = new ConcurrentDictionary<string, string>();
        private string _name;
        private string _to;


        protected override void OnMessage(MessageEventArgs e)
        {
            if (_clients.ContainsKey(ID))
            {
                Dictionary<string, object> model = new Dictionary<string, object>();
                model.Add("name", _to);
                model.Add("data", e.Data);
                //var result = ServiceLocator.GetService<IServiceProxyProvider>()
                //     .Invoke<object>(model, "api/chat/SendMessage").Result;

            }
        }

        protected override void OnOpen()
        {
            _name = Context.QueryString["name"];
            _to = Context.QueryString["to"];
            if (!string.IsNullOrEmpty(_name))
            {
                _clients[ID] = _name;
                _users[_name] = ID;
            }
        }
        public Task SendMessage(string name, string data)
        {
            if (_users.ContainsKey(name))
            {
                this.GetClient().SendTo($"hello,{name},{data}", _users[name]);
            }
            return Task.CompletedTask;
        }
        public async Task RunMqttClient(MqttClientOption mqttClientOption)
        {
           await base.RunMqttClientBase(mqttClientOption);
            _mqttClient.ApplicationMessageReceived += _mqttClient_ApplicationMessageReceived;
            _mqttClient.Connected += _mqttClient_Connected;
            _mqttClient.Disconnected += _mqttClient_Disconnected;
            await _mqttClient.ConnectAsync(_mqttClientOptions);
        }
        #region mqttClient
        /// <summary>
        /// 接收消息
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void _mqttClient_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
        {
            if (e.ApplicationMessage != null)
            {

            }
        }
        /// <summary>
        /// 斷開鏈接
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private async void _mqttClient_Disconnected(object sender, MqttClientDisconnectedEventArgs e)
        {
            Console.WriteLine("mqtt客戶端與服務端斷開鏈接!");
            await Task.Delay(TimeSpan.FromSeconds(5));
            try
            {
                await _mqttClient.ConnectAsync(_mqttClientOptions);
            }
            catch
            {
                Console.WriteLine("mqtt客戶端與服務端嘗試鏈接失敗!");
            }
        }
        /// <summary>
        /// 鏈接成功
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void _mqttClient_Connected(object sender, MqttClientConnectedEventArgs e)
        {
            _mqttClient.SubscribeAsync(_topicFilters);
        }
        #endregion
    }
}

結構如圖

建立控制檯應用程序，用於項目啓動

新建控制檯程序Surging.MqttClientWithWsServices.Server將Surging.Services.Server中的配置文件cacheSettings.json、eventBusSettings.json、log4net.config、NLog.config、以及surgingSettings.json文件都拷貝到新建的這個控制檯程序中。

添加依賴項：Autofac Autofac.Extensions.DependencyInjection System.Text.Encoding.CodePages Microsoft.Extensions.Logging Microsoft.Extensions.Logging.Console

添加引用

修改Program.cs

using Autofac;
using Microsoft.Extensions.Logging;
using Surging.Core.Caching.Configurations;
using Surging.Core.CPlatform;
using Surging.Core.CPlatform.Configurations;
using Surging.Core.CPlatform.Utilities;
using Surging.Core.Nlog;
using Surging.Core.ProxyGenerator;
using Surging.Core.ServiceHosting;
using Surging.Core.ServiceHosting.Internal.Implementation;
using Surging.Modules.WSWithMqtt.Configurations;
using System;
using System.Text;

namespace Surging.MqttClientWithWsServices.Server
{
    class Program
    {
        static void Main(string[] args)
        {
            Encoding.RegisterProvider(CodePagesEncodingProvider.Instance);
            var host = new ServiceHostBuilder()
                .RegisterServices(builder =>
                {
                    builder.AddMicroService(option =>
                    {
                        option.AddServiceRuntime()
                        .AddClientProxy()
                            .AddRelateService()
                            .AddConfigurationWatch();
                        builder.Register(p => new CPlatformContainer(ServiceLocator.Current));
                    });
                })
                .ConfigureLogging(logger =>
                {
                    logger.AddConfiguration(
                        Surging.Core.CPlatform.AppConfig.GetSection("Logging"));
                })
                .UseNLog(LogLevel.Debug, "NLog.config")
                .UseServer(options => { })
                .UseConsoleLifetime()
                .UseMqttClient()
                .UseProxy()
                .Configure(build =>
                    build.AddCacheFile("${cachepath}|cacheSettings.json", optional: false, reloadOnChange: true))
                .Configure(build =>
                    build.AddCPlatformFile("${surgingpath}|surgingSettings.json", optional: false, reloadOnChange: true))
                .UseStartup<Startup>()
                .Build();

            using (host.Run())
            {
                Console.WriteLine($"服務端啓動成功，{DateTime.Now}。");
            }
        }
    }
}

注意添加是須要加一段代碼.UseMqttClient()用於啓動MqttClient。

添加Startup.cs 此類與Surging.Services.Server中的Startup.cs一致。

using Autofac;
using Autofac.Extensions.DependencyInjection;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Surging.Core.Caching.Configurations;
using Surging.Core.CPlatform.Utilities;
using System.IO;
namespace Surging.MqttClientWithWsServices.Server
{
    public class Startup
    {

        public Startup(IConfigurationBuilder config)
        {
            config.SetBasePath(Directory.GetCurrentDirectory());
            ConfigureCache(config);
        }

        public IContainer ConfigureServices(ContainerBuilder builder)
        {
            var services = new ServiceCollection();
            ConfigureLogging(services);
            builder.Populate(services);
            ServiceLocator.Current = builder.Build();
            return ServiceLocator.Current;
        }

        public void Configure(IContainer app)
        {

        }

        #region 私有方法
        /// <summary>
        /// 配置日誌服務
        /// </summary>
        /// <param name="services"></param>
        private void ConfigureLogging(IServiceCollection services)
        {
            services.AddLogging();
        }


        /// <summary>
        /// 配置緩存服務
        /// </summary>
        private void ConfigureCache(IConfigurationBuilder build)
        {
            build
                .AddCacheFile("cacheSettings.json", optional: false);
        }
        #endregion
    }
}

如今基本上已經完事了，下面再說下surgingSettings.json配置信息爲了不與Surging.Services.Server的端口重複所以配置以下：

{
  "Surging": {
    "Ip": "${Surging_Server_IP}",
    "WatchInterval": 30,
    "Port": "${Surging_Server_Port}|100",
    "MappingIp": "${Mapping_ip}",
    "MappingPort": "${Mapping_Port}",
    "Token": "true",
    "WanIp": "${Mapping_WanIp}",
    "Libuv": true,
    "MaxConcurrentRequests": 20,
    "ExecutionTimeoutInMilliseconds": 30000,
    "Protocol": "${Protocol}|None", //Http、Tcp、None
    "RootPath": "${RootPath}|",
    "Ports": {
      "HttpPort": "${HttpPort}|2801",
      "MQTTPort": "${MQTTPort}|971",
      "WSPort": "${WSPort}|961"
    },
    "RequestCacheEnabled": false,
    "Packages": [
      {
        "TypeName": "EnginePartModule",
        "Using": "${UseEngineParts}|DotNettyModule;NLogModule;MessagePackModule;ServiceProxyModule;ConsulModule;EventBusRabbitMQModule;CachingModule;"
      }
    ]
  }, //若是引用多個同類型的組件，須要配置Packages，若是是自定義按需引用，無需配置Packages
  "Consul": {
    "ConnectionString": "${Register_Conn}|127.0.0.1:8500", // "127.0.0.1:8500",
    "SessionTimeout": "${Register_SessionTimeout}|50",
    "RoutePath": "${Register_RoutePath}",
    "ReloadOnChange": true,
    "EnableChildrenMonitor": false
  },
  "Swagger": {
    "Version": "${SwaggerVersion}|V1", // "127.0.0.1:8500",
    "Title": "${SwaggerTitle}|Surging Demo",
    "Description": "${SwaggerDes}|surging demo",
    "Contact": {
      "Name": "API Support",
      "Url": "https://github.com/dotnetcore/surging",
      "Email": "fanliang1@hotmail.com"
    },
    "License": {
      "Name": "MIT",
      "Url": "https://github.com/dotnetcore/surging/blob/master/LICENSE"
    }
  },
  "EventBus_Kafka": {
    "Servers": "${EventBusConnection}|localhost:9092",
    "MaxQueueBuffering": "${MaxQueueBuffering}|10",
    "MaxSocketBlocking": "${MaxSocketBlocking}|10",
    "EnableAutoCommit": "${EnableAutoCommit}|false",
    "LogConnectionClose": "${LogConnectionClose}|false",
    "OffsetReset": "${OffsetReset}|earliest",
    "GroupID": "${EventBusGroupID}|surgingdemo"
  },
  "WebSocket": {
    "WaitTime": 2,
    "KeepClean": false,
    "Behavior": {
      "IgnoreExtensions": true,
      "EmitOnPing": false
    }
  },
  "EventBus": {
    "EventBusConnection": "${EventBusConnection}|192.168.1.127",
    "EventBusUserName": "${EventBusUserName}|guest",
    "EventBusPassword": "${EventBusPassword}|guest",
    "VirtualHost": "${VirtualHost}|/",
    "MessageTTL": "${MessageTTL}|30000",
    "RetryCount": "${RetryCount}|1",
    "FailCount": "${FailCount}|3",
    "prefetchCount": "${PrefetchCount}|0",
    "BrokerName": "${BrokerName}|surging_demo",
    "Port": "${EventBusPort}|32671"
  },
  "Zookeeper": {
    "ConnectionString": "${Zookeeper_ConnectionString}|127.0.0.1:2181",
    "SessionTimeout": 50,
    "ReloadOnChange": true
  },
  "Logging": {
    "Debug": {
      "LogLevel": {
        "Default": "Information"
      }
    },
    "Console": {
      "IncludeScopes": true,
      "LogLevel": {
        "Default": "${LogLevel}|Debug"
      }
    },
    "LogLevel": {
      "Default": "${LogLevel}|Debug"
    }
  },
  "MqttClient": {
    "ClientID": "${MqttClientID}|serverclientid",
    "MqttClientConnection": "${MqttClientConnection}|127.0.0.1",
    "MqttClientUserName": "admin",
    "MqttClientPassword": "123456",
    "Port": 97,
    "KeepAlivedTime": 60,
    "CleanSession": true,
    "Topics": [ "test1","test2" ]
  }
}

　總的Server代碼如圖所示：

此時代碼開發階段結束。咱們能夠設置多項目啓動，啓動項目前肯定你的 Consul可以正常使用。本地啓動Consul 能夠經過控制檯來作測試就OK

一、Surging.Services.Server 二、Surging.MqttClientWithWsServices.Server即（紅框標註內容）：

說到此處，想必你們都知道怎麼使用SurgingWS、MqttClient 與MqttBroker進行鏈接了。寫出這段代碼主要是針對於Surging不瞭解，或者攝入不深的人，能直接快速的使用本代碼讓用戶與設備間能夠正常通信。

注意若是使用Docker編排，或者Rancher編排Surging Broker 或者 WSMqttClient 時若是涉及到多個編排咱們須要進行相應的邏輯判斷。

近期因爲我的需求，須要把設備在線狀態通知給各個DBMqttClient端，用於保存現有設備狀態。在不處理設備鏈接時，咱們就能夠知道設備是否在線，是否有異常。異常時長等。若有此需求可在下方留言。此代碼近期可能會貢獻給Surging，讓Surging更增強大。寫這篇文章呢，主要目的是再沒有看懂做者代碼的狀況下，能夠盡情使用MqttBroker的功能。次處只是引用了WS與MqttClient，其實能夠以此爲參考部署更多的MqttClient，好比數據保存與服務通信分開等。在此感謝Surging做者在業餘時間爲咱們作了這麼好的開源項目，願Surging愈來愈好。

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。