C#实现MQTT服务端与客户端通讯功能
时间：2023-02-06 10:30:13 | 栏目：.NET代码 | 点击：次
关于MQTT
MQTT(消息队列遥测传输)是ISO 标准(ISO/IEC PRF 20922)下基于发布/订阅范式的消息协议。它工作在 TCP/IP协议族上，是为硬件性能低下的远程设备以及网络状况糟糕的情况下而设计的发布/订阅型消息协议，为此，它需要一个消息中间件 。
MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。
MQTT示例
注: 该示例演示统一使用WPF, 简单MVVM模式演示, 复制代码需注意引用 NuGet包 GalaSoft
MQTT服务端建立：
演示界面:
演示代码：
public class MainViewModel : ViewModelBase
{
/// <summary>
/// Initializes a new instance of the MainViewModel class.
/// </summary>
public MainViewModel()
{
ClientInsTances = new ObservableCollection<ClientInstance>();
}
IMqttServer mqttServer; //MQTT服务端实例
string message;
/// <summary>
/// 消息 用于界面显示
/// </summary>
public string Message
{
get { return message; }
set { message = value; RaisePropertyChanged(); }
}
ObservableCollection<ClientInstance> clientInstances; //客户端登陆缓存信息
/// <summary>
/// 客户端实例
/// </summary>
public ObservableCollection<ClientInstance> ClientInsTances
{
get { return clientInstances; }
set { clientInstances = value; RaisePropertyChanged(); }
}
//开启MQTT服务
public void OpenMqttServer()
{
mqttServer = new MqttFactory().CreateMqttServer();
var options = new MqttServerOptions();
//拦截登录
options.ConnectionValidator = c =>
{
try
{
Message += string.Format("用户尝试登录:用户ID：{0}\t用户信息：{1}\t用户密码：{2}", c.ClientId, c.Username, c.Password) + "\r\n";
if (string.IsNullOrWhiteSpace(c.Username))
{
Message += string.Format("用户：{0}登录失败，用户信息为空", c.ClientId) + "\r\n";
c.ReturnCode = MQTTnet.Protocol.MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
return;
}
//解析用户名和密码，这个地方需要改成查找我们自己创建的用户名和密码。
if (c.Username == "admin" && c.Password == "123456")
{
c.ReturnCode = MqttConnectReturnCode.ConnectionAccepted;
Message += c.ClientId + " 登录成功" + "\r\n";
ClientInsTances.Add(new ClientInstance()
{
ClientID = c.ClientId,
UserName = c.Username,
PassWord = c.Password
});
return;
}
else
{
c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
Message += "用户名密码错误登陆失败" + "\r\n";
return;
}
}
catch (Exception ex)
{
Console.WriteLine("登录失败:" + ex.Message);
c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedIdentifierRejected;
return;
}
};
//拦截订阅
options.SubscriptionInterceptor = async context =>
{
try
{
Message += "用户" + context.ClientId + "订阅" + "\r\n";
}
catch (Exception ex)
{
Console.WriteLine("订阅失败:" + ex.Message);
context.AcceptSubscription = false;
}
};
//拦截消息
options.ApplicationMessageInterceptor = context =>
{
try
{
//一般不需要处理消息拦截
// Console.WriteLine(Encoding.UTF8.GetString(context.ApplicationMessage.Payload));
}
catch (Exception ex)
{
Console.WriteLine("消息拦截:" + ex.Message);
}
};
mqttServer.ClientDisconnected += ClientDisconnected;
mqttServer.ClientConnected += MqttServer_ClientConnected;
mqttServer.Started += MqttServer_Started;
mqttServer.StartAsync(options);
}
private void MqttServer_Started(object sender, EventArgs e)
{
Message += "消息服务启动成功：任意键退出" + "\r\n";
}
private void MqttServer_ClientConnected(object sender, MqttClientConnectedEventArgs e)
{
//客户端链接
Message += e.ClientId + "连接" + "\r\n";
}
private void ClientDisconnected(object sender, MqttClientDisconnectedEventArgs e)
{
//客户端断开
Message += e.ClientId + "断开" + "\r\n";
}
/// <summary>
/// 客户端推送信息 - 用于测试服务推送
/// </summary>
/// <param name="clientID"></param>
/// <param name="message"></param>
public void SendMessage(string clientID, string message)
{
mqttServer.PublishAsync(new MqttApplicationMessage
{
Topic = clientID,
QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce,
Retain = false,
Payload = Encoding.UTF8.GetBytes(message),
});
}
}
添加MQTT 客户端登陆实例, 用于保存客户的登陆信息,如下：
演示界面：
/// <summary>
/// 登陆客户端信息
/// </summary>
public class ClientInstance : ViewModelBase
{
private string clientID;
private string userName;
private string passWord;
/// <summary>
/// 识别ID
/// </summary>
public string ClientID
{
get { return clientID; }
set { clientID = value; RaisePropertyChanged(); }
}
/// <summary>
/// 账户
/// </summary>
public string UserName
{
get { return userName; }
set { userName = value; RaisePropertyChanged(); }
}
/// <summary>
/// 密码
/// </summary>
public string PassWord
{
get { return passWord; }
set { passWord = value; RaisePropertyChanged(); }
}
}
MQTT客户端建立：
演示代码:
public class MainViewModel : ViewModelBase
{
/// <summary>
/// Initializes a new instance of the MainViewModel class.
/// </summary>
public MainViewModel()
{
clientID = new Random().Next(999, 9999) + ""; //测试随机生成ClientID
}
IMqttClient mqttClient; //MQTT客户端实例
string clientID; //机器ID
string message;
public string Message //用于接收当前 消息
{
get { return message; }
set { message = value; RaisePropertyChanged(); }
}
//开启MQTT连接
public async void SignMqttServer()
{
var options = new MqttClientOptionsBuilder()
.WithClientId(clientID) //传递ClientID
.WithTcpServer("127.0.0.1", 1883) //MQTT服务的地址
.WithCredentials("admin", "123456") //传递账号密码
.WithCleanSession()
.Build();
mqttClient = new MqttFactory().CreateMqttClient();// .CreateManagedMqttClient();
mqttClient.Connected += MqttClient_Connected;
mqttClient.Disconnected += MqttClient_Disconnected;
mqttClient.ApplicationMessageReceived += MqttClient_ApplicationMessageReceived; //创建消息接受事件
await mqttClient.ConnectAsync(options);
//await mqttClient.SubscribeAsync(clientID);
}
private void MqttClient_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
{
Message += "收到的信息:" + Encoding.UTF8.GetString(e.ApplicationMessage.Payload) + "\r\n";
}
private void MqttClient_Disconnected(object sender, MqttClientDisconnectedEventArgs e)
{
Message += "客户端断开";
}
private void MqttClient_Connected(object sender, MqttClientConnectedEventArgs e)
{
Message += "客户端已连接" + "\r\n";
mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(clientID).Build()); //关联服务端订阅, 用于接受服务端推送信息
}
}
演示界面：
实际演示效果(GIF)
