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; }
    }

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);
    }
}

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;
        }
    }
}

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;
        }
    }
}

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
    }
}

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}。");
            }
        }
    }
}

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
    }
}