使用C#连接MQTT 进行数据接收和数据处理，QoS1 持久会话模式解决服务掉选 - ToB企服应用市场:ToB评测及商务社交产业平台

// 创建MQTT客户端工厂
var mqttFactory = new MqttFactory();
var mqttClient = mqttFactory.CreateMqttClient();
// 配置MQTT客户端选项
var options = new MqttClientOptionsBuilder()
.WithClientId("org-机构id-quickstart") // 修改客户端ID
.WithTcpServer("服务器地址", 1883) // 修改服务器地址，使用默认端口
.WithCredentials("org-机构id", "秘钥") // 修改用户名
.WithCleanSession(false) // 添加这行，不清除会话
.Build(); // 移除 TLS 配置，因为使用的是普通连接
try
{
// 添加连接状态处理
mqttClient.UseDisconnectedHandler(async e =>
{
Console.WriteLine("已断开连接！正在尝试重新连接...");
await Task.Delay(TimeSpan.FromSeconds(5));
try
{
await mqttClient.ConnectAsync(options);
}
catch
{
Console.WriteLine("重新连接失败");
}
});
// 修改订阅主题以匹配您的格式
string topic = "/device_sensor_data/机构id/+/+/+/+";
// 添加订阅处理
mqttClient.UseApplicationMessageReceivedHandler(e =>
{
string payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
string topic = e.ApplicationMessage.Topic ?? "";
// 解析主题数据
string[] topicParts = topic.Split('/');
//将解析后的出具进行处理<br>
});
// 修改订阅配置
var subscribeOptions = new MqttClientSubscribeOptionsBuilder()
.WithTopicFilter(topic, MqttQualityOfServiceLevel.AtLeastOnce)
.Build();
// 连接后进行订阅
await mqttClient.ConnectAsync(options);
await mqttClient.SubscribeAsync(subscribeOptions);
Console.WriteLine("已成功连接并订阅主题 (QoS1, 持久会话)");
// 保持程序运行
Console.WriteLine("按任意键退出...");
Console.ReadKey();
// 断开连接
await mqttClient.DisconnectAsync();
}
catch (Exception ex)
{
Console.WriteLine($"发生错误: {ex.Message}");
}

复制代码