C#编写上位机连接华为云平台IoTDA

需求

我们在平时的物联网开发中，往往需要用到云平台来收发数据。而光有云平台往往是不行的，我们还需要有上位机。那么怎样通过上位机连接到云平台，并且收发数据呢？本文将介绍如何通过C#编写的上位机和华为云平台通过物联网最常用MQTT协议进行连接并收发数据。介绍设备通MQTTS/MQTT协议接入平台，通过平台接口实现“数据上报”、“命令下发”的功能。

前期准备

1. VS2019
2. 华为云平台

具体设计 代码目录简述：

1. App.config：Server地址和设备信息配置文件
2. C#：项目C#代码；
3. EncryptUtil.cs：设备密钥加密辅助类；
4. FrmMqttDemo.cs：窗体界面；
5. Program.cs：Demo程序启动入口。
6. dll：项目中使用到了第三方库
7. MQTTnet：v3.0.11，是一个基于 MQTT 通信的高性能 .NET 开源库，它同时支持MQTT 服务器端和客户端，引用库文件包含MQTTnet.dll。
8. MQTTnet.Extensions.ManagedClient：v3.0.11，这是一个扩展库，它使用MQTTnet为托管MQTT客户机提供附加功能。

工程配置参数

App.config：需要配置服务器地址、设备ID和设备密钥，用于启动Demo程序的时 候，程序将此信息自动写到Demo主界面。

具体程序 App.config

本文件中存放的是Server地址和设备信息配置文件，我们每个人的程序主要的不同就是这个文件，虽然后面也可以在运行的时候改，但最好提前写入。

我们将自己的自己的设备ID和自己的设备密钥填入其中，选择相应的语言，最好填中文，毕竟看的方便。

主程序 连接服务器

				try
            {
                int portIsSsl = int.Parse(ConfigurationManager.AppSettings["portIsSsl"]);
                int portNotSsl = int.Parse(ConfigurationManager.AppSettings["portNotSsl"]);

                if (client == null)
                {
                    client = new MqttFactory().CreateManagedMqttClient();
                }

                string timestamp = DateTime.Now.ToString("yyyyMMddHH");
                string clientID = txtDeviceId.Text + "_0_0_" + timestamp;

                // 对密码进行HmacSHA256加密
                string secret = string.Empty;
                if (!string.IsNullOrEmpty(txtDeviceSecret.Text))
                {
                    secret = EncryptUtil.HmacSHA256(txtDeviceSecret.Text, timestamp);
                }

                // 判断是否为安全连接
                if (!cbSSLConnect.Checked)
                {
                    options = new ManagedMqttClientOptionsBuilder()
                    .WithAutoReconnectDelay(TimeSpan.FromSeconds(RECONNECT_TIME))
                    .WithClientOptions(new MqttClientOptionsBuilder()
                        .WithTcpServer(txtServerUri.Text, portNotSsl)
                        .WithCommunicationTimeout(TimeSpan.FromSeconds(DEFAULT_CONNECT_TIMEOUT))
                        .WithCredentials(txtDeviceId.Text, secret)
                        .WithClientId(clientID)
                        .WithKeepAlivePeriod(TimeSpan.FromSeconds(DEFAULT_KEEPLIVE))
                        .WithCleanSession(false)
                        .WithProtocolVersion(MqttProtocolVersion.V311)
                        .Build())
                    .Build();
                }
                else
                {
                    string caCertPath = Environment.CurrentDirectory + @"\certificate\rootcert.pem";
                    X509Certificate2 crt = new X509Certificate2(caCertPath);

                    options = new ManagedMqttClientOptionsBuilder()
                    .WithAutoReconnectDelay(TimeSpan.FromSeconds(RECONNECT_TIME))
                    .WithClientOptions(new MqttClientOptionsBuilder()
                        .WithTcpServer(txtServerUri.Text, portIsSsl)
                        .WithCommunicationTimeout(TimeSpan.FromSeconds(DEFAULT_CONNECT_TIMEOUT))
                        .WithCredentials(txtDeviceId.Text, secret)
                        .WithClientId(clientID)
                        .WithKeepAlivePeriod(TimeSpan.FromSeconds(DEFAULT_KEEPLIVE))
                        .WithCleanSession(false)
                        .WithTls(new MqttClientOptionsBuilderTlsParameters()
                        {
                            AllowUntrustedCertificates = true,
                            UseTls = true,
                            Certificates = new List { crt },
                            CertificateValidationHandler = delegate { return true; },
                            IgnoreCertificateChainErrors = false,
                            IgnoreCertificateRevocationErrors = false
                        })
                        .WithProtocolVersion(MqttProtocolVersion.V311)
                        .Build())
                    .Build();
                }

                Invoke((new Action(() =>
                {
                    ShowLogs($"{"try to connect to server " + txtServerUri.Text}{Environment.NewLine}");
                })));

                if (client.IsStarted)
                {
                    await client.StopAsync();
                }

                // 注册事件
                client.ApplicationMessageProcessedHandler = new ApplicationMessageProcessedHandlerDelegate(new Action(ApplicationMessageProcessedHandlerMethod)); // 消息发布回调

                client.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(new Action(MqttApplicationMessageReceived)); // 命令下发回调

                client.ConnectedHandler = new MqttClientConnectedHandlerDelegate(new Action(OnMqttClientConnected)); // 连接成功回调

                client.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(new Action(OnMqttClientDisconnected)); // 连接断开回调

                // 连接平台设备
                await client.StartAsync(options);

            }
            catch (Exception ex)
            {
                Invoke((new Action(() =>
                {
                    ShowLogs($"connect to mqtt server fail" + Environment.NewLine);
                })));
            }

接收到消息

Invoke((new Action(() =>
            {
                ShowLogs($"received message is {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}{Environment.NewLine}");

                string msg = "{\"result_code\": 0,\"response_name\": \"COMMAND_RESPONSE\",\"paras\": {\"result\": \"success\"}}";

                string topic = "$oc/devices/" + txtDeviceId.Text + "/sys/commands/response/request_id=" + e.ApplicationMessage.Topic.Split('=')[1];

                ShowLogs($"{"response message msg = " + msg}{Environment.NewLine}");
                
                var appMsg = new MqttApplicationMessage();
                appMsg.Payload = Encoding.UTF8.GetBytes(msg);
                appMsg.Topic = topic;
                appMsg.QualityOfServiceLevel = int.Parse(cbOosSelect.SelectedValue.ToString()) == 0 ? MqttQualityOfServiceLevel.AtMostOnce : MqttQualityOfServiceLevel.AtLeastOnce;
                appMsg.Retain = false;

                // 上行响应
                client.PublishAsync(appMsg).Wait();
            })));

消息发布回调

 try
            {
                if (e.HasFailed)
                {
                    Invoke((new Action(() =>
                    {
                        ShowLogs("publish messageId is " + e.ApplicationMessage.Id + ", topic: " + e.ApplicationMessage.ApplicationMessage.Topic + ", payload: " + Encoding.UTF8.GetString(e.ApplicationMessage.ApplicationMessage.Payload) + " is published fail");
                    })));
                }
                else if (e.HasSucceeded)
                {
                    Invoke((new Action(() =>
                    {
                        ShowLogs("publish messageId " + e.ApplicationMessage.Id + ", topic: " + e.ApplicationMessage.ApplicationMessage.Topic + ", payload: " + Encoding.UTF8.GetString(e.ApplicationMessage.ApplicationMessage.Payload) + " is published success");
                    })));
                }
            }
            catch (Exception ex)
            {
                Invoke((new Action(() =>
                {
                    ShowLogs("mqtt demo message publish error: " + ex.Message + Environment.NewLine);
                })));
            }

服务器连接成功

        Invoke((new Action(() =>
        {
            ShowLogs("connect to mqtt server success, deviceId is " + txtDeviceId.Text + Environment.NewLine);

            btnConnect.Enabled = false;
            btnDisconnect.Enabled = true;
            btnPublish.Enabled = true;
            btnSubscribe.Enabled = true;
        })));

断开服务器连接

        try {
            Invoke((new Action(() =>
            {
                ShowLogs("mqtt server is disconnected" + Environment.NewLine);

                txtSubTopic.Enabled = true;
                btnConnect.Enabled = true;
                btnDisconnect.Enabled = false;
                btnPublish.Enabled = false;
                btnSubscribe.Enabled = false;
            })));
            
            if (cbReconnect.Checked)
            {
                Invoke((new Action(() =>
                {
                    ShowLogs("reconnect is starting" + Environment.NewLine);
                })));

                //退避重连
                int lowBound = (int)(defaultBackoff * 0.8);
                int highBound = (int)(defaultBackoff * 1.2);
                long randomBackOff = random.Next(highBound - lowBound);
                long backOffWithJitter = (int)(Math.Pow(2.0, retryTimes)) * (randomBackOff + lowBound);
                long waitTImeUtilNextRetry = (int)(minBackoff + backOffWithJitter) > maxBackoff ? maxBackoff : (minBackoff + backOffWithJitter);

                Invoke((new Action(() =>
                {
                    ShowLogs("next retry time: " + waitTImeUtilNextRetry + Environment.NewLine);
                })));

                Thread.Sleep((int)waitTImeUtilNextRetry);

                retryTimes++;

                Task.Run(async () => { await ConnectMqttServerAsync(); });
            }
        }
        catch (Exception ex)
        {
            Invoke((new Action(() =>
            {
                ShowLogs("mqtt demo error: " + ex.Message + Environment.NewLine);
            })));
        }

格式

设备端上传到云平台 Topic：$oc/devices/{device_id}/sys/properties/report 数据格式：

数据格式：  
{
    "services": [{
            "service_id": "Temperature",
            "properties": {
                "smoke": 57,
                "temperature": 60,
                "humidity":78.37673
            },
            "event_time": "20151212T121212Z"
        }
    ]
}

运行界面

当我发送如下数据时：

{
    "services": [{
            "service_id": "BS",
            "properties": {
                "smoke": 57,
                "temperature": 60,
                "humidity":78.37673
            },
            "event_time": "20151212T121212Z"
        }
    ]
}

后续

欢迎关注我的毕业设计专栏。 关注微信公众号，发送“C#连接华为云平台”获取源码。

编写不易，感谢支持。