简介这份资源是基于MQTTnet库在Winform中实现MQTT服务端与客户端通信的完整示例工程适合正在学习C#桌面应用开发或物联网消息通信的中级开发者。包内同时包含服务端与客户端两个独立项目代码结构清晰可直接运行并在此基础上扩展。资源共67个文件以cs源码、csproj工程配置、dll依赖库、json配置文件及exe可执行程序为主另有少量资源文件与说明文档压缩包整体约494KB体积小巧便于快速下载部署。目前已有2132人学习浏览适用于理解MQTT发布订阅模型、掌握Winform界面与通信逻辑结合的实战场景。示例中不仅演示了客户端与服务端的双向通信过程还实现了将订阅到的消息自动保存到本地文件方便后续分析消息内容是一份兼顾通信机制与数据持久化需求的实用参考代码。1. 用 MQTTnet 在 Winform 里自建服务端和客户端这套示例到底解决什么问题用 MQTTnet 在 Winform 里同时拉起 MQTT 服务端和客户端让两个角色跑在同一个进程内服务端收到的消息不仅在界面上滚动还按主题追加写进本地文件这是典型的本地调试工具形态。很多单片机设备调试程序、上位机数据采集软件、教学 Demo 都是这样组织的。它解决的不是 MQTT 协议原理而是怎么把 server、client、落盘三件事串成一个能跑的桌面程序。适合谁正在做设备调试工具、想快速搭一个本地 broker 验证协议、以及下了示例包但不知道怎么改端口、改主题、改保存路径的 .NET 开发者。这类场景的核心矛盾在于部署独立 broker 太重但完全自己拿 Socket 写 MQTT 协议又太傻。MQTTnet 是这个场景下最直接的答案它同时提供服务端和客户端两套对象一个 Winform 进程就能完成收、发、存。下面按落地顺序拆先讲为什么这么搭再给服务端、客户端的最小代码然后是文件落盘和典型踩坑。2. 先想清楚架构为什么把服务端和客户端放在同一个 Winform 进程里2.1 进程内双角色调试工具最常用的拓扑我见过不少从示例包入手的人第一反应是把服务端做成一个独立控制台程序Winform 只当客户端。这么做不是不行但会带来一个实际麻烦你调试时得同时盯两个窗口服务端起了没、端口被没被占、消息到底有没有进 broker都得来回切。把 MqttServer 和 MqttClient 放进同一个 Winform 进程理论上没有障碍——MQTTnet 的服务端和客户端是同一个库里的两个对象各自持有独立的连接状态互不干扰。这种拓扑的真正价值在于你的程序既可以扮演 broker接收外部设备或外部客户端上报的数据也可以扮演一个普通客户端去订阅别的 broker 上的主题。两种角色共享同一套 UI、同一个配置文件、同一个日志文件。做设备调试工具时尤其方便左边面板显示当前有多少客户端连着右边面板把自己订阅的消息实时列出来同时落到文件里。最常见的用法是Winform 内置一个 MqttServer监听本机端口外部设备或手机上的调试 App 往这个 broker 发消息Winform 自己再创建一个 MqttClient 订阅这些主题收到后更新界面并写文件。整个过程不需要安装任何额外服务。如果只是两个进程都在本机跑消息路径是 TCP 回环延迟可以忽略不计。生产环境当然不推荐这么干但作为本地调试和采集工具这个拓扑是性价比最高的。2.2 MQTTnet 的能力边界哪些能做哪些必须交给独立 brokerMQTTnet 不是 Mosquitto 的平替它的服务端定位是“可嵌入的轻量 broker”不是“集群级消息中间件”。能做的事不少支持 QoS 0/1/2、遗嘱消息、保留消息、通配符订阅、会话维持还能通过事件回调拿到客户端上下线、消息收发等全量生命周期。对一个调试工具来说这些已经完全够用。做不了的事情也得很清楚没有内置的账号体系和细粒度 ACL 管理没有消息持久化到数据库的重放机制没有多节点集群和负载均衡。这些都意味着它不适合作为公网长期服务的消息层。我见过有人拿它做内部小工具的消息总线跑了几个月没问题但一旦要求“断网期间消息不能丢”“按用户分配权限”就得往独立 broker 迁移。选型时先问自己一句这个 broker 是要跟着我的桌面程序走还是要跟着业务走。前者用 MQTTnet 很舒服后者应该直接上独立部署。另外要注意协议版本。MQTTnet 同时实现了 MQTT 3.1.1 和 5.0创建服务端和客户端时可以指定协议版本。自己程序内部通信用默认值就行但接外部设备时最好确认设备支持哪个版本。老设备多数只支持 3.1.1如果你的 broker 默认跑 5.0 又没有做协议协商可能会出现连接被拒的怪问题。2.3 引入包与最小工程结构一个项目里放 server 和 client 两个对象NuGet 引入 MQTTnet 包命令很简单注意这是整个项目唯一的第三方依赖不必装额外的东西。我用的是最新稳定版示例包里通常也是直接引这个包。dotnet add package MQTTnet如果你用的是 Visual Studio 的 NuGet 包管理器直接在 UI 里搜索 MQTTnet 安装也一样。装完后项目里会多出 MqttFactory、MqttServer、MqttClient 这几个核心类型都在 MQTTnet 命名空间下。工程结构上示例包最常见的就是一个 Form 搞定全部Form 持有一个 MqttServer 字段、一个 MqttClient 字段、一个日志文件路径字段。启动时先起服务端再连客户端并订阅关闭时逆序释放。这个结构不是最优的框架设计但足够直观适合照着改。我一般会在 Form 里额外放一个 MqttFactory 字段因为创建 server 和 client 都通过它避免到处 new。还有一点MQTTnet 的方法大量是异步的Winform 事件处理器里调用时记得处理异步上下文否则死锁和跨线程问题会一起找上门这个在后面的避坑章节里单独说。3. 用 MQTTnet 跑通服务端和客户端通信的最小闭环3.1 服务端最小代码本地 broker 怎么在 Winform 里启动先看服务端这一段它是整个通信闭环的地基。下面的代码按 MQTTnet 3.x 常见写法给出你的版本事件回调名如果带 Async 后缀属于 4.x 分支在第 3.3 节我会说明差异。// 创建 MqttFactory后续创建 server 和 client 都复用它 var factory new MqttFactory(); // 构建服务端选项 var serverOptions new MqttServerOptionsBuilder() .WithDefaultEndpoint() // 开启默认 TCP 端点 .WithDefaultEndpointPort(1883) // 监听 1883 端口 .Build(); // 创建并启动服务端 var mqttServer factory.CreateMqttServer(serverOptions); await mqttServer.StartAsync(); // 服务端收到任意客户端发来的消息时触发 mqttServer.ApplicationMessageReceivedHandler e { string topic e.ApplicationMessage.Topic; string payload e.ApplicationMessage.ConvertPayloadToString(); Console.WriteLine($收到主题 {topic}内容 {payload}); return Task.CompletedTask; };这一段做了三件事构建选项、启动服务端、挂消息入口。WithDefaultEndpoint()表示使用默认 TCP 端点不调用的话服务端不会监听任何端口这是初学者最容易漏的一步。WithDefaultEndpointPort(1883)指定监听端口1883 是 MQTT 标准端口但如果本机已经跑着别的 broker就得改掉。ApplicationMessageReceivedHandler是服务端视角的“所有消息汇聚点”不管是哪个客户端发上来的都会先经过这里。注意这个回调运行在线程池线程做耗时操作要谨慎UI 更新必须封送。ConvertPayloadToString()是 MQTTnet 提供的便捷方法内部按 UTF-8 解码比手动取 Bytes 再转字符串少写两行。启动服务端之前最好先把监听地址也确认一下。默认绑定的是所有网卡如果想只允许本机连接可以加.WithEndpointBoundIPAddress(IPAddress.Loopback)双网卡机器上这个参数能避免不少安全问题。提示服务端启动后要保留mqttServer字段引用不能丢局部变量否则垃圾回收会把监听对象带走客户端连不上时你还在怀疑端口被占其实是引用没了。3.2 客户端连接与订阅从 server 手里拿到消息服务端起来了接着创建客户端并让它订阅一个主题。客户端是消息闭环的另一半代码套路和服务端类似构建选项、连接、挂事件。// 构建客户端连接选项 var clientOptions new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) // 连刚才启动的服务端 .WithClientId(winform-client-001) // 客户端 ID必须唯一 .WithCleanSession() // 使用干净会话 .Build(); var mqttClient factory.CreateMqttClient(); // 收到订阅主题的消息时触发 mqttClient.ApplicationMessageReceivedHandler e { string topic e.ApplicationMessage.Topic; string payload e.ApplicationMessage.ConvertPayloadToString(); Console.WriteLine($客户端收到 {topic}: {payload}); return Task.CompletedTask; }; // 连接并订阅一个主题 await mqttClient.ConnectAsync(clientOptions); await mqttClient.SubscribeAsync(new MqttTopicFilterBuilder() .WithTopic(device//data) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .Build());ClientId是 MQTT 协议里唯一标识客户端的字段同一个 broker 下重复会互相踢下线。如果两个连接用了同一个 ID后连的会把先连的踢掉这在多客户端调试时经常莫名其妙发生。WithCleanSession()表示不保留会话状态重连后需要重新订阅调试工具的客户端保持这个默认值就好。SubscribeAsync传入的是一个主题过滤器device//data里的是单层通配符能匹配device/a123/data、device/b456/data这样的主题但匹配不了device/a/b/data。如果需要匹配多级主题用#比如device/#。QoS 选了AtLeastOnce表示消息至少送达一次可能重复适合采集端如果对重复敏感用ExactlyOnce代价是吞吐量下降。3.3 双角色互通的完整消息流与事件分工很多第一次照着示例包改的人会困惑服务端有 ApplicationMessageReceivedHandler客户端也有 ApplicationMessageReceivedHandler消息到底在哪个里处理答案是两个都会触发但看到的视角不同。客户端 A 发布一条消息到 broker 后broker 先在自己的ApplicationMessageReceivedHandler里看到这条消息然后根据主题匹配把它转发给所有订阅了这个主题的客户端这些客户端的ApplicationMessageReceivedHandler再各自触发。也就是说服务端的回调是全局视角客户端的回调是订阅者视角。// 客户端 A 发布一条消息 await mqttClient.PublishAsync(new MqttApplicationMessageBuilder() .WithTopic(device/001/data) .WithPayload({\temp\:36.5}) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .Build());上面的发布代码就是完整的一条消息流入口。实际调试时可以用 Winform 里的一个按钮触发发布看另一处订阅是否收到以此验证整个链路。如果收不到先查服务端回调有没有触发有说明发布和路由正常问题在订阅过滤器没有说明消息根本没进 broker问题在连接或主题参数。关于版本差异MQTTnet 4.x 把 3.x 的属性赋值式事件改成了方法注册式比如服务端的UseApplicationMessageReceivedHandler、客户端的ConnectedAsync。逻辑没有大变化但照着 3.x 示例抄会编译报错。先看项目里 NuGet 提示实际装的哪个大版本再对应改事件名不要盲抄网上一段就粘贴。4. 把订阅消息保存到文件落盘格式、线程安全与性能取舍4.1 落地格式与编码时间戳、主题、Payload 怎么排保存订阅消息第一件事是定行格式。我建议把时间戳、主题、消息内容按固定分隔符排成一行这是最容易被后续脚本读取的结构。时间戳用 ISO 或yyyy-MM-dd HH:mm:ss.fff都行主要看要不要跨天分析和排序。分隔符建议避开消息内容里可能出现的字符竖线|或者制表符\t都可以JSON 消息里常见逗号和冒号但很少出现竖线。编码是新手翻车重灾区。默认的File.AppendAllText如果不指定编码在中文系统下会走系统 ANSI 编码短消息看着正常一旦消息里混入 emoji 或特殊符号再读出来就是乱码。写文件一律显式指定Encoding.UTF8读取时也用 UTF-8两边对齐才不会有玄学问题。如果落盘后还要被其它工具读建议文件开头不要带 BOM否则部分解析器第一行会多出三个字符。我用的落盘格式如下表字段顺序固定后续用 awk 或 Excel 分列都很方便。字段示例说明时间戳2024-06-01 14:30:22.123接收到的本地时间精确到毫秒主题device/001/data消息原始主题含通配符展开后的实际值QoS1消息实际到达的 QoS 等级Payload{temp:36.5}原始消息内容按 UTF-8 解码4.2 线程安全的追加写一个 lock 和一个 StreamWriter消息接收事件是并发触发的多个主题同时到达时如果直接在回调里各自调用File.AppendAllText会偶发文件被占用异常。这是典型的并发写文件问题解决办法是给写入动作加锁让同一时刻只有一个线程在写文件。private readonly object fileLock new object(); private readonly string logFilePath Path.Combine(AppDomain.CurrentDomain.BaseDirectory, mqtt_messages.log); // 线程安全的追加写文件 private void SaveMessage(string topic, string payload, int qos) { string line ${DateTime.Now:yyyy-MM-dd HH:mm:ss.fff}|{topic}|{qos}|{payload}{Environment.NewLine}; lock (fileLock) { File.AppendAllText(logFilePath, line, Encoding.UTF8); } }lock是 C# 最简单实用的互斥手段这里锁住整个写入过程保证多线程回调下每次只写一行不会出现两行内容交错。File.AppendAllText内部是打开文件、写入、关闭对低频调试场景足够不用自己管 StreamWriter 的释放。有些示例包会在 Form 里维护一个 StreamWriter并且设置AutoFlush true。这个做法本身没问题但要注意StreamWriter 默认是缓冲的不及时 Flush 的话程序崩溃或断电时会丢最后一段数据。如果消息量不大AppendAllText这种“写一次关一次”的模式反而更稳代价是每秒写几百条时会有点慢。对于一次调试会话几千条消息的规模完全不用优化。4.3 数据量上来之后从同步写改成队列消费如果消息频率到了每秒几十条甚至上百条每次都AppendAllText打开关闭文件磁盘开销会拖累整个 Winform 界面。这时候要改写入模型接收回调只把消息丢进队列由后台线程统一消费写盘。// 用并发队列做消息缓冲 private readonly ConcurrentQueuestring msgQueue new ConcurrentQueuestring(); private readonly System.Windows.Forms.Timer flushTimer; // 接收回调里只入队 private void EnqueueMessage(string topic, string payload, int qos) { msgQueue.Enqueue(${DateTime.Now:yyyy-MM-dd HH:mm:ss.fff}|{topic}|{qos}|{payload}{Environment.NewLine}); } // 定时器触发批量写盘例如每 500ms 一次 private void FlushMessages(object sender, EventArgs e) { var lines new Liststring(); while (msgQueue.TryDequeue(out string line)) { lines.Add(line); } if (lines.Count 0) { lock (fileLock) { File.AppendAllLines(logFilePath, lines, Encoding.UTF8); } } }这个方案的要点是ConcurrentQueue保证入队出队线程安全接收回调里绝不碰磁盘只做内存操作所以响应很快。定时器FlushMessages批量取走队列里的消息一次性写成多行磁盘读写的次数从“每条一次”降到“每 500ms 一次”性能提升非常明显。队列积压是一个需要留意的信号如果每次 Flush 后队列还是持续增长说明写入速度跟不上产消息速度。这时先看是不是文件在一个目录下过大可以考虑按天拆文件再考虑批量大小调优。调试工具一般到不了这个量级但知道这个演进路径等真正需要时不会手忙脚乱。5. MQTTnet 落地避坑连不上、收不到、乱码和端口占用的 5 个典型问题5.1 服务端和客户端端口冲突客户端连上了别的 broker现象服务端明明启动了客户端连接却报错或者连上了但发消息没反应。排查时发现本机 1883 端口被一个旧服务占着客户端实际连的是那个服务。原因WithDefaultEndpointPort(1883)只表示“我希望监听 1883”如果端口被占用服务端启动时可能已经悄悄失败或者另一个进程先占用了端口。客户端WithTcpServer(127.0.0.1, 1883)连的也是一个端口号不会区分是谁在监听。解决启动服务端后立刻检查端口用下面命令确认监听进程。netstat -ano | findstr 1883如果看到非本程序的 PID 占用要么停掉旧服务要么把服务端和客户端的端口一起改成比如 18830。改端口要记得客户端和服务端同步改只改一端是无效的。5.2 事件回调里直接操作 UI抛跨线程异常现象消息一来想在界面上显示在 TextBox 或 Label 里结果抛InvalidOperationException提示“线程间操作无效”。原因MQTTnet 的事件回调不在 Winform 的 UI 线程上执行而是在线程池线程。Winform 控件要求只能在创建它的线程上操作直接赋值必然触发跨线程保护。解决在回调里先判断是否需要封送再更新控件。mqttClient.ApplicationMessageReceivedHandler e { string payload e.ApplicationMessage.ConvertPayloadToString(); if (txtMessage.InvokeRequired) { txtMessage.BeginInvoke(new Action(() txtMessage.Text payload)); } else { txtMessage.Text payload; } return Task.CompletedTask; };BeginInvoke是异步封送不会阻塞消息接收线程。如果回调里还有其他逻辑建议只把 UI 更新部分包起来其余数据处理留在回调里做。5.3 订阅主题能收到但保存到文件的中文变成乱码现象界面上显示的中文正常打开日志文件看到æ¶æ¯之类的乱码。原因File.AppendAllText没指定编码时在简体中文系统下默认走 GBK 或系统 ANSI 编码UTF-8 的消息内容被转换后就成了乱码。更隐蔽的情况是消息发出端用的编码和接收端解码不一致。解决写入和读取都统一显式指定 UTF-8并且不在文件头写 BOM。File.AppendAllText(logFilePath, line, new UTF8Encoding(false));new UTF8Encoding(false)会生成不带 BOM 的 UTF-8对后续程序解析更友好。注意读取文件时也要用File.ReadAllText(logFilePath, Encoding.UTF8)不要依赖自动识别。5.4 程序退出后端口还在占用重启起不来现象调试时反复改代码重启程序某次启动直接报“地址已被占用”但任务管理器里明明没有自己的进程。原因Winform 窗口关闭时进程退出但如果服务端没有调用StopAsync和DisposeTCP 连接可能进入 TIME_WAIT 状态一时半会不释放端口。尤其你在调试器里强停进程资源释放不完整的情况更多。解决在 Form 关闭事件里显式停止服务端和客户端。protected override async void OnFormClosing(FormClosingEventArgs e) { if (mqttClient ! null) { await mqttClient.DisconnectAsync(); mqttClient.Dispose(); } if (mqttServer ! null) { await mqttServer.StopAsync(); mqttServer.Dispose(); } base.OnFormClosing(e); }如果强停进程后端口仍被占用执行netstat -ano | findstr 1883找到 PID再在任务管理器里结束该 PID。正常情况下只要正常关闭窗口显式释放及时不会反复出现这个问题。5.5 客户端断线后不重连消息静默丢失现象外部设备或 broker 重启后Winform 客户端不再收到任何消息界面不报错像是死了一样。原因MQTTnet 的客户端在连接断开后不会自动重连也不会自动重新订阅。断线期间的订阅关系已经失效broker 不会因为你断线了还继续给你推送。解决挂一个断线处理自动重连并重新订阅。mqttClient.DisconnectedHandler async e { for (int i 0; i 5; i) { await Task.Delay(3000); try { await mqttClient.ConnectAsync(clientOptions, CancellationToken.None); await mqttClient.SubscribeAsync(new MqttTopicFilterBuilder() .WithTopic(device//data) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .Build()); break; } catch { // 重连失败等待下一次重试 } } };重连之后重新订阅是关键很多人重连成功但没重新订阅结果还是收不到消息误以为重连逻辑有 bug。另外如果消息量重要程度高可以考虑把 QoS 调成ExactlyOnce并开启会话延续但这会牺牲一部分性能要在可靠性和吞吐之间做取舍。6. 进阶验证用外部客户端验证服务端、加上重连和遗嘱6.1 用外部 MQTT 调试客户端验证本地 broker程序跑起来后最直接的验证方式是用外部的 MQTT 调试工具连到127.0.0.1:1883发布一条消息到device/test/data观察你的 Winform 客户端是否收到并写入文件。这一步能快速确认你写的代码链路是通的而不是在某些环节静默失败。// 可选的遗嘱消息配置客户端异常断线时 broker 帮它广播 var clientOptions new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) .WithClientId(winform-client-001) .WithWillTopic(device/status) .WithWillPayload(offline) .WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .WithWillRetain() .Build();WithWillTopic和WithWillPayload是遗嘱消息的关键参数当客户端非正常断开网络闪断、进程崩溃时broker 会代替它发布这条遗嘱其它订阅者能立刻感知设备掉线。这是设备监控场景的标配能力示例包里未必有但值得自己加上。验证完成后我的习惯是先确认服务端事件触发再确认客户端订阅收到最后检查文件落盘。按这个顺序排查链路问题几分钟就能定位比拿到代码就乱改参数高效得多。对于这套方案我的建议是如果只是本地调试和采集MQTTnet 的进程内双角色完全够用值得投入端口和编码两个坑记住基本不会翻车。如果将来要接公网多人使用趁早规划独立 broker 的迁移路径但消息格式和文件落盘这套设计可以原样继承。希望帮到你。本文还有配套的精品资源点击获取