大家好,又见面了,我是你们的朋友全栈君。
服务端:
class Program
{
private static MqttServer mqttServer = null;
static void Main(string[] args)
{
MqttNetTrace.TraceMessagePublished += MqttNetTrace_TraceMessagePublished;
new Thread(StartMqttServer).Start();
while (true)
{
var inputString = Console.ReadLine().ToLower().Trim();
if (inputString == "exit")
{
mqttServer?.StopAsync();
Console.WriteLine("MQTT服务已停止!");
break;
}
else if (inputString == "clients")
{
foreach (var item in mqttServer.GetConnectedClients())
{
Console.WriteLine($"客户端标识:{item.ClientId},协议版本:{item.ProtocolVersion}");
}
}
else
{
Console.WriteLine($"命令[{inputString}]无效!");
}
}
}
private static void StartMqttServer()
{
if (mqttServer == null)
{
try
{
var options = new MqttServerOptions
{
ConnectionValidator = p =>
{
if (p.ClientId == "c001")
{
if (p.Username != "u001" || p.Password != "p001")
{
return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
}
}
return MqttConnectReturnCode.ConnectionAccepted;
}
};
mqttServer = new MqttServerFactory().CreateMqttServer(options) as MqttServer;
mqttServer.ApplicationMessageReceived += MqttServer_ApplicationMessageReceived;
mqttServer.ClientConnected += MqttServer_ClientConnected;
mqttServer.ClientDisconnected += MqttServer_ClientDisconnected;
}
catch (Exception ex)
{
Console.WriteLine(ex.Message);
return;
}
}
mqttServer.StartAsync();
Console.WriteLine("MQTT服务启动成功!");
}
private static void MqttServer_ClientConnected(object sender, MqttClientConnectedEventArgs e)
{
Console.WriteLine($"客户端[{e.Client.ClientId}]已连接,协议版本:{e.Client.ProtocolVersion}");
}
private static void MqttServer_ClientDisconnected(object sender, MqttClientDisconnectedEventArgs e)
{
Console.WriteLine($"客户端[{e.Client.ClientId}]已断开连接!");
}
private static void MqttServer_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
{
Console.WriteLine($"客户端[{e.ClientId}]>> 主题:{e.ApplicationMessage.Topic} 负荷:{Encoding.UTF8.GetString(e.ApplicationMessage.Payload)} Qos:{e.ApplicationMessage.QualityOfServiceLevel} 保留:{e.ApplicationMessage.Retain}");
}
private static void MqttNetTrace_TraceMessagePublished(object sender, MqttNetTraceMessagePublishedEventArgs e)
{
/*Console.WriteLine($">> 线程ID:{e.ThreadId} 来源:{e.Source} 跟踪级别:{e.Level} 消息: {e.Message}");
if (e.Exception != null)
{
Console.WriteLine(e.Exception);
}*/
}
}
客户端
class Program
{
private static MqttClient mqttClient = null;
static string topic = "测试1/测试2/测试3";
static void Main(string[] args)
{
mqttClient = new MqttClientFactory().CreateMqttClient() as MqttClient;
mqttClient.ApplicationMessageReceived += MqttClient_ApplicationMessageReceived;
mqttClient.Connected += MqttClient_Connected;
mqttClient.Disconnected += MqttClient_Disconnected;
var options = new MqttClientTcpOptions
{
Server = "127.0.0.1",
ClientId = Guid.NewGuid().ToString().Substring(0, 5),
UserName = "u001",
Password = "p001",
CleanSession = true
};
mqttClient.ConnectAsync(options);
Console.ReadLine();
Console.WriteLine("Hello World!");
}
private static void MqttClient_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
{
Console.WriteLine($">> {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}{Environment.NewLine}");
}
private static void MqttClient_Connected(object sender, EventArgs e)
{
Console.WriteLine("已连接到MQTT服务器!" + Environment.NewLine);
mqttClient.SubscribeAsync(new List<TopicFilter> {
new TopicFilter(topic, MqttQualityOfServiceLevel.AtMostOnce)
});
Console.WriteLine($"已订阅[{topic}]主题" + Environment.NewLine);
//开始发布消息
string inputString = "你好么";
var appMsg = new MqttApplicationMessage(topic, Encoding.UTF8.GetBytes(inputString), MqttQualityOfServiceLevel.AtMostOnce, false);
mqttClient.PublishAsync(appMsg);
}
private static void MqttClient_Disconnected(object sender, EventArgs e)
{
Console.WriteLine("已断开MQTT连接!" + Environment.NewLine);
}
}
源码下载地址:MQTT
MQTT Windows 端验证程序: http://mqttfx.jensd.de/index.php/download
https://download.csdn.net/download/kesshei/10906614
发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/152900.html原文链接:https://javaforall.cn
【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛
【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...