简介面向C#物联网开发者的MQTT客户端与服务端测试资源包基于MQTTnet库实现适合需要掌握MQTT通信机制、快速搭建测试环境的初中级开发者。内置MQTTnet.dll运行库、可直接启动的exe程序、pdb调试符号、xml接口文档及config配置文件共5个文件、约145KB轻量紧凑便于下载后直接本地验证。已有906人学习下载在C#/MQTT主题下具备一定参考价值。包内样例能让读者直观观察客户端连接代理服务器、订阅主题、发布消息的完整处理流程通过exe与config组合快速启动借助pdb和xml可深入跟进MQTTnet API调用与调试排错。描述中涉及的QoS等级、异常处理、自动重连机制以及TLS/SSL安全连接等要点也可结合示例代码理解可靠通信与安全防护的落地方式。 前段时间接了个工业数据采集的活设备控制器支持 MQTT 上报我要做一套 C# 上位机去接收数据同时还要反向给设备下发指令。联调阶段开始之前我原本打算直接拿 MQTTX 这类现成工具测协议测到一半发现很多场景根本模拟不出来设备异常断电时遗嘱消息到底发没发客户端重连之后收不收得到断线期间的消息不同 QoS 级别下消息会不会重复这些问题用工具点几下是点不出结论的。所以我最后干脆在 C# 项目里用 MQTTnet 自己起了服务端又写了客户端把整个测试链路完全握在自己手里。这里说的“服务端”并不是去部署 EMQX、Mosquitto 这类独立 Broker而是在测试程序里用 MQTTnet 的 Server 能力起一个进程内的 MQTT Broker专门用来做验证。好处有几个可以精确控制 Broker 的行为比如查看每一条经过的消息、哪些客户端在连接、什么时候断开。测试代码和业务代码在同一套工程里能直接调试不用切来切去。不依赖外网也不用装额外的软件局域网里跑起来就能用。当然如果是正式环境要扛高并发、做集群还是得用专业的 Broker。这个方案定位是“联调测试和功能验证”不要拿它去当生产环境用。1. 为什么测试 MQTT 要客户端和服务端一起写先说结论MQTT 联调测试最大的难点不是不会用库而是出了问题不知道出在哪一层。客户端、服务端、网络、协议设置任何一个环节出错表象都是“消息没到”。如果你手里只有一个别人写的现成工具就只能干瞪眼。自己把客户端和服务端都写出来等于把每一层都变成可观察、可控制的。我这次场景很典型设备端是嵌入式控制器通过 MQTT 上报温度、湿度等运行数据上位机是 C# 写的一方面要订阅这些数据做展示和存储另一方面要发布控制指令给设备。两边都是 MQTT 客户端需要一个 Broker 中转。正式环境肯定用 EMQX但测试阶段我直接在开发机上跑一个进程内 Broker完全够用。用 C# 自己写测试客户端还有一个额外的好处能直接复用业务代码里的数据模型。比如 JSON payload 的序列化和反序列化在测试工程里引用同一个 DTO 类调试起来少走很多弯路。这是用通用测试工具做不到的。另外如果团队有多个人同时做联调这个方案也能快速铺开一个人把服务端和客户端的测试工程提交到代码库其他人克隆下来直接跑不用各自去下载配置工具。测试脚本还能沉淀成自动化测试以后每次改协议都有回归保障。2. MQTT 协议里决定测试成败的四个细节MQTT 看起来简单客户端连上 Broker订阅主题发布消息好像就完事了。但真正到了测试阶段能不能预期到消息的行为取决于你对协议细节的把握。以下几个点是我必须弄明白的。2.1 QoS消息可靠性的三档位QoSQuality of Service是 MQTT 的重头戏一共三档QoS级别语义实际表现适用场景0最多一次发完即忘可能丢消息不重试传感器高频上报、日志丢了也能接受1至少一次保证到达会重试可能重复控制指令、状态变更重复要自己做幂等2恰好一次不丢不重四次握手开销大计费、订单等关键业务测试时一个常见误区是默认用 QoS 0本地测一切正常一上真实网络就发现消息莫名其妙“消失”了。如果对可靠性有要求至少得用 QoS 1。而使用 QoS 1 之后又可能出现客户端收到重复消息的情况这时候业务端要做幂等处理这也是测试要覆盖的。2.2 Retained留给后来者的最后一条消息MQTT 里有“保留消息”Retain Flag的概念。发布者发消息时把 Retain 位置 1Broker 就会保存这条主题的最后一条消息。之后任何客户端订阅这个主题会立刻收到这条保留消息。这在设备上线初始化场景里特别有用比如设备重连后不需要等下一个上报周期就能立刻拿到最新状态。但测试时要小心如果你改了配置忘掉旧的保留消息还留在 Broker 上新订阅者看到的还是旧数据特别容易误判。2.3 Will异常掉线的最后遗言Will Message遗嘱消息是 MQTT 一个很有特色的机制。客户端在连接时带上遗嘱主题和遗嘱消息如果之后客户端没有正常发送 DISCONNECT 报文就断开连接比如断电、网络断开、进程崩溃Broker 就会代替客户端发布这条遗嘱消息。测试遗嘱消息的推荐做法是正常连上之后直接杀掉进程或者拔掉网线看看 Broker 那边能不能收到遗嘱发布事件。这里有个坑值得注意——如果你在代码里主动调用DisconnectAsync这是“优雅断开”Broker 不会发遗嘱。很多人第一次测遗嘱消息发现没触发就是这个原因。2.4 CleanSession 与会话保持断线重连到底剩什么MQTT 客户端连接时可以设置 CleanSession。设成 true每次连接都是全新会话订阅全部清空设成 falseBroker 会保留客户端的订阅信息以及离线期间的消息前提是 QoS 大于 0。这个细节对“断线重连后还要不要重新订阅”影响非常大。如果你在测试中设置了 CleanSession false 却没用固定的 ClientIdBroker 认不出同一个客户端会话照样对不上号。正确姿势是固定 ClientId CleanSession false才能在重连后恢复会话。3. 用 MQTTnet 把服务端跑在本地代码与配置MQTTnet 是目前 .NET 生态里最常用的 MQTT 库它不只是客户端也内置了完整的 Server 能力。NuGet 搜索 MQTTnet 安装即可我这边用的 4.x 版本。3.1 一段能直接跑的服务端代码using MQTTnet; using MQTTnet.Server; var mqttFactory new MqttFactory(); using var mqttServer mqttFactory.CreateMqttServer(); var mqttServerOptions new MqttServerOptionsBuilder() .WithDefaultEndpoint() .WithDefaultEndpointPort(1883) .Build(); mqttServer.ClientConnectedAsync e { Console.WriteLine($[{DateTime.Now:HH:mm:ss}] 客户端连接: {e.ClientId}); return Task.CompletedTask; }; mqttServer.ClientDisconnectedAsync e { Console.WriteLine($[{DateTime.Now:HH:mm:ss}] 客户端断开: {e.ClientId}); return Task.CompletedTask; }; mqttServer.InterceptingPublishAsync e { var payload e.ApplicationMessage.ConvertPayloadToString(); Console.WriteLine($[{DateTime.Now:HH:mm:ss}] 收到发布: {e.ApplicationMessage.Topic} - {payload}); return Task.CompletedTask; }; await mqttServer.StartAsync(mqttServerOptions); Console.WriteLine(MQTT 服务端已启动监听 1883 端口...); Console.ReadLine(); await mqttServer.StopAsync();这段代码做的事非常直观起一个监听 1883 端口的 Broker然后把三件事打出来——有人连上来、有人断开、有人发消息。别小看这三个日志联调的时候你会感谢它们的。MQTTnet 4.x 的事件回调签名都是返回 Task 的委托所以你在写处理函数的时候记得return Task.CompletedTask;或者用 async 方法不然会留下隐患。这也是 3.x 升级到 4.x 之后很多人踩的第一个坑。3.2 服务端的进阶玩法除了打日志服务端还能做更多事情。比如用InterceptingSubscriptionAsync拦截订阅请求看看某个客户端请求订阅了哪些主题用ValidateConnectionAsync做简单的用户名密码校验模拟真实 Broker 的鉴权流程。我测试时最常用的是拦截订阅事件确认客户端的主题过滤器真的按预期发过来了这比猜测“客户端订阅了什么样的主题”靠谱得多。mqttServer.InterceptingSubscriptionAsync e { Console.WriteLine($[{DateTime.Now:HH:mm:ss}] 订阅请求: {e.ClientId} - {e.TopicFilter.Topic}); return Task.CompletedTask; };如果要在同一台机器上起多个 Broker 实例做隔离测试用WithDefaultEndpointPort指定不同端口就行比如一个监听 1883一个监听 1884两套互不干扰适合做转发测试或双 Broker 数据同步的验证。4. 客户端连接、订阅、发布一整套可直接抄的代码服务端有了接下来是客户端。这个客户端既用来模拟设备端上报数据也用来模拟应用端接收数据一套代码搞定两种角色。4.1 连接前的选项设置using MQTTnet; using MQTTnet.Client; var mqttFactory new MqttFactory(); using var mqttClient mqttFactory.CreateMqttClient(); var mqttClientOptions new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) .WithClientId(device-test-001) .WithCleanSession(true) .WithKeepAlivePeriod(TimeSpan.FromSeconds(30)) .Build(); var connectResult await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); if (connectResult.ResultCode ! MqttClientConnectResultCode.Success) { Console.WriteLine($连接失败: {connectResult.ResultCode}); }几点说明ClientId 一定要有而且同一时间不能有重复。两个客户端用同一个 ClientId 连接先连的那个会被 Broker 踢掉。KeepAlive 是心跳间隔默认 60 秒。网络环境不稳定时把这个值调小比如 15 秒或 30 秒能让 Broker 更快发现死连接。如果要做遗嘱消息用.WithWill(willMessage)在连接选项里带上。4.2 订阅主题var subscribeOptions new MqttClientSubscribeOptionsBuilder() .WithTopicFilter(device//data, MqttQualityOfServiceLevel.AtLeastOnce) .Build(); var subscribeResult await mqttClient.SubscribeAsync(subscribeOptions, CancellationToken.None); foreach (var item in subscribeResult.Items) { Console.WriteLine($订阅结果: {item.TopicFilter.Topic} - {item.ResultCode}); }订阅接口会返回每个主题过滤器的订阅结果码不是订阅完就拉倒。如果 Broker 拒绝了某个主题ResultCode 会给出原因。测试联调时养成“都打印出来看一眼”的习惯能少翻很多日志。4.3 接收消息与回调用法MQTTnet 4.x 版本里接收消息的入口是ApplicationMessageReceivedAsync事件mqttClient.ApplicationMessageReceivedAsync e { var topic e.ApplicationMessage.Topic; var payload e.ApplicationMessage.ConvertPayloadToString(); Console.WriteLine($[{DateTime.Now:HH:mm:ss}] 收到消息: {topic} - {payload}); return Task.CompletedTask; };注意这里从 3.x 的ApplicationMessageReceived事件改成了ApplicationMessageReceivedAsync而且参数类型也变了。如果你在网上搜到老代码照着敲会直接编译不过需要顺手改成 4.x 的写法。ConvertPayloadToString()默认按 UTF-8 解 payload中文内容也不会乱码。如果服务端发的是二进制数据直接访问e.ApplicationMessage.PayloadSegment拿原始字节数组即可。4.4 发布消息var payload {\deviceId\:\device-001\,\temperature\:23.6,\humidity\:45}; var message new MqttApplicationMessageBuilder() .WithTopic(device/device-001/data) .WithPayload(payload) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .Build(); await mqttClient.PublishAsync(message, CancellationToken.None);Payload 直接用 JSON 字符串是最常见的做法设备端、应用端都用 UTF-8 序列化和反序列化能最大程度避免编码问题。发布完成后最好把返回的MqttClientPublishResult打出来看下里面有ReasonCode能确认这条消息是否被 Broker 接受。4.5 断线重连的兜底逻辑联调测试里网络抖动、服务端重启都是常事。客户端代码里必须处理断线重连不然测试到一半客户端就“假死”了。我比较习惯用 MQTTnet 的DisconnectedAsync事件配合简单的重连循环mqttClient.DisconnectedAsync async e { Console.WriteLine(连接断开准备重连...); await Task.Delay(TimeSpan.FromSeconds(3)); for (int retry 1; retry 5; retry) { try { await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); Console.WriteLine(重连成功); break; } catch (Exception ex) { Console.WriteLine($第 {retry} 次重连失败: {ex.Message}); await Task.Delay(TimeSpan.FromSeconds(5)); } } };重连成功之后如果之前用的 CleanSession false 且 ClientId 固定订阅关系会自动恢复不需要重新订阅。这正好呼应了前面讲的会话保持机制。5. 实测中踩过的坑从连不上到消息静默丢失5.1 双客户端同 ID后连的挤掉先连的这是最容易踩的坑。某次测试我起了两个订阅端忘了改 ClientId日志里就出现了一个订阅端反复“连接 - 断开”的现象。查了半天才发现是第二个客户端用同一个 ClientId 把第一个顶掉了。MQTT 规范就是这么设计的每个 ClientId 在同一时间只能对应一个活动连接测试时一定要保证每个客户端 ClientId 唯一。5.2 QoS 0 消息在断线节点丢失且没有任何报错默认情况下客户端和服务端的收发都是 QoS 0。如果客户端在消息发布的瞬间处于断线状态或者 Broker 刚好在处理重连这条消息就丢了而且没有任何日志和异常。我第一次遇到时排查了很久最后在发布端把 QoS 改成 1问题立刻暴露出来——原来不是代码逻辑错是协议本身就不保证 QoS 0 的送达。5.3 保留消息残留导致的“幽灵数据”某次测试改了设备上报格式但老格式的保留消息还留在 Broker 上。结果新的订阅端一连上就收到一条老格式的 JSON解析直接抛异常。排查下来就是 Broker 上保留消息没清理。处理办法在测试开始时主动发一条空 payload 的保留消息把旧消息清掉或者先确认这个主题是否还有历史保留消息再决定要不要处理。5.4 中文 payload 乱码MQTT 的 payload 本身是二进制没有编码约定。如果发布端用的是本地默认编码订阅端又按 UTF-8 解中文就会乱码。最省心的做法是发布端和订阅端都统一使用 UTF-8JSON 序列化时也指定 UTF-8 编码两边对齐就不会出问题。5.5 服务端能连上但订阅不到消息这个现象通常是三个原因主题写错了、订阅和发布的主题层级对不上、或者用了通配符但理解错了匹配规则。device//data只匹配device/xxx/data这种结构而device/#才能匹配device下面的任意层级。测试时把服务端的InterceptingSubscriptionAsync日志打出来看订阅过滤器到底发的什么内容一目了然。5.6 KeepAlive 与网络设备超时打架如果路由器或防火墙的空闲超时时间比 KeepAlive 短连接会在没发心跳的空档被切断。把 KeepAlive 调短一些比如 15 秒能减少这种情况。另外 Broker 端也会根据 KeepAlive 判断客户端是否存活如果客户端设置太长甚至不设断线后 Broker 要等很久才能感知遗嘱消息也会延迟触发。5.7 排查路径总结我遇到问题后的排查顺序基本是先看 Broker 日志确认客户端有没有连上、有没有订阅再确认发布端有没有消息经过 Broker然后看订阅端有没有收到。三步日志一对照问题出在哪一段就非常清楚了。这也是我坚持“客户端和服务端都自己写”的原因——每一层的行为都在掌控里不存在黑盒。最后再分享一个小技巧测试完记得把 Broker 的日志级别调出来MQTTnet 的MqttServerOptions支持配置日志输出能看到更底层的报文交互细节。如果哪天消息行为怎么都对不上打开协议日志看原始报文往往一眼就能定位问题。本文还有配套的精品资源点击获取 C#上位机用MQTTnet自建MQTT服务端：联调测试与协议细节全解析
简介面向C#物联网开发者的MQTT客户端与服务端测试资源包基于MQTTnet库实现适合需要掌握MQTT通信机制、快速搭建测试环境的初中级开发者。内置MQTTnet.dll运行库、可直接启动的exe程序、pdb调试符号、xml接口文档及config配置文件共5个文件、约145KB轻量紧凑便于下载后直接本地验证。已有906人学习下载在C#/MQTT主题下具备一定参考价值。包内样例能让读者直观观察客户端连接代理服务器、订阅主题、发布消息的完整处理流程通过exe与config组合快速启动借助pdb和xml可深入跟进MQTTnet API调用与调试排错。描述中涉及的QoS等级、异常处理、自动重连机制以及TLS/SSL安全连接等要点也可结合示例代码理解可靠通信与安全防护的落地方式。 前段时间接了个工业数据采集的活设备控制器支持 MQTT 上报我要做一套 C# 上位机去接收数据同时还要反向给设备下发指令。联调阶段开始之前我原本打算直接拿 MQTTX 这类现成工具测协议测到一半发现很多场景根本模拟不出来设备异常断电时遗嘱消息到底发没发客户端重连之后收不收得到断线期间的消息不同 QoS 级别下消息会不会重复这些问题用工具点几下是点不出结论的。所以我最后干脆在 C# 项目里用 MQTTnet 自己起了服务端又写了客户端把整个测试链路完全握在自己手里。这里说的“服务端”并不是去部署 EMQX、Mosquitto 这类独立 Broker而是在测试程序里用 MQTTnet 的 Server 能力起一个进程内的 MQTT Broker专门用来做验证。好处有几个可以精确控制 Broker 的行为比如查看每一条经过的消息、哪些客户端在连接、什么时候断开。测试代码和业务代码在同一套工程里能直接调试不用切来切去。不依赖外网也不用装额外的软件局域网里跑起来就能用。当然如果是正式环境要扛高并发、做集群还是得用专业的 Broker。这个方案定位是“联调测试和功能验证”不要拿它去当生产环境用。1. 为什么测试 MQTT 要客户端和服务端一起写先说结论MQTT 联调测试最大的难点不是不会用库而是出了问题不知道出在哪一层。客户端、服务端、网络、协议设置任何一个环节出错表象都是“消息没到”。如果你手里只有一个别人写的现成工具就只能干瞪眼。自己把客户端和服务端都写出来等于把每一层都变成可观察、可控制的。我这次场景很典型设备端是嵌入式控制器通过 MQTT 上报温度、湿度等运行数据上位机是 C# 写的一方面要订阅这些数据做展示和存储另一方面要发布控制指令给设备。两边都是 MQTT 客户端需要一个 Broker 中转。正式环境肯定用 EMQX但测试阶段我直接在开发机上跑一个进程内 Broker完全够用。用 C# 自己写测试客户端还有一个额外的好处能直接复用业务代码里的数据模型。比如 JSON payload 的序列化和反序列化在测试工程里引用同一个 DTO 类调试起来少走很多弯路。这是用通用测试工具做不到的。另外如果团队有多个人同时做联调这个方案也能快速铺开一个人把服务端和客户端的测试工程提交到代码库其他人克隆下来直接跑不用各自去下载配置工具。测试脚本还能沉淀成自动化测试以后每次改协议都有回归保障。2. MQTT 协议里决定测试成败的四个细节MQTT 看起来简单客户端连上 Broker订阅主题发布消息好像就完事了。但真正到了测试阶段能不能预期到消息的行为取决于你对协议细节的把握。以下几个点是我必须弄明白的。2.1 QoS消息可靠性的三档位QoSQuality of Service是 MQTT 的重头戏一共三档QoS级别语义实际表现适用场景0最多一次发完即忘可能丢消息不重试传感器高频上报、日志丢了也能接受1至少一次保证到达会重试可能重复控制指令、状态变更重复要自己做幂等2恰好一次不丢不重四次握手开销大计费、订单等关键业务测试时一个常见误区是默认用 QoS 0本地测一切正常一上真实网络就发现消息莫名其妙“消失”了。如果对可靠性有要求至少得用 QoS 1。而使用 QoS 1 之后又可能出现客户端收到重复消息的情况这时候业务端要做幂等处理这也是测试要覆盖的。2.2 Retained留给后来者的最后一条消息MQTT 里有“保留消息”Retain Flag的概念。发布者发消息时把 Retain 位置 1Broker 就会保存这条主题的最后一条消息。之后任何客户端订阅这个主题会立刻收到这条保留消息。这在设备上线初始化场景里特别有用比如设备重连后不需要等下一个上报周期就能立刻拿到最新状态。但测试时要小心如果你改了配置忘掉旧的保留消息还留在 Broker 上新订阅者看到的还是旧数据特别容易误判。2.3 Will异常掉线的最后遗言Will Message遗嘱消息是 MQTT 一个很有特色的机制。客户端在连接时带上遗嘱主题和遗嘱消息如果之后客户端没有正常发送 DISCONNECT 报文就断开连接比如断电、网络断开、进程崩溃Broker 就会代替客户端发布这条遗嘱消息。测试遗嘱消息的推荐做法是正常连上之后直接杀掉进程或者拔掉网线看看 Broker 那边能不能收到遗嘱发布事件。这里有个坑值得注意——如果你在代码里主动调用DisconnectAsync这是“优雅断开”Broker 不会发遗嘱。很多人第一次测遗嘱消息发现没触发就是这个原因。2.4 CleanSession 与会话保持断线重连到底剩什么MQTT 客户端连接时可以设置 CleanSession。设成 true每次连接都是全新会话订阅全部清空设成 falseBroker 会保留客户端的订阅信息以及离线期间的消息前提是 QoS 大于 0。这个细节对“断线重连后还要不要重新订阅”影响非常大。如果你在测试中设置了 CleanSession false 却没用固定的 ClientIdBroker 认不出同一个客户端会话照样对不上号。正确姿势是固定 ClientId CleanSession false才能在重连后恢复会话。3. 用 MQTTnet 把服务端跑在本地代码与配置MQTTnet 是目前 .NET 生态里最常用的 MQTT 库它不只是客户端也内置了完整的 Server 能力。NuGet 搜索 MQTTnet 安装即可我这边用的 4.x 版本。3.1 一段能直接跑的服务端代码using MQTTnet; using MQTTnet.Server; var mqttFactory new MqttFactory(); using var mqttServer mqttFactory.CreateMqttServer(); var mqttServerOptions new MqttServerOptionsBuilder() .WithDefaultEndpoint() .WithDefaultEndpointPort(1883) .Build(); mqttServer.ClientConnectedAsync e { Console.WriteLine($[{DateTime.Now:HH:mm:ss}] 客户端连接: {e.ClientId}); return Task.CompletedTask; }; mqttServer.ClientDisconnectedAsync e { Console.WriteLine($[{DateTime.Now:HH:mm:ss}] 客户端断开: {e.ClientId}); return Task.CompletedTask; }; mqttServer.InterceptingPublishAsync e { var payload e.ApplicationMessage.ConvertPayloadToString(); Console.WriteLine($[{DateTime.Now:HH:mm:ss}] 收到发布: {e.ApplicationMessage.Topic} - {payload}); return Task.CompletedTask; }; await mqttServer.StartAsync(mqttServerOptions); Console.WriteLine(MQTT 服务端已启动监听 1883 端口...); Console.ReadLine(); await mqttServer.StopAsync();这段代码做的事非常直观起一个监听 1883 端口的 Broker然后把三件事打出来——有人连上来、有人断开、有人发消息。别小看这三个日志联调的时候你会感谢它们的。MQTTnet 4.x 的事件回调签名都是返回 Task 的委托所以你在写处理函数的时候记得return Task.CompletedTask;或者用 async 方法不然会留下隐患。这也是 3.x 升级到 4.x 之后很多人踩的第一个坑。3.2 服务端的进阶玩法除了打日志服务端还能做更多事情。比如用InterceptingSubscriptionAsync拦截订阅请求看看某个客户端请求订阅了哪些主题用ValidateConnectionAsync做简单的用户名密码校验模拟真实 Broker 的鉴权流程。我测试时最常用的是拦截订阅事件确认客户端的主题过滤器真的按预期发过来了这比猜测“客户端订阅了什么样的主题”靠谱得多。mqttServer.InterceptingSubscriptionAsync e { Console.WriteLine($[{DateTime.Now:HH:mm:ss}] 订阅请求: {e.ClientId} - {e.TopicFilter.Topic}); return Task.CompletedTask; };如果要在同一台机器上起多个 Broker 实例做隔离测试用WithDefaultEndpointPort指定不同端口就行比如一个监听 1883一个监听 1884两套互不干扰适合做转发测试或双 Broker 数据同步的验证。4. 客户端连接、订阅、发布一整套可直接抄的代码服务端有了接下来是客户端。这个客户端既用来模拟设备端上报数据也用来模拟应用端接收数据一套代码搞定两种角色。4.1 连接前的选项设置using MQTTnet; using MQTTnet.Client; var mqttFactory new MqttFactory(); using var mqttClient mqttFactory.CreateMqttClient(); var mqttClientOptions new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) .WithClientId(device-test-001) .WithCleanSession(true) .WithKeepAlivePeriod(TimeSpan.FromSeconds(30)) .Build(); var connectResult await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); if (connectResult.ResultCode ! MqttClientConnectResultCode.Success) { Console.WriteLine($连接失败: {connectResult.ResultCode}); }几点说明ClientId 一定要有而且同一时间不能有重复。两个客户端用同一个 ClientId 连接先连的那个会被 Broker 踢掉。KeepAlive 是心跳间隔默认 60 秒。网络环境不稳定时把这个值调小比如 15 秒或 30 秒能让 Broker 更快发现死连接。如果要做遗嘱消息用.WithWill(willMessage)在连接选项里带上。4.2 订阅主题var subscribeOptions new MqttClientSubscribeOptionsBuilder() .WithTopicFilter(device//data, MqttQualityOfServiceLevel.AtLeastOnce) .Build(); var subscribeResult await mqttClient.SubscribeAsync(subscribeOptions, CancellationToken.None); foreach (var item in subscribeResult.Items) { Console.WriteLine($订阅结果: {item.TopicFilter.Topic} - {item.ResultCode}); }订阅接口会返回每个主题过滤器的订阅结果码不是订阅完就拉倒。如果 Broker 拒绝了某个主题ResultCode 会给出原因。测试联调时养成“都打印出来看一眼”的习惯能少翻很多日志。4.3 接收消息与回调用法MQTTnet 4.x 版本里接收消息的入口是ApplicationMessageReceivedAsync事件mqttClient.ApplicationMessageReceivedAsync e { var topic e.ApplicationMessage.Topic; var payload e.ApplicationMessage.ConvertPayloadToString(); Console.WriteLine($[{DateTime.Now:HH:mm:ss}] 收到消息: {topic} - {payload}); return Task.CompletedTask; };注意这里从 3.x 的ApplicationMessageReceived事件改成了ApplicationMessageReceivedAsync而且参数类型也变了。如果你在网上搜到老代码照着敲会直接编译不过需要顺手改成 4.x 的写法。ConvertPayloadToString()默认按 UTF-8 解 payload中文内容也不会乱码。如果服务端发的是二进制数据直接访问e.ApplicationMessage.PayloadSegment拿原始字节数组即可。4.4 发布消息var payload {\deviceId\:\device-001\,\temperature\:23.6,\humidity\:45}; var message new MqttApplicationMessageBuilder() .WithTopic(device/device-001/data) .WithPayload(payload) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .Build(); await mqttClient.PublishAsync(message, CancellationToken.None);Payload 直接用 JSON 字符串是最常见的做法设备端、应用端都用 UTF-8 序列化和反序列化能最大程度避免编码问题。发布完成后最好把返回的MqttClientPublishResult打出来看下里面有ReasonCode能确认这条消息是否被 Broker 接受。4.5 断线重连的兜底逻辑联调测试里网络抖动、服务端重启都是常事。客户端代码里必须处理断线重连不然测试到一半客户端就“假死”了。我比较习惯用 MQTTnet 的DisconnectedAsync事件配合简单的重连循环mqttClient.DisconnectedAsync async e { Console.WriteLine(连接断开准备重连...); await Task.Delay(TimeSpan.FromSeconds(3)); for (int retry 1; retry 5; retry) { try { await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); Console.WriteLine(重连成功); break; } catch (Exception ex) { Console.WriteLine($第 {retry} 次重连失败: {ex.Message}); await Task.Delay(TimeSpan.FromSeconds(5)); } } };重连成功之后如果之前用的 CleanSession false 且 ClientId 固定订阅关系会自动恢复不需要重新订阅。这正好呼应了前面讲的会话保持机制。5. 实测中踩过的坑从连不上到消息静默丢失5.1 双客户端同 ID后连的挤掉先连的这是最容易踩的坑。某次测试我起了两个订阅端忘了改 ClientId日志里就出现了一个订阅端反复“连接 - 断开”的现象。查了半天才发现是第二个客户端用同一个 ClientId 把第一个顶掉了。MQTT 规范就是这么设计的每个 ClientId 在同一时间只能对应一个活动连接测试时一定要保证每个客户端 ClientId 唯一。5.2 QoS 0 消息在断线节点丢失且没有任何报错默认情况下客户端和服务端的收发都是 QoS 0。如果客户端在消息发布的瞬间处于断线状态或者 Broker 刚好在处理重连这条消息就丢了而且没有任何日志和异常。我第一次遇到时排查了很久最后在发布端把 QoS 改成 1问题立刻暴露出来——原来不是代码逻辑错是协议本身就不保证 QoS 0 的送达。5.3 保留消息残留导致的“幽灵数据”某次测试改了设备上报格式但老格式的保留消息还留在 Broker 上。结果新的订阅端一连上就收到一条老格式的 JSON解析直接抛异常。排查下来就是 Broker 上保留消息没清理。处理办法在测试开始时主动发一条空 payload 的保留消息把旧消息清掉或者先确认这个主题是否还有历史保留消息再决定要不要处理。5.4 中文 payload 乱码MQTT 的 payload 本身是二进制没有编码约定。如果发布端用的是本地默认编码订阅端又按 UTF-8 解中文就会乱码。最省心的做法是发布端和订阅端都统一使用 UTF-8JSON 序列化时也指定 UTF-8 编码两边对齐就不会出问题。5.5 服务端能连上但订阅不到消息这个现象通常是三个原因主题写错了、订阅和发布的主题层级对不上、或者用了通配符但理解错了匹配规则。device//data只匹配device/xxx/data这种结构而device/#才能匹配device下面的任意层级。测试时把服务端的InterceptingSubscriptionAsync日志打出来看订阅过滤器到底发的什么内容一目了然。5.6 KeepAlive 与网络设备超时打架如果路由器或防火墙的空闲超时时间比 KeepAlive 短连接会在没发心跳的空档被切断。把 KeepAlive 调短一些比如 15 秒能减少这种情况。另外 Broker 端也会根据 KeepAlive 判断客户端是否存活如果客户端设置太长甚至不设断线后 Broker 要等很久才能感知遗嘱消息也会延迟触发。5.7 排查路径总结我遇到问题后的排查顺序基本是先看 Broker 日志确认客户端有没有连上、有没有订阅再确认发布端有没有消息经过 Broker然后看订阅端有没有收到。三步日志一对照问题出在哪一段就非常清楚了。这也是我坚持“客户端和服务端都自己写”的原因——每一层的行为都在掌控里不存在黑盒。最后再分享一个小技巧测试完记得把 Broker 的日志级别调出来MQTTnet 的MqttServerOptions支持配置日志输出能看到更底层的报文交互细节。如果哪天消息行为怎么都对不上打开协议日志看原始报文往往一眼就能定位问题。本文还有配套的精品资源点击获取