\u53c3\u8003\u539f\u6587https:\/\/blog.csdn.net\/chenlu5201314\/article\/details\/94740765<\/p>\n\n\n\n
\u53c3\u8003\u539f\u6587https:\/\/zendei.com\/article\/79696.html<\/p>\n\n\n\n
<\/p>\n\n\n\n
\u975e\u540c\u6b65\u9023\u7dda\u65b9\u6cd5<\/p>\n\n\n\n
private async Task ConnectMqttServerAsync(string PROJECT_KEY)\n {\n\n try\n {\n if (client == null)\n {\n client = new MqttFactory().CreateManagedMqttClient();\n }\n\n string timestamp = DateTime.Now.ToString(\"yyyyMMddHH\");\n string clientID = \"JAX_\" + timestamp;\n\n options = new ManagedMqttClientOptionsBuilder()\n .WithAutoReconnectDelay(TimeSpan.FromSeconds(RECONNECT_TIME))\n .WithClientOptions(new MqttClientOptionsBuilder()\n .WithTcpServer(txtServerUri, portNotSsl)\n .WithCommunicationTimeout(TimeSpan.FromSeconds(DEFAULT_CONNECT_TIMEOUT))\n .WithCredentials(PROJECT_KEY, PROJECT_KEY)\n .WithClientId(clientID)\n .WithKeepAlivePeriod(TimeSpan.FromSeconds(DEFAULT_KEEPLIVE))\n .WithCleanSession(false)\n .WithProtocolVersion(MqttProtocolVersion.V311)\n .Build())\n .Build();\n\n\n Debug.WriteLine(\"\u5617\u8a66\u9023\u63a5SERVER\");\n\n if (client.IsStarted)\n {\n Debug.WriteLine(\"stopAsync\");\n client.ApplicationMessageProcessedHandler = null;\n client.ApplicationMessageReceivedHandler = null;\n client.ConnectedHandler = null;\n client.DisconnectedHandler = null;\n await client.StopAsync();\n }\n\n Debug.WriteLine(\"\u6ce8\u518c\u4e8b\u4ef6\");\n \/\/ \u6ce8\u518c\u4e8b\u4ef6\n client.ApplicationMessageProcessedHandler =\n new ApplicationMessageProcessedHandlerDelegate(\n new Action<ApplicationMessageProcessedEventArgs>(\n ApplicationMessageProcessedHandlerMethod)); \/\/ \u6d88\u606f\u53d1\u5e03\u56de\u8c03\n\n Debug.WriteLine(\"\u547d\u4ee4\u4e0b\u53d1\u56de\u8c03\");\n client.ApplicationMessageReceivedHandler =\n new MqttApplicationMessageReceivedHandlerDelegate(\n new Action<MqttApplicationMessageReceivedEventArgs>(\n MqttApplicationMessageReceived)); \/\/ \u547d\u4ee4\u4e0b\u53d1\u56de\u8c03\n\n Debug.WriteLine(\"\u8fde\u63a5\u6210\u529f\u56de\u8c03\");\n client.ConnectedHandler =\n new MqttClientConnectedHandlerDelegate(\n new Action<MqttClientConnectedEventArgs>(\n OnMqttClientConnected)); \/\/ \u8fde\u63a5\u6210\u529f\u56de\u8c03\n\n Debug.WriteLine(\"\u8fde\u63a5\u65ad\u5f00\u56de\u8c03\");\n client.DisconnectedHandler =\n new MqttClientDisconnectedHandlerDelegate(\n new Action<MqttClientDisconnectedEventArgs>(\n OnMqttClientDisconnected)); \/\/ \u8fde\u63a5\u65ad\u5f00\u56de\u8c03\n\n reConnnectPROJECTKEY = PROJECT_KEY;\n \/\/ \u8fde\u63a5\u5e73\u53f0\u8bbe\u5907\n await client.StartAsync(options);\n\n }\n catch (Exception ex)\n {\n Debug.WriteLine(\"connect to mqtt server fail\" + Environment.NewLine);\n }\n }\n<\/pre>\n\n\n\n\u5be6\u4f5c\u4e0a\u8ff0\u56db\u500bCallback<\/p>\n\n\n\n
\n private void ApplicationMessageProcessedHandlerMethod(ApplicationMessageProcessedEventArgs e)\n {\n try\n {\n if (e.HasFailed)\n {\n \/\/Debug.WriteLine(\"publish messageId is \" + e.ApplicationMessage.Id + \", topic: \" + e.ApplicationMessage.ApplicationMessage.Topic + \", payload: \" + Encoding.UTF8.GetString(e.ApplicationMessage.ApplicationMessage.Payload) + \" is published fail\");\n\n }\n else if (e.HasSucceeded)\n {\n \/\/Debug.WriteLine(\"publish messageId \" + e.ApplicationMessage.Id + \", topic: \" + e.ApplicationMessage.ApplicationMessage.Topic + \", payload: \" + Encoding.UTF8.GetString(e.ApplicationMessage.ApplicationMessage.Payload) + \" is published success\");\n\n }\n }\n catch (Exception ex)\n {\n Debug.WriteLine(\"mqtt demo message publish error: \" + ex.Message + Environment.NewLine);\n\n }\n\n }\n\n private void OnMqttClientConnected(MqttClientConnectedEventArgs e)\n {\n Debug.WriteLine(\"connect to mqtt server success \" + Environment.NewLine);\n status = Status.connected;\n WhenConnectedNotify();\n }\n\n private void OnMqttClientDisconnected(MqttClientDisconnectedEventArgs e)\n {\n status = Status.disconnected;\n try\n {\n Debug.WriteLine(\"mqtt server is disconnected\" + Environment.NewLine);\n Debug.WriteLine(\"reconnect is starting\" + Environment.NewLine);\n Thread.Sleep(1000);\n Task.Run(async () => { await ConnectMqttServerAsync(reConnnectPROJECTKEY); });\n\n }\n catch (Exception ex)\n {\n Debug.WriteLine(\"mqtt demo error: \" + ex.Message + Environment.NewLine);\n }\n }\n\n private void MqttApplicationMessageReceived(MqttApplicationMessageReceivedEventArgs e)\n {\n\n \/\/Debug.WriteLine($\"received message is {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}{Environment.NewLine}\");\n\n string msg = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);\n\n\n var appMsg = new MqttApplicationMessage();\n appMsg.Payload = Encoding.UTF8.GetBytes(msg);\n \/\/appMsg.Topic = txtSubTopic;\n appMsg.Topic = e.ApplicationMessage.Topic;\n\n appMsg.QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;\n appMsg.Retain = false;\n\n \/\/\/\/ \u4e0a\u884c\u54cd\u5e94\n client.PublishAsync(appMsg).Wait();\n\n \/*\n * \u89f8\u767c\u4e8b\u4ef6\uff0c\u50b3\u5165msg\u5230message\u4e8b\u4ef6\n *\/\n ReceiveMessageNotify(msg);\n\n }\n<\/pre>\n\n\n\n\u5c01\u88dd\u65b9\u6cd5\u7d66\u9023\u7dda\u3001\u65b7\u7dda\u4f7f\u7528<\/p>\n\n\n\n
\/*\n * Connect\/Disconnect\n *\/\n\n public void connect(string PROJECT_KEY)\n {\n Task.Run(async () => { await ConnectMqttServerAsync(PROJECT_KEY); });\n }\n public void disconnect()\n {\n Task.Run(async () => { await DisconnectMqttServerAsync(); });\n }\n\n private async Task DisconnectMqttServerAsync()\n {\n \/\/\u65ad\u5f00\u8fde\u63a5\n await client.StopAsync();\n }<\/pre>\n\n\n\n\u8a02\u95b1\u4e3b\u984c<\/p>\n\n\n\n
\/*\n * Subscribe\n *\/\n\n public void subscribe(List<string> sourceList)\n {\n \/\/ if (string.IsNullOrEmpty(txtSubTopic))\n if (sourceList.Count == 0)\n {\n Debug.WriteLine(\"\u8ba2\u9605\u4e3b\u9898\u4e0d\u80fd\u4e3a\u7a7a\uff01\");\n return;\n }\n\n if (!client.IsConnected)\n {\n Debug.WriteLine(\"MQTT\u5ba2\u6237\u7aef\u5c1a\u672a\u8fde\u63a5\uff01\");\n return;\n }\n\n \n var topicFilterBulderPreTopic = new MqttTopicFilterBuilder().WithTopic(\"NA\").Build();\n List<MqttTopicFilter> listTopic = new List<MqttTopicFilter>();\n\n for (int i = 0; i < sourceList.Count; i++)\n {\n string topictemp = \"topic\/rawdata\";\n topicFilterBulderPreTopic = new MqttTopicFilterBuilder().WithTopic(topictemp).Build();\n listTopic.Add(topicFilterBulderPreTopic);\n }\n\n \/\/ \u8ba2\u9605Topic\n client.SubscribeAsync(listTopic.ToArray()).Wait();\n\n Debug.WriteLine($\"topic : [{txtSubTopic}] is subscribe success\" + Environment.NewLine);\n\n }<\/pre>\n\n\n\n<\/p>","protected":false},"excerpt":{"rendered":"