C# MQTT 断线重连（参考1）

初次接触MQTT

这篇博客借鉴了：https://blog.csdn.net/lxrj2008/article/details/76067242

自己做了一些修改

1.添加全局静态变量 uPLibrary.Networking.M2Mqtt.MQTTConfig.IsSocketRun;

    class MQTTConfig{
     public static bool IsSocketRun = false;
    }

2.修改MqttClient 类 的Connect 方法，在连接成功后把IsSocketRun = true.
MQTTConfig.IsSocketRun = true;

      /// <summary>
            /// Connect to broker
            /// </summary>
            /// <param name="clientId">Client identifier</param>
            /// <param name="username">Username</param>
            /// <param name="password">Password</param>
            /// <param name="willRetain">Will retain flag</param>
            /// <param name="willQosLevel">Will QOS level</param>
            /// <param name="willFlag">Will flag</param>
            /// <param name="willTopic">Will topic</param>
            /// <param name="willMessage">Will message</param>
            /// <param name="cleanSession">Clean sessione flag</param>
            /// <param name="keepAlivePeriod">Keep alive period</param>
            /// <returns>Return code of CONNACK message from broker</returns>
            public byte Connect(string clientId,
                string username,
                string password,
                bool willRetain,
                byte willQosLevel,
                bool willFlag,
                string willTopic,
                string willMessage,
                bool cleanSession,
                ushort keepAlivePeriod)
            {
                // create CONNECT message
                MqttMsgConnect connect = new MqttMsgConnect(clientId,
                    username,
                    password,
                    willRetain,
                    willQosLevel,
                    willFlag,
                    willTopic,
                    willMessage,
                    cleanSession,
                    keepAlivePeriod,
                    (byte)this.ProtocolVersion);
     
                try
                {
                    // connect to the broker
                    this.channel.Connect();
                }
                catch (Exception ex)
                {
                    throw new MqttConnectionException("Exception connecting to the broker", ex);
                }
     
                this.lastCommTime = 0;
                this.isRunning = true;
                MQTTConfig.IsSocketRun = true;
                this.isConnectionClosing = false;
                // start thread for receiving messages from broker
                Fx.StartThread(this.ReceiveThread);
     
    ....

3.继续修改 MqttClient .cs类中的Ping() 方法，MQTTConfig.IsSocketRun = false;

    /// <summary>
    /// Execute ping to broker for keep alive
    /// </summary>
    /// <returns>PINGRESP message from broker</returns>
    private MqttMsgPingResp Ping()
    {
        MqttMsgPingReq pingreq = new MqttMsgPingReq();
        try
        {
            // broker must send PINGRESP within timeout equal to keep alive period
            return (MqttMsgPingResp)this.SendReceive(pingreq, this.keepAlivePeriod);
        }
        catch (Exception e)
        {
    #if TRACE
            MqttUtility.Trace.WriteLine(TraceLevel.Error, "Exception occurred: {0}", e.ToString());
    #endif
            MQTTConfig.IsSocketRun = false;
            // client must close connection
            this.OnConnectionClosing();
            return null;
        }
    }

4.继续修改 MqttClient .cs类中的Close() 方法，关闭后设置值为MQTTConfig.IsSocketRun = false;

     private void Close()
    #endif
            {
                // stop receiving thread
                 this.isRunning = false;
                MQTTConfig.IsSocketRun = false;
                // wait end receive event thread
                if (this.receiveEventWaitHandle != null)
                    this.receiveEventWaitHandle.Set();
     
     
                // wait end process inflight thread
                if (this.inflightWaitHandle != null)
                    this.inflightWaitHandle.Set();
     
     
    #if BROKER
                // unlock keep alive thread
                this.keepAliveEvent.Set();
    #else
                // unlock keep alive thread and wait
                this.keepAliveEvent.Set();
     
     
                if (this.keepAliveEventEnd != null)
                    this.keepAliveEventEnd.WaitOne();
    #endif
     
     
                // clear all queues
                this.inflightQueue.Clear();
                this.internalQueue.Clear();
                this.eventQueue.Clear();
     
     
                // close network channel
                this.channel.Close();
     
     
                this.IsConnected = false;
            }

——————————————————————————————————————————————

上述修改增加了一个判断是否连接的属性，如果不想增加字段MQTTConfig.IsSocketRun，可以用底层自带的isRunning属性，

private bool isRunning;是私有属性，需要改成public才可以在外部调用

_________________________________________________________________________________________________________


5.继续修改 MqttClient .cs类中的Close() 方法，关闭后设置值为MQTTConfig.IsSocketRun = false;

     //加入一个线程死循环
            private void ThreadSendKey()
            {
                while (true)
                {
                    //在M2Mqtt.Net.dll中加入了IsSocketRun属性，判断连接是否断开
                    if (!uPLibrary.Networking.M2Mqtt.MQTTConfig.IsSocketRun)
                    {
                        //连接断开则重新连接
                        connectServer();
                        if (uPLibrary.Networking.M2Mqtt.MQTTConfig.IsSocketRun) {
                            //连接成功重新订阅
                            subMessages();
                            //关闭当前线程
                            System.Threading.Thread.CurrentThread.Abort();
                        }
                    }
                    Thread.Sleep(2000);
                }
            }
     //在connectServer方法成功后启动线程                 
                    Thread th = new Thread(ThreadSendKey);
                    th.Start(); //启动线程
————————————————
版权声明：本文为CSDN博主「lmy_loveF」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/lmy_loveF/article/details/80002530