简介：面向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：消息可靠性的三档位
QoS（Quality of Service）是 MQTT 的重头戏，一共三档：
| QoS | 级别 | 语义 | 实际表现 | 适用场景 |
|---|---|---|---|---|
| 0 | 最多一次 | 发完即忘 | 可能丢消息，不重试 | 传感器高频上报、日志，丢了也能接受 |
| 1 | 至少一次 | 保证到达 | 会重试，可能重复 | 控制指令、状态变更，重复要自己做幂等 |
| 2 | 恰好一次 | 不丢不重 | 四次握手，开销大 | 计费、订单等关键业务 |
测试时一个常见误区是：默认用 QoS 0，本地测一切正常，一上真实网络就发现消息莫名其妙“消失”了。如果对可靠性有要求，至少得用 QoS 1。而使用 QoS 1 之后，又可能出现客户端收到重复消息的情况，这时候业务端要做幂等处理，这也是测试要覆盖的。
2.2 Retained：留给后来者的最后一条消息
MQTT 里有“保留消息”（Retain Flag）的概念。发布者发消息时把 Retain 位置 1，Broker 就会保存这条主题的最后一条消息。之后任何客户端订阅这个主题，会立刻收到这条保留消息。
这在设备上线初始化场景里特别有用，比如设备重连后不需要等下一个上报周期，就能立刻拿到最新状态。但测试时要小心：如果你改了配置，忘掉旧的保留消息还留在 Broker 上，新订阅者看到的还是旧数据，特别容易误判。
2.3 Will：异常掉线的最后遗言
Will Message（遗嘱消息）是 MQTT 一个很有特色的机制。客户端在连接时带上遗嘱主题和遗嘱消息，如果之后客户端没有正常发送 DISCONNECT 报文就断开连接（比如断电、网络断开、进程崩溃），Broker 就会代替客户端发布这条遗嘱消息。
测试遗嘱消息的推荐做法是：正常连上之后，直接杀掉进程，或者拔掉网线，看看 Broker 那边能不能收到遗嘱发布事件。这里有个坑值得注意——如果你在代码里主动调用DisconnectAsync，这是“优雅断开”，Broker 不会发遗嘱。很多人第一次测遗嘱消息发现没触发，就是这个原因。
2.4 CleanSession 与会话保持：断线重连到底剩什么
MQTT 客户端连接时可以设置 CleanSession。设成 true，每次连接都是全新会话，订阅全部清空；设成 false，Broker 会保留客户端的订阅信息以及离线期间的消息（前提是 QoS 大于 0）。这个细节对“断线重连后还要不要重新订阅”影响非常大。
如果你在测试中设置了 CleanSession = false 却没用固定的 ClientId，Broker 认不出同一个客户端，会话照样对不上号。正确姿势是：固定 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-8，JSON 序列化时也指定 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支持配置日志输出，能看到更底层的报文交互细节。如果哪天消息行为怎么都对不上，打开协议日志看原始报文，往往一眼就能定位问题。
本文还有配套的精品资源，点击获取
