基于MqttNet实现MQTT服务器接收客户端发布消息的方案与示例

需求可实现性 #

完全可以实现，MqttNet的新版本（v4.x及以上）只是调整了API命名和结构，替代了旧版的UseApplicationMessageReceivedHandler，提供了更灵活的事件订阅方式。

核心实现方案 #

在MqttNet服务器实例中，通过订阅ApplicationMessageReceivedAsync事件来捕获客户端发布的所有入站消息，包括传感器数据。

完整代码示例 #

using MqttNet;
using MqttNet.Server;

// 1. 创建服务器配置选项
var options = new MqttServerOptionsBuilder()
    .WithDefaultEndpoint() // 使用默认端口1883
    .Build();

// 2. 创建MqttServer实例
using var mqttServer = new MqttFactory().CreateMqttServer();

// 3. 订阅消息接收事件
mqttServer.ApplicationMessageReceivedAsync += async e =>
{
    // 获取发送消息的客户端ID
    var clientId = e.ClientId;
    // 获取消息主题
    var topic = e.ApplicationMessage.Topic;
    // 将消息负载转为字符串（传感器数据通常为二进制或JSON格式，按需解析）
    var payload = System.Text.Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment);
    
    // 自定义处理逻辑：比如存储到数据库、触发业务告警、转发到其他服务等
    Console.WriteLine($"收到客户端[{clientId}]的传感器数据：主题={topic}，内容={payload}");

    // 如需给客户端发送响应消息，可调用PublishAsync方法
    // await mqttServer.PublishAsync(new MqttApplicationMessageBuilder()
    //     .WithTopic($"response/{clientId}")
    //     .WithPayload("数据已接收")
    //     .Build());
};

// 4. 启动服务器
await mqttServer.StartAsync(options);
Console.WriteLine("MQTT服务器已启动，等待客户端消息...");

// 保持服务器运行（实际生产环境可结合服务托管框架）
Console.ReadLine();
await mqttServer.StopAsync();

关键说明 #

• ApplicationMessageReceivedAsync为异步事件，处理逻辑建议使用异步操作，避免阻塞服务器主线程
• e.ApplicationMessage包含完整消息元数据：主题、QoS级别、负载、保留标志、消息ID等
• 可通过e.ClientId识别客户端，结合服务器的客户端连接事件，实现客户端权限校验、会话管理等扩展功能
• 如需过滤特定主题的传感器数据，可在事件逻辑中添加主题判断，例如if (topic.StartsWith("sensor/temperature/"))

内容的提问来源于stack exchange，提问作者hmfarimani