简介：本资源是一套完整的MQTT协议C#实现教学与测试案例，面向物联网开发初学者、C#中级开发者及需要快速搭建轻量级消息通信系统的工程师。内容涵盖基于MQTTnet库的服务器端（Broker）与客户端（Client）双端可运行代码，深入演示发布/订阅机制、主题订阅、QoS消息控制、连接配置与事件回调等核心实践环节，适用于远程监控、嵌入式设备通信等低带宽场景。压缩包共2000个文件，以1095个XML配置与说明文件、809个隐藏系统文件（含调试与构建中间产物）、290个DLL依赖库为主，辅以14个关键CS源码文件、5个EXE可执行程序及完整SLN解决方案，整体大小42.99MB，结构体现典型.NET项目分层组织逻辑。目前已有693人学习下载，读者可直接编译运行服务端与多个客户端实例，观察实时消息流转过程，并通过源码注释与事件处理逻辑掌握MQTT在C#中的工程化落地方法。
1. 为什么一个 MQTT C# Demo 要同时跑通服务端和客户端？——不是“能连上”就叫通，而是“消息不丢、时序不错、断线能续”才算落地
很多刚接触工业物联网或边缘设备通信的开发者，拿到“MQTT C# demo”第一反应是：找一个现成客户端库连上公开 Broker（比如 test.mosquitto.org），发几条Hello World，截图控制台输出，任务就算完成了。但真实产线里，这种“能连上”的 demo 一上环境就翻车：设备重启后订阅丢失、QoS1 消息重复消费、客户端断网重连时堆积大量未确认报文、服务端日志里满屏CONNACK return code: 5……根本原因在于——MQTT 不是 HTTP，它是一套有状态、讲时序、需双向保活的轻量级发布/订阅协议。只测客户端，等于只验了半边协议栈；不带自建服务端压测，你永远不知道Clean Session = false在断线重连时到底会恢复哪些订阅、Will Message是否真被触发、Retain标志在服务端重启后是否还生效。本篇聚焦一个可本地闭环验证的完整方案：用开源、零依赖、纯 .NET 实现的 MQTT 服务端（HiveMQ CE 的 .NET 移植版） + 官方推荐的 MQTTnet 客户端库，从零搭建可调试、可断点、可修改 QoS/KeepAlive/Session 策略的最小可运行系统。适合需要快速验证设备接入逻辑、调试重连策略、或为嵌入式 C# 边缘网关做通信底座的工程师。
2. 用 MQTTnet + MQTTnet.Server 搭建本地闭环测试环境：不装 Docker、不配 Linux，纯 Windows/.NET 6+ 一行命令启动
MQTT 协议栈的落地难点从来不在“连”，而在“控”——控制连接生命周期、控制消息分发语义、控制服务端行为边界。所以本方案放弃 Mosquitto（C 写的，Windows 调试链路长）、放弃 EMQX（功能全但启动慢、日志冗余），选用MQTTnet.Server—— 这是一个由 MQTTnet 原作者维护的、与客户端库同源、API 风格一致、支持 .NET 6+ 的纯托管服务端实现。它不依赖任何外部进程，可直接作为库集成进控制台项目，方便加断点、改配置、看内存对象状态。
2.1 初始化服务端：三行代码启动一个可调试 Broker
新建一个 .NET 6 控制台项目（命名为MqttTestBroker），执行：
dotnet new console -n MqttTestBroker cd MqttTestBroker dotnet add package MQTTnet.Server --version 4.3.4.974提示：MQTTnet.Server 当前最新稳定版为
4.3.4.974（截至 2024 年中），不要用预览版（如5.0.0-preview-*），其 API 与正式版不兼容，且文档缺失。版本号必须显式指定，否则dotnet add package MQTTnet.Server默认拉取旧版3.x，不支持 TLS 1.3 和 WebSocket 端口复用。
编辑Program.cs，替换为以下最小启动逻辑：
using MQTTnet; using MQTTnet.Server; var options = new MqttServerOptionsBuilder() .WithDefaultEndpointPort(1883) // TCP 端口 .WithDefaultEndpointHost("localhost") // 绑定地址（非 0.0.0.0，便于本地调试） .WithConnectionBacklog(100) // 连接队列长度，防突发连接打满 .WithMaxPendingMessagesPerClient(1000) // 每客户端最大待发消息数，防内存溢出 .Build(); var server = new MqttFactory().CreateMqttServer(options); // 注册连接事件，用于观察客户端行为 server.ClientConnectedAsync += async e => { Console.WriteLine($"[Broker] Client '{e.ClientId}' connected from {e.Endpoint}"); }; server.ClientDisconnectedAsync += async e => { Console.WriteLine($"[Broker] Client '{e.ClientId}' disconnected. Reason: {e.Reason}"); }; await server.StartAsync(); Console.WriteLine("[Broker] Started on mqtt://localhost:1883"); Console.ReadLine(); // 阻塞，保持进程运行这段代码做了四件关键事：
- 显式绑定
localhost:1883，避免 Windows 防火墙弹窗干扰（0.0.0.0会触发权限请求）； - 设置
ConnectionBacklog=100，防止高并发连接时SocketException: 10035（资源暂时不可用）； - 限制
MaxPendingMessagesPerClient=1000，这是血泪经验：不设上限时，QoS1 客户端断线后，服务端会无限缓存未确认消息，最终 OOM； - 注册
ClientConnectedAsync/ClientDisconnectedAsync事件，这是调试重连逻辑的唯一可靠入口——比抓包更准，比日志更实时。
2.2 启动客户端：用 MQTTnet 发送带 QoS 和 Will Message 的结构化消息
再新建一个 .NET 6 控制台项目MqttTestClient：
dotnet new console -n MqttTestClient cd MqttTestClient dotnet add package MQTTnet --version 4.3.4.974注意：客户端必须用与服务端完全相同版本的 MQTTnet（
4.3.4.974），否则ConnectAsync()可能静默失败（底层序列化协议微调导致 CONNECT 报文解析异常）。
Program.cs内容如下（含重连逻辑、Will Message、结构化 Payload）：
using MQTTnet; using MQTTnet.Client; using MQTTnet.Client.Options; using MQTTnet.Client.Receiving; using MQTTnet.Protocol; var factory = new MqttFactory(); var client = factory.CreateMqttClient(); // 构建连接选项：关键参数全部显式设置 var options = new MqttClientOptionsBuilder() .WithClientId("test-client-001") // 必须固定，否则 CleanSession=false 无意义 .WithTcpServer("localhost", 1883) // 服务端地址 .WithCredentials("user", "pass") // 可选认证（本 demo 关闭鉴权，但保留字段） .WithCleanSession(false) // 关键！启用会话持久化 .WithKeepAlivePeriod(TimeSpan.FromSeconds(30)) // 心跳周期，必须 < 服务端 KeepAlive 上限 .WithWillMessage(new MqttApplicationMessageBuilder() .WithTopic("system/status") .WithPayload("offline") .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .WithRetain(true) .Build()) .Build(); // 连接事件监听（调试核心） client.ConnectedAsync += async e => { Console.WriteLine($"[Client] Connected. Session present: {e.SessionPresent}"); // 订阅主题，QoS1 确保消息不丢 await client.SubscribeAsync("sensor/+/temperature", MqttQualityOfServiceLevel.AtLeastOnce); }; client.DisconnectedAsync += async e => { Console.WriteLine($"[Client] Disconnected. Reason: {e.Exception?.Message ?? "Unknown"}"); // 简单重连策略：指数退避，最多 5 次 for (int i = 1; i <= 5; i++) { try { await Task.Delay(TimeSpan.FromSeconds(Math.Pow(2, i))); await client.ConnectAsync(options); Console.WriteLine($"[Client] Reconnect attempt {i} succeeded."); break; } catch (Exception ex) { Console.WriteLine($"[Client] Reconnect attempt {i} failed: {ex.Message}"); } } }; // 消息接收处理 client.ApplicationMessageReceivedAsync += e => { var topic = e.ApplicationMessage.Topic; var payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload); var qos = e.ApplicationMessage.QualityOfServiceLevel; Console.WriteLine($"[Client] Received: {topic} = {payload} (QoS{qos})"); }; // 执行连接 await client.ConnectAsync(options); // 模拟发送传感器数据（每 5 秒一条） var timer = new PeriodicTimer(TimeSpan.FromSeconds(5)); while (await timer.WaitForNextTickAsync()) { var msg = new MqttApplicationMessageBuilder() .WithTopic($"sensor/{Guid.NewGuid():N}/temperature") .WithPayload($"{{\"value\":{new Random().Next(20, 35)},\"ts\":{DateTimeOffset.UtcNow.ToUnixTimeSeconds()}}}") .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .WithRetain(false) .Build(); try { await client.PublishAsync(msg); Console.WriteLine($"[Client] Published to {msg.Topic}"); } catch (Exception ex) { Console.WriteLine($"[Client] Publish failed: {ex.Message}"); } }这段客户端代码的关键设计点：
WithClientId("test-client-001")固定 ID +WithCleanSession(false)是会话恢复的前提，缺一不可；WithWillMessage(...)设置遗嘱消息，当客户端异常断开（非主动 Disconnect）时，服务端自动发布system/status = offline，这是设备在线状态监控的基础；SubscribeAsync("sensor/+/temperature", QoS1)使用通配符订阅，验证服务端主题树匹配能力；PeriodicTimer替代Thread.Sleep，避免阻塞异步上下文；- 所有
await都包裹try/catch，因为 MQTTnet 在网络抖动时抛出MqttCommunicationException是常态，不捕获会导致进程崩溃。
3. 验证消息可靠性：用 QoS1 + Retain + Session 搭建“断电不丢数”的最小闭环
MQTT 的价值不在“快”，而在“稳”。一个合格的 demo 必须能回答三个问题：
- 客户端断网重连后，服务端是否还记得它的订阅？
- QoS1 消息在传输中途断连，重连后是否会补发？
Retain=true的消息，新订阅者是否能立即收到最新值？
本节用一组可手动触发的测试步骤，逐项验证。
3.1 测试会话恢复：断开客户端 → 修改服务端代码 → 重启服务端 → 观察重连日志
先确保服务端和客户端都已运行。此时客户端已订阅sensor/+/temperature，服务端日志应显示Client 'test-client-001' connected。
步骤 1：手动终止客户端进程（Ctrl+C 或关闭终端）。
→ 服务端日志立刻输出：Client 'test-client-001' disconnected. Reason: Connection closed by client（正常退出）。
步骤 2：修改服务端Program.cs，在server.StartAsync()前插入一行日志：
Console.WriteLine("[Broker] Starting with session store size: " + server.GetSessionStoreSize());然后重新编译启动服务端：
dotnet build && dotnet run→ 服务端启动后，你会看到：[Broker] Starting with session store size: 1
这证明：客户端上次连接的会话（含订阅列表、未确认的 QoS1 消息）已被持久化到内存中，未因服务端重启而丢失。
步骤 3：重新启动客户端
→ 客户端日志中ConnectedAsync事件的e.SessionPresent值为true，且服务端日志再次显示Client 'test-client-001' connected，但不会重复打印订阅日志（因为订阅已在会话中恢复）。
提示：
SessionPresent=true是 MQTT 协议层明确返回的标志，比查服务端日志更可信。若为false，说明ClientId不一致或CleanSession=true被误设。
3.2 测试 QoS1 消息重传：强制杀进程模拟网络中断，验证消息不丢
QoS1 的语义是“至少一次送达”，但实现依赖客户端和服务端双方的报文确认机制。测试必须破坏这个机制。
步骤 1：在客户端PublishAsync后添加断点或延时，让消息发出但不等待 PUBACK：
// 替换原 publish 代码段 var pubResult = await client.PublishAsync(msg); // 此处不 await，模拟“发了但没等回执” Console.WriteLine($"[Client] Sent PUBLISH, waiting for PUBACK..."); await Task.Delay(1000); // 等 1 秒，大概率 PUBACK 还没回来步骤 2：在await client.PublishAsync(msg)执行后、Task.Delay(1000)期间，手动 kill 客户端进程（Ctrl+C）。
→ 此时客户端已发出 PUBLISH，但未收到 PUBACK，按协议必须重发。
步骤 3：重启客户端
→ 客户端重连后，服务端会将这条未确认的 PUBLISH 重新推送给它（表现为ApplicationMessageReceivedAsync触发两次同一消息）。
→ 若你用Console.WriteLine打印消息体，会看到两条完全相同的 JSON（"value":28,"ts":171xxxxx）。
注意：这不是 Bug，是 QoS1 的正确行为。业务层需用
MessageId或业务主键去重。MQTTnet 本身不提供去重 API，这是应用层责任。
3.3 测试 Retain 消息：发布 Retain 消息 → 新客户端订阅 → 立即收到历史值
Retain 是 MQTT 的“状态快照”机制，对设备状态同步至关重要。
步骤 1：用当前客户端发送一条 Retain 消息：
// 在客户端循环 publish 外，单独加一段 await client.PublishAsync(new MqttApplicationMessageBuilder() .WithTopic("device/status") .WithPayload("online") .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce) .WithRetain(true) .Build());步骤 2：启动第三个控制台，运行一个全新客户端（ClientId不同，如new-client-002），只做一件事：订阅device/status并打印。
→ 它一连接成功，立刻收到device/status = online，无需等待下一次发布。
→ 此时即使你停止第一个客户端，device/status的 Retain 值仍存在，新订阅者永远能获取最新状态。
血泪经验：Retain 消息不占用会话存储空间，它独立存储在服务端主题索引中。所以
MaxPendingMessagesPerClient限制不影响 Retain 消息数量，但Retain=true的消息过多会撑大服务端内存，生产环境需定期清理（MQTTnet.Server 无自动清理 API，需自行定时server.UnsubscribeAsync+server.PublishAsync清空）。
4. 避坑指南：MQTT C# 开发中 5 个高频翻车点与现场排查法
MQTT 协议看似简单，但 .NET 生态下的实现细节极易踩坑。以下是我在多个边缘网关项目中记录的真实问题，按发生频率排序，每条附带现象 → 原因 → 解决三要素，拒绝玄学。
4.1 现象：客户端ConnectAsync()一直挂起，无异常、无日志，CPU 占用 0%
原因：服务端WithDefaultEndpointHost("localhost")与客户端WithTcpServer("127.0.0.1", 1883)地址格式不一致，触发 .NET DNS 解析阻塞（localhost解析为 IPv6::1，而服务端只监听 IPv4127.0.0.1）。
解决：客户端和服务端统一用127.0.0.1，或服务端改为WithDefaultEndpointHost("0.0.0.0")（仅开发环境）。
4.2 现象：客户端频繁断连，日志显示Reason: Connection closed by server，但服务端无对应日志
原因：客户端KeepAlivePeriod（如 30 秒）大于服务端默认KeepAlive上限（MQTTnet.Server 默认为 10 分钟，但某些版本实际硬编码为 60 秒）。当客户端心跳超时，服务端强制断连。
解决：服务端显式设置WithDefaultCommunicationTimeout(TimeSpan.FromMinutes(2))，并确保客户端KeepAlivePeriod小于该值的 2/3。
4.3 现象：QoS1 消息在客户端重连后重复消费，业务逻辑出错
原因：开发者误以为 MQTTnet 会自动去重，未在业务层实现幂等。ApplicationMessageReceivedAsync事件对每条到达消息都触发，无论是否重传。
解决：在消息 Payload 中嵌入唯一message_id字段，用ConcurrentDictionary<string, bool>缓存已处理 ID（内存去重），或写入本地 SQLite 表（持久化去重）。
4.4 现象：Will Message从未触发，客户端异常退出（如 Ctrl+C）后，订阅者收不到offline
原因：客户端调用了await client.DisconnectAsync()主动断开，此时 Will Message 不生效（协议规定：Will 只在 TCP 连接异常中断时触发）。
解决：测试 Will 必须用kill -9或任务管理器结束进程，不能调用DisconnectAsync。生产环境需用AppDomain.CurrentDomain.ProcessExit事件注册强制退出逻辑。
4.5 现象：服务端启动时报System.Net.Sockets.SocketException: 以一种访问权限不允许的方式做了一个访问套接字的尝试
原因：Windows 系统中，1024 以下端口（如 1883）需要管理员权限。Visual Studio 或终端未以管理员身份运行。
解决：右键 VS / PowerShell / CMD → “以管理员身份运行”，或改用 1883 以外的端口（如 18833），并在客户端连接字符串中同步修改。
提示：所有这些坑，90% 都能在服务端
ClientConnectedAsync/ClientDisconnectedAsync和客户端ConnectedAsync/DisconnectedAsync事件中第一时间捕获。把它们当成你的“黑匣子飞行记录仪”，而不是等出问题再翻日志。
5. 进阶技巧：用内存快照 + 自定义 Topic Handler 实现“所见即所得”的消息流调试
当 demo 跑通后，下一步是让它成为你的调试利器。与其在控制台刷屏日志，不如把消息流变成可查询、可过滤、可回溯的“数据湖”。
5.1 用ConcurrentQueue缓存最近 1000 条消息，提供实时查询 API
在服务端项目中，添加一个静态消息仓库：
public static class MqttMessageLog { private static readonly ConcurrentQueue<(string Topic, byte[] Payload, DateTime Timestamp)> _queue = new(1000); public static void Log(string topic, byte[] payload) => _queue.Enqueue((topic, payload, DateTime.UtcNow)); public static IEnumerable<(string Topic, string Payload, DateTime Timestamp)> GetRecent(int count = 100) { var list = new List<(string, string, DateTime)>(); foreach (var item in _queue) { list.Add((item.Topic, Encoding.UTF8.GetString(item.Payload), item.Timestamp)); if (list.Count >= count) break; } return list.AsEnumerable(); } }然后在服务端ApplicationMessageReceivedAsync事件中调用MqttMessageLog.Log(...)。
再添加一个简单的 HTTP 端点（用Microsoft.AspNetCore.Http）暴露查询接口：
// 在服务端 Program.cs 中，StartAsync 后添加 var http = WebApplication.CreateBuilder().Build(); http.MapGet("/messages", () => Results.Json(MqttMessageLog.GetRecent(50))); http.RunAsync("http://localhost:5000"); // 启动轻量 HTTP 服务现在访问http://localhost:5000/messages，就能看到 JSON 格式的消息流，前端用fetch直接渲染，比看控制台日志高效十倍。
5.2 用正则 Topic Handler 实现“动态路由”，绕过硬编码订阅
MQTTnet.Server 支持自定义 Topic 匹配器。我们可以用正则替代+/#通配符，实现更灵活的路由：
// 在服务端 options 中添加 .WithTopicMatcher(new RegexTopicMatcher()) // 自定义匹配器 // 实现类 public class RegexTopicMatcher : IMqttTopicMatcher { public bool Match(string topic, string pattern) { // 将 MQTT 通配符转为正则：sensor/+/temperature → sensor/\w+/temperature var regexPattern = Regex.Escape(pattern) .Replace("\\+", ".+") .Replace("\\#", ".*"); return Regex.IsMatch(topic, $"^{regexPattern}$"); } }这样，你就可以用sensor/[a-z]+/temperature这样的正则订阅，精准匹配设备型号，避免sensor/+/temperature订阅到sensor/debug/log这类干扰消息。
5.3 用MemoryCache实现设备状态聚合，替代轮询
在工业场景中，“某车间温度是否超限”这类需求，不应让每个客户端都订阅所有传感器，而应由服务端聚合后发布状态摘要：
// 在服务端中维护一个缓存 private static readonly MemoryCache _deviceStatus = new MemoryCache(new MemoryCacheOptions { SizeLimit = 1000 }); // 当收到 sensor/*/temperature 消息时 if (e.ApplicationMessage.Topic.StartsWith("sensor/") && e.ApplicationMessage.Topic.EndsWith("/temperature")) { var deviceId = e.ApplicationMessage.Topic.Split('/')[1]; var value = JsonSerializer.Deserialize<JsonElement>(e.ApplicationMessage.Payload).GetProperty("value").GetInt32(); _deviceStatus.Set($"temp:{deviceId}", value, TimeSpan.FromMinutes(5)); // 如果超限，主动推送告警 if (value > 32) { await server.PublishAsync(new MqttApplicationMessageBuilder() .WithTopic($"alert/{deviceId}") .WithPayload($"HIGH_TEMP:{value}") .Build()); } }这样，前端只需订阅alert/+，就能实时获知所有异常，而不用维持数百个传感器主题的连接。
我坚持在每个 MQTT 项目初期就集成这套调试骨架——不是为了炫技，而是因为在产线里，80% 的通信问题不是协议没懂，而是你根本不知道消息到底发没发、到没到、谁收到了、谁漏了。把服务端变成你的“消息黑匣子”，把客户端变成你的“行为记录仪”，才能把玄学调试变成确定性工程。希望帮到你。
本文还有配套的精品资源，点击获取
