C# Mqtt 断线重连

  • 2019 年 10 月 3 日
  • 筆記

在通过 MqttClient 客户端连接之后,在服务端服务重启时,客户端如果没有重连机制,则无法再接收到订阅的消息。

使用的 Mqtt 组件为:M2Mqtt.Net.dll

一些特性发现

(1)如果提供的服务端地址是不可解析的,会引发异常无法实例化 MqttClient 对象。
(2)Connect 无法连接时会引发异常,IsConnected 为 false。
(3)服务端断开会触发客户端的 ConnectionClosed 事件,IsConnected 为 false。
(4)重新 Connect 需要重新 Subscribe 订阅主题。
(5)MqttClient.Subscribe 第一个参数为订阅主题数组,第二个为相应的 qosLevel,两个数组长度必须一致,否则会异常。

重连流程控制

主要代码实现

(1)线程主体

// 自动重连主体  private void _TryContinueConnect()  {      if (IsConnected) return;        Thread retryThread = new Thread(new ThreadStart(delegate      {          while (_MqttClient == null || !_MqttClient.IsConnected)          {              if (_ToClose) break;                if (_MqttClient == null)              {                  _BuildClient();                  Thread.Sleep(3000);                  continue;              }                try              {                  _TryCount++;                  _Connect();              }              catch (Exception ce)              {                  Debug.WriteLine("re connect exception:" + ce.Message);              }                // 如果还没连接不符合结束条件则睡2秒              if (!_MqttClient.IsConnected)              {                  Thread.Sleep(2000);              }          }      }));        retryThread.Start();  }

(2)实例化部分

// 实例化客户端  private void _BuildClient()  {      try      {          _MqttClient = new MqttClient(_MqttServer);      }      catch (Exception e)      {          Debug.WriteLine("build client error:" + e.Message);          return;      }        // 消息到达事件绑定      _MqttClient.MqttMsgPublishReceived += client_MqttMsgPublishReceived;        // 连接断开事件绑定      _MqttClient.ConnectionClosed += (sender, e) =>      {          if (!_ToClose)          {              // 尝试重连              _TryContinueConnect();          }      };  }

(3)尝试连接部分

// 发起一次连接,连接成功则订阅相关主题  private void _Connect()  {      if (String.IsNullOrEmpty(_MqttUsername))      {          var b = _MqttClient.Connect(_MqttClientId);      }      else      {          var b = _MqttClient.Connect(_MqttClientId, _MqttUsername, _MqttUserpass);      }        if (_MqttClient.IsConnected)      {          _MqttClient.Subscribe(new string[] { "topic1", "topic2" },              new byte[] { MqttMsgBase.QOS_LEVEL_AT_MOST_ONCE, MqttMsgBase.QOS_LEVEL_AT_MOST_ONCE });      }  }

实测效果不错,其中延时时间可以适当调整。