Winform MQTT客户端连接方式

news2025/4/8 0:45:42

项目中使用到Winform的数据转发服务，所以记录下使用到的方法。

一.创建单例模板

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace ConsoleApp.Scripts
{
    public class SingleTon<T>where T:SingleTon<T>,new()
    {
        private static T m_instance;
        private static readonly object m_lockObject = new object();

        public static T GetInstance 
        {

            get 
            {
                if (m_instance==null)
                {
                   
                    lock (m_lockObject)
                    {
                        if (m_instance == null)
                        {
                            m_instance = new T();
                        }
                    }
                }
                return m_instance;
            }
        }
    }
}

二. 创建客户端连接

using MQTTnet; 
using MQTTnet.Client; 
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Protocol;
using MQTTnet.Server;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace ConsoleApp.Scripts
{
  public class MQTTClientConnent:SingleTon<MQTTClientConnent>
    {
        private IMqttClient m_mqttClient = null;

        private IManagedMqttClient m_managedmqttClient;
        private MqttFactory Factory = new MqttFactory();
        /// <summary>
        /// 操作参数
        /// </summary>
        private MqttClientOptions Options;

        /// <summary>
        /// 消息参数
        /// </summary>
        private ManagedMqttClientOptions ManagedOptions;

        private string m_UDPTitleS;
        private string MQTTUserName;
        private string MQTTPWD;
        /// <summary>
        /// 客户端初始化
        /// </summary>
        public void Client_Init()
        {
            m_UDPTitleS = GetConfigurationSetting.GetInstance.GetKeyValue("UDPTitle");
            MQTTUserName = GetConfigurationSetting.GetInstance.GetKeyValue("MQTTUserName");
            MQTTPWD = GetConfigurationSetting.GetInstance.GetKeyValue("MQTTPWD");

            //配置初始化
            //配置连接参数
            Options = new MqttClientOptionsBuilder()
                  .WithTcpServer("127.0.0.1", 1883)
                .WithClientId("XWJiJuID")
                .WithCredentials(MQTTUserName, MQTTPWD)
                .WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce)
               // .WithCleanSession()
                .WithKeepAlivePeriod(TimeSpan.FromSeconds(5))
                .Build();

            //配置重连机制
            ManagedOptions = new ManagedMqttClientOptionsBuilder()
                .WithAutoReconnectDelay(TimeSpan.FromSeconds(5))
                .WithClientOptions(Options)
                .Build();

            //创建客户端
            m_managedmqttClient = Factory.CreateManagedMqttClient();
            //配置数据接收事件
          //  m_managedmqttClient.ApplicationMessageReceivedAsync += Receive_Event;
            m_managedmqttClient.ConnectedAsync += Connected_Event;
            m_managedmqttClient.DisconnectedAsync += Disconnected_Event;

             

        }

        public async Task ConnectMqttServerAsync()
        {

            Client_Init();

            try
            {
                //订阅主题
                string[] topicArray = new string[] { };
                if (topicArray.Length > 0)
                {
                    await m_managedmqttClient.SubscribeAsync(
                        topicArray.Select(t => new MqttTopicFilterBuilder().WithTopic(t).WithAtMostOnceQoS().Build()).ToArray()
                        );
                }
                //连接服务器
                await m_managedmqttClient.StartAsync(ManagedOptions);

            }
            catch (Exception e)
            {
                Console.WriteLine(e);
            }

            #region m_mqttClient

            //配置连接参数
            //var options = new MqttClientOptionsBuilder()
            //    .WithTcpServer("127.0.0.1", 1883)
            //    .WithClientId(Guid.NewGuid().ToString())
            //     .WithCleanSession()
            //     .WithKeepAlivePeriod(TimeSpan.FromSeconds(5))
            //    .Build();
            //if (m_mqttClient == null)
            //{
            //    m_mqttClient = new MqttFactory().CreateMqttClient();


            //    //  MqttServerOptions mqttServerOptions = new MqttServerOptions();
            //    //   mqttServer = new MqttFactory().CreateMqttServer(mqttServerOptions);
            //}

            // await m_mqttClient.ConnectAsync(options);
            // await m_mqttClient.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("test/topic").Build());


            // 发布一条消息
            //var applicationMessage = new MqttApplicationMessageBuilder()
            //    .WithTopic("4001")
            //    .WithPayload("Hello, MQTT!")
            //    .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce)
            //    .Build();

            // await m_mqttClient.PublishAsync(applicationMessage);

            #endregion
            Console.WriteLine("MQTT start-----=====");
        }


        private async Task Disconnected_Event(MqttClientDisconnectedEventArgs arg)
        {
            await Task.Delay(100);
            Console.WriteLine("MQTT断开链接=====" + arg.ToString());
        }



        private async Task Receive_Event(MqttApplicationMessageReceivedEventArgs arg)
        {
          await Task.Delay(100);
        }

        private async Task Connected_Event(MqttClientConnectedEventArgs arg)
        {
            await Connected_Func(arg);
        }

        /// <summary>
        /// 连接服务器处理方法
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        public virtual async Task Connected_Func(MqttClientConnectedEventArgs arg)
        {
            await Task.Delay(100);
            Console.WriteLine("MQTT连接成功====="+ arg.ToString());
        }

        //  MqttServer mqttServer;
     

        /// <summary>
        /// 客户端断开
        /// </summary>
        public void Disconnected()
        {
            m_managedmqttClient.StopAsync();
           // m_managedmqttClient.ApplicationMessageReceivedAsync -= Receive_Event;
            m_managedmqttClient.ConnectedAsync -= Connected_Event;
            m_managedmqttClient.DisconnectedAsync -= Disconnected_Event;

        }

        /// <summary>
        /// 发送消息
        /// </summary>
        /// <param name="topic">主题</param>
        /// <param name="content">内容</param>
        public virtual void SendMessage(string topic, string content)
        {
          
            if (m_managedmqttClient != null && m_managedmqttClient.IsConnected)
                m_managedmqttClient.EnqueueAsync(m_UDPTitleS + topic, content);

         //   Console.WriteLine("MQTTSend=====" + m_UDPTitleS + topic + "__" + content);
        }

    }
}

三.配置文件

<?xml version="1.0" encoding="utf-8" ?>
<configuration>
    <startup> 
        <supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.7.2" />
    </startup>
	<appSettings> 
		<add key="UDPPort" value="4001|4002|4003"/>
		<add key="UDPTitle" value="DW"/>
		<add key="DataStorageIP" value="192.168.1.110"/>
		<add key="DataStoragePort" value="6666"/>
		
		
		<add key="MQTTUserName" value="DW001"/>
		<add key="MQTTPWD" value="123456"/>

	</appSettings>
</configuration>

四.主程序调用

namespace ConsoleApp
{
    internal class Program
    {
        static void Main(string[] args)
        {
            AppDomain.CurrentDomain.ProcessExit += CurrentDomain_ProcessExit;
            Task task = MQTTClientConnent.GetInstance.ConnectMqttServerAsync();
            Console.ReadKey();
        }

        private static void CurrentDomain_ProcessExit(object sender, EventArgs e)
        {
            Console.WriteLine("应用程序正在退出！");
            MQTTClientConnent.GetInstance.Disconnected();
        }
    }
}