最近做了应该MQTT通讯的服务&＃xff0c;参考了网上的一些例子&＃xff0c;代码分享一下。
MQTT通讯帮助类&＃xff1a;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using MQTTnet;
using MQTTnet.Core;
using MQTTnet.Core.Client;
using MQTTnet.Core.Packets;
using MQTTnet.Core.Protocol;
using Newtonsoft.Json;
namespace MQTTService.Common
{
///
/// MQTT通讯
///
public partial class MQTTnetHelper : BaseJobLog
{
private MqttClient mqttClient &＃61; null;
public bool ConnectState { get { return mqttClient.IsConnected; } }
public async Task ConnectMqttServerAsync(string url,int? port, string clientId, string username, string password)
{
if (mqttClient &＃61;&＃61; null)
{
mqttClient &＃61; new MqttClientFactory().CreateMqttClient() as MqttClient;
mqttClient.ApplicationMessageReceived &＃43;&＃61; MqttClient_ApplicationMessageReceived;
mqttClient.Connected &＃43;&＃61; MqttClient_Connected;
mqttClient.Disconnected &＃43;&＃61; MqttClient_Disconnected;
}
try
{
var options &＃61; new MqttClientTcpOptions
{
Server &＃61; url,
Port&＃61; port,
ClientId &＃61; clientId,
UserName &＃61; username,
Password &＃61; password,
CleanSession &＃61; true
};
await mqttClient.ConnectAsync(options);
}
catch (Exception ex)
{
var msg &＃61;$"连接到MQTT服务器失败&＃xff01;" &＃43; Environment.NewLine &＃43; ex.Message &＃43; Environment.NewLine;
}
}
private void MqttClient_Connected(object sender, EventArgs e)
{
var msg &＃61; "已连接到MQTT服务器&＃xff01;" &＃43; Environment.NewLine;
}
private void MqttClient_Disconnected(object sender, EventArgs e)
{
var msg &＃61; "已断开MQTT连接&＃xff01;" &＃43; Environment.NewLine;
}
private void MqttClient_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
{
var msg &＃61; $">> {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}{Environment.NewLine}";
}
private void Subscribe_ClickAsync(object sender, EventArgs e)
{
mqttClient.SubscribeAsync(new List {
new TopicFilter("", MqttQualityOfServiceLevel.AtMostOnce)
});
}
public Task DataPublishAsync(string topic, object data)
{
var strdata &＃61; JsonConvert.SerializeObject(data);
var byteData &＃61; Encoding.UTF8.GetBytes(strdata);
var appMsg &＃61; new MqttApplicationMessage(topic, byteData, MqttQualityOfServiceLevel.AtMostOnce, false);
var result &＃61; mqttClient.PublishAsync(appMsg);
return result;
}
}
}
MQTT连接上报类&＃xff1a;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using MQTTService.Models;
using Newtonsoft.Json;
namespace MQTTService.Common
{
public class ConnectOrPushHelper : BaseJob
{
private MQTTnetHelper mqttHelper &＃61; new MQTTnetHelper();
///
/// MQTT连接
///
///
public bool MQTTConnect(DeviceAuthModel deviceAuthModel)
{
if (deviceAuthModel.IsNull()) return false;
//MQTT连接
var device &＃61; deviceAuthModel.deviceInfo;
var mq &＃61; deviceAuthModel.mqInfo;
var IP &＃61; CommonConfig.ApiIP_MQTT.IsNotNullOrWhiteSpace() ? CommonConfig.ApiIP_MQTT : mq.mqIp;
var port &＃61; CommonConfig.ApiPort_MQTT.IsNotNullOrWhiteSpace() ? CommonConfig.ApiPort_MQTT : mq.mqPort;
_log.Info(string.Format("MQTT连接&＃xff0c;IP&＃xff1a;{0},Port&＃xff1a;{1}&＃xff0c;clientId&＃xff1a;{2}&＃xff0c;mqUsername&＃xff1a;{3}&＃xff0c;mqPassword&＃xff1a;{4}", IP, port, device.deviceName &＃43; "&＃64;" &＃43; device.deviceSecret, mq.mqUsername, mq.mqPassword));
var res &＃61; mqttHelper.ConnectMqttServerAsync(IP, int.Parse(port), device.deviceName &＃43; "&＃64;" &＃43; device.deviceSecret, mq.mqUsername, mq.mqPassword);
System.Threading.Thread.Sleep(120);
if (!mqttHelper.ConnectState)
{
_log.Info("MQTT连接失败&＃xff0c;DeviceAuthModel&＃xff1a;" &＃43; JsonConvert.SerializeObject(deviceAuthModel));
return false;
}
else
{
return true;
}
}
///
/// MQTT上报
///
///
///
///
public bool Push(string url,string deviceName, T reportModel)
{
try
{
_log.Info(string.Format("MQTT推送数据&＃xff0c;url&＃xff1a;{0},数据&＃xff1a;{1}", url, JsonConvert.SerializeObject(reportModel)));
var result &＃61; mqttHelper.DataPublishAsync(url, reportModel);
if (result.Status.ToString() &＃61;&＃61; "Faulted")
{
_log.Info(string.Format("MQTT推送数据失败&＃xff0c;url&＃xff1a;{0}&＃xff0c;错误信息&＃xff1a;{1}&＃xff0c;推送数据&＃xff1a;{2}", url, result.Exception.Message, JsonConvert.SerializeObject(reportModel)));
return false;
}
System.Threading.Thread.Sleep(50);
return true;
}
catch (Exception ex)
{
_log.Info(string.Format("发布失败&＃xff0c;编号&＃xff1a;{0}&＃xff0c;异常信息&＃xff1a;{1}", deviceName, ex.Message));
return false;
}
}
}
}
Http请求帮助类&＃xff1a;
using log4net;
using Newtonsoft.Json;
using RestSharp;
using System;
using System.Diagnostics;
namespace MQTTService.Common
{
public class BaseJob
{
private string host;
public ILog _log;
public BaseJob()
{
_log &＃61; LogManager.GetLogger(typeof(BaseJob));
}
public string Post(string url, object data)
{
var client &＃61; new RestClient(url);
if (url.Contains("https://"))
System.Net.ServicePointManager.ServerCertificateValidationCallback &＃43;&＃61; (sender, certificate, chain, sslPolicyErrors) &＃61;> true; //SSL证书验证
var request &＃61; new RestRequest(Method.POST);
try
{
var str &＃61; JsonConvert.SerializeObject(data);
request.AddHeader("cache-control", "no-cache");
request.AddHeader("content-type", "application/json");
request.AddParameter("application/json", JsonConvert.SerializeObject(data), ParameterType.RequestBody);
Stopwatch sw &＃61; new Stopwatch();
sw.Start();
var response &＃61; client.Execute(request);
sw.Stop();
if(response.ErrorMessage.IsNotNullOrWhiteSpace()) _log.Info(url &＃43; "&＃xff0c;传参&＃xff1a;" &＃43; JsonConvert.SerializeObject(data) &＃43; "&＃xff0c;错误信息&＃xff1a;" &＃43; response.ErrorMessage&＃43; "&＃xff0c;接口执行时间&＃xff1a;" &＃43; sw.ElapsedMilliseconds);
return response.Content;
}
catch (Exception e)
{
_log.Error(e);
_log.Info(url &＃43; "&＃xff0c;传参&＃xff1a;" &＃43; JsonConvert.SerializeObject(request.Parameters));
return string.Empty;
}
}
public T Post(string url, object data) where T : new()
{
url &＃61; host &＃43; url;
var client &＃61; new RestClient(url);
var request &＃61; new RestRequest(Method.POST);
try
{
request.AddHeader("cache-control", "no-cache");
request.AddHeader("content-type", "application/json");
request.AddParameter("application/json", JsonConvert.SerializeObject(data), ParameterType.RequestBody);
Stopwatch sw &＃61; new Stopwatch();
sw.Start();
var response &＃61; client.Execute(request);
sw.Stop();
_log.Info(url &＃43; "&＃xff0c;传参&＃xff1a;" &＃43; JsonConvert.SerializeObject(data) &＃43; "&＃xff0c;返回结果&＃xff1a;" &＃43; response.Content &＃43; "&＃xff0c;接口执行时间&＃xff1a;" &＃43; sw.ElapsedMilliseconds);
return response.Data;
}
catch (Exception e)
{
_log.Error(e);
_log.Info(url &＃43; "&＃xff0c;传参&＃xff1a; " &＃43; JsonConvert.SerializeObject(request.Parameters));
return default(T);
}
}
}
}
京公网安备 11010802041100号