MQTTnet 同进程 Broker/Client 联调与落盘实践

简介：这是一份面向.NET桌面开发者的MQTT通信示例代码包，围绕WinForm与MQTTnet展开，重点演示服务端与客户端之间的连接、订阅/发布流程，以及将订阅到的消息保存到文件。包内包含服务端与客户端两个独立工程，8个cs源码文件配合2个csproj与1个sln，便于直接打开编译；10个dll提供MQTTnet等依赖，4个exe与4个pdb可用于运行和调试，json、cache及resx等文件则承载配置、缓存与窗体资源。压缩包共67个文件，约494KB，体量轻便，适合在本地快速搭建实验环境。已有2133人学习，说明该示例在MQTT入门与WinForm集成场景中具有一定参考价值。工程按服务端和客户端两条线组织，便于分别阅读与调试；读者可从中了解服务端启动、客户端连接、主题订阅、消息接收后写入文件的基本方式，并据此替换Broker地址与保存路径，改造为桌面端消息采集或设备通信小工具。

1. 一个能同时跑 Broker 和 Client 的 Winform 小工具，比装三个软件省事

调 mqtt 接口最烦的不是协议本身，是环境：服务端装一个、图形客户端装一个、再开个命令行窗口收订阅，来回切。这份 Winform 示例代码做的事很直接——用 MQTTnet 在同一个桌面进程里同时起服务端和客户端，界面上点一下把 Broker 拉起来，另一个面板连上去发消息、订阅通配符，收到的消息按 Topic 落到本地文件。它解决的是联调期「看得见」的问题：谁连上来了、谁订了什么、消息到底有没有到、到了之后长什么样、落盘那一行有没有丢字段。适合做设备上位机、产线工装、教学演示的开发者，也适合刚接触 mqtt 想找个能跑起来的骨架的人。下面按「选型 → 起服务端 → 连客户端 → 落盘 → 避坑」的顺序拆，每段都落到能抄的代码上。

2. 三条链路先定下来：Broker、Client、落盘各自用什么、为什么

在动手改这个示例之前，有一件事必须先定：你手上拿到的这份代码到底是 MQTTnet 哪一代 API。这不是洁癖，是能不能编译通过的问题。同一个文件名MqttServer，在 3.x 和 4.x 里的创建方式、事件挂载方式、连接校验的挂载位置全都不一样，网上搜到的代码片段八成是更早那一代的写法，直接贴进新工程会报「事件只能 += 或 -=」「找不到 WithConnectionValidator」之类的错。三条链路——Broker 怎么起、Client 怎么连、消息怎么落——的选择理由，下面一条条说清。

2.1 MQTTnet 的版本分水岭：先对齐 API 再动手

大版本之间的搬家主要发生在几个地方：工厂类被拆细、回调从同步 Handler 改成返回 Task 的 Async 事件、消息载荷从byte[]变成ArraySegment<byte>、连接校验从 options builder 挪到 server 的事件上。判断手里这份代码属于哪一代，最快的办法是搜三个关键词：ApplicationMessageReceivedHandler（老）、ApplicationMessageReceivedAsync（新）、ValidatingConnectionAsync（新）。

对着表改一遍，能省掉后面 80% 的编译错误。我的习惯是先不管业务，把一份最小可跑的 Broker + Client 跑通，确认版本对上了再去套界面逻辑，否则你会同时面对「API 不对」和「界面逻辑不对」两个变量，排错时根本分不清是谁的问题。

2.2 为什么把 Broker 塞进 Winform 进程

生产环境当然应该让 Broker 独立部署，但联调阶段单进程有它的价值：端口占用一眼能看见、启停和界面按钮绑在一起、抓日志不用去翻另一个服务。代价也很明确——进程崩溃等于 Broker 崩溃，窗体关闭时如果没走StopAsync，1883 端口会短暂处于 TIME_WAIT，下一次启动直接抛「地址已在使用」。所以起停顺序必须写死：窗体加载时先起服务端，客户端连接放在服务端StartAsync之后；窗体关闭时反过来，先断开客户端、再停服务端、最后释放。

另一个取舍是绑定地址。示例默认用IPAddress.Any，本机调试时我更倾向于绑127.0.0.1，一是避免被同网段其他机器误连，二是省掉防火墙弹窗。等要接真实设备了，再改成Any并在防火墙里放行 1883，这一步不要提前做。

2.3 落盘链路为什么从 JSON Lines 起步

订阅到的消息要保存，第一反应往往是 SQLite 或者 CSV。SQLite 的问题是要建表、要处理并发写、字段一改就得迁移；CSV 的问题是 Topic 或 payload 里一旦出现逗号、换行、引号，整个文件就废了，而 MQTT 的 payload 恰恰经常是带换行的 JSON。

JSON Lines 的好处是一行一个完整 JSON，追加写不需要读全文件，新加字段老行照样能解析，出问题直接tail -f就能看。等数据量真上来了再谈入库，联调阶段不值得为存储提前上复杂度。

3. 服务端：1883 端口、账号校验、三件事件挂载

服务端这块代码量不大，但每一行都影响后面客户端能不能连上、连上之后能不能看到状态。示例里最值得抄的是事件挂载部分，因为「谁连上来了、谁订了什么、谁发了什么」这三件事，全靠几个+=撑着。

3.1 组装 MqttServerOptions 与端口绑定

// 服务端初始化：绑定端口 + 绑定网卡地址 var factory = new MqttFactory(); var serverOptions = new MqttServerOptionsBuilder() .WithDefaultEndpoint() // 启用默认 TCP 端点 .WithDefaultEndpointPort(1883) // 标准 MQTT 端口，改这里要同步改客户端 .WithDefaultEndpointBoundIPAddress(IPAddress.Loopback) // 本机调试先绑回环，接设备时改 Any .Build(); var mqttServer = factory.CreateMqttServer(serverOptions); // 启动：异步，务必 await，否则紧接着的客户端连接会失败 await mqttServer.StartAsync();

WithDefaultEndpoint()不加的话后面两个With都无效，这是最常见的漏写。端口号选 1883 是习惯问题，实际项目里如果本机已经跑着别的 Broker，改成 1884、1885 都行，只要客户端同步改。WithDefaultEndpointBoundIPAddress接受IPAddress类型，写字符串会编译不过，这是从别处抄代码时的高频错误。

启动之后建议立刻在界面状态栏打一行「Broker listening on 1883」，不要只写「启动成功」，端口号写出来，后面排查端口占用时能省一次翻配置。

3.2 用 ValidatingConnectionAsync 做账号校验

// 连接校验：在握手阶段拒绝非法客户端 mqttServer.ValidatingConnectionAsync += e => { // 演示用的固定账号，实际项目从配置或数据库读 if (e.UserName != "demo" || e.Password != "demo123") { e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword; return Task.CompletedTask; } // 可以在这里按 ClientId 前缀做设备分组，比如只允许 "dev-" 开头的客户端接入 if (!e.ClientId.StartsWith("dev-") && !e.ClientId.StartsWith("winform-")) { e.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid; return Task.CompletedTask; } e.ReasonCode = MqttConnectReasonCode.Success; return Task.CompletedTask; };

ReasonCode不是可有可无的赋值，不写的话默认值可能不是 Success，客户端会收到一个语义含糊的拒绝。校验失败时客户端侧ConnectAsync会抛异常，异常信息里带的是返回码，所以服务端这边给的码越准确，客户端排查越快。密码不要硬编码在窗体代码里，示例为了能跑起来才这么写，实际至少放配置文件。

按 ClientId 前缀做一层过滤很实用：产线上一批设备刷了同一个固件，ClientId 全是默认值，会互相顶掉连接，服务端提前拦住比事后查日志强。

3.3 客户端上下线与会话事件：把连接状态打到界面上

// 上下线：更新界面上的在线列表 mqttServer.ClientConnectedAsync += e => { AppendLog($"[+] {e.ClientId} 已连接，用户名 {e.UserName}"); return Task.CompletedTask; }; mqttServer.ClientDisconnectedAsync += e => { AppendLog($"[-] {e.ClientId} 已断开，类型 {e.DisconnectType}"); return Task.CompletedTask; }; // 订阅关系：谁订了哪个 filter，在服务端这一侧就能看到 mqttServer.ClientSubscribedTopicAsync += e => { AppendLog($"[sub] {e.ClientId} -> {e.TopicFilter.Topic} (QoS {e.TopicFilter.QualityOfServiceLevel})"); return Task.CompletedTask; }; // 拦截发布：既能看消息，也能做权限或内容改写 mqttServer.InterceptingPublishAsync += e => { AppendLog($"[pub] {e.ClientId} -> {e.ApplicationMessage.Topic}"); return Task.CompletedTask; };

这四个事件里，ClientSubscribedTopicAsync最容易被忽略，但它恰恰是排查「客户端明明连上了却收不到消息」的第一现场——如果这里没打印，说明订阅请求根本没到 Broker，问题在客户端侧；如果打印了但客户端没回调，问题在分发或 Topic 匹配上。

InterceptingPublishAsync默认会放行，想拦截特定 Topic 可以把e.ProcessPublish置 false，消息就不会转发给订阅者，但仍然会走到这里，适合做灰度或者审计。注意这个事件在消息热路径上，里面不要做耗时操作，写日志用队列异步落，别直接File.AppendAllText。

3.4 服务端启停顺序与资源释放

// 窗体关闭：先断客户端、再停服务端，顺序反了会看到一堆断开异常 private async Task ShutdownAsync() { if (_client != null && _client.IsConnected) { await _client.DisconnectAsync(); // 优雅断开，让 Broker 走正常流程 } if (_mqttServer != null) { await _mqttServer.StopAsync(); // 释放 1883 端口 } _mqttServer?.Dispose(); }

DisconnectAsync不带参数时发的是正常断开报文，Broker 会立刻触发ClientDisconnectedAsync；如果是网络层直接断，要等 KeepAlive 超时才判定离线，界面上会「假在线」一段时间。这也是联调时常见的困惑来源：拔了网线，服务端列表里的人还在。

Dispose放在StopAsync之后，顺序反了偶尔会抛对象已释放。这类顺序问题不体现为编译错误，只体现为偶发异常，写完就把这段封成一个方法，别在FormClosing里现写。

4. 客户端：连接参数、通配符订阅与跨线程刷新

客户端部分是最容易写出「看着能跑但数据不对」的地方，因为连接参数、订阅时机、线程模型三件事互相耦合。参数错了表现为连不上，订阅时机错了表现为收不到，线程模型错了表现为界面卡死或者直接抛异常。

4.1 ConnectAsync 参数怎么定：ClientId、CleanSession、KeepAlive

var factory = new MqttFactory(); var client = factory.CreateMqttClient(); var options = new MqttClientOptionsBuilder() .WithTcpServer("127.0.0.1", 1883) .WithClientId("winform-" + Guid.NewGuid().ToString("N").Substring(0, 8)) // 唯一，重复会被顶下线 .WithCredentials("demo", "demo123") .WithCleanSession(true) // true：每次连接都是干净会话 .WithKeepAlivePeriod(TimeSpan.FromSeconds(30)) // 心跳周期，服务端 1.5 倍超时判离线 .WithTimeout(TimeSpan.FromSeconds(10)) // 连接超时，内网 5~10 秒足够 .Build(); await client.ConnectAsync(options);

ClientId必须唯一，这一点在联调时坑过很多人：两个客户端用了同一个默认 Id，后连的把先连的顶掉，先连的那个还在界面上显示「已连接」，因为本地状态没刷新。加个 GUID 前缀是最省事的解法，但要注意有些 Broker 对 Id 长度有限制，别拼太长。

CleanSession取 true 时，断线期间服务端不保留订阅关系和离线消息，重连之后要重新订阅；取 false 则需要 Broker 支持会话保持，并且 ClientId 固定不变，否则会话找不到。示例默认 true，因为它更符合「打开就能看到当前消息」的直觉，做断线补发时才改成 false。

KeepAlivePeriod和服务端的超时判定是联动的，客户端 30 秒心跳，服务端一般在 1.5 倍即 45 秒后判定离线。设太短会增加空包，设太长会让离线发现变慢，30 秒是桌面端比较稳的取值。

4.2 订阅通配符：+ 与 # 的边界

// 订阅：+ 匹配单层，# 匹配多层（必须放最后） var subOptions = new MqttClientSubscribeOptionsBuilder() .WithTopicFilter(f => f .WithTopic("demo/+/state") // 只关心 state 这一层 .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)) .WithTopicFilter(f => f .WithTopic("demo/#") // 兜底，收全部子话题 .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce)) .Build(); await client.SubscribeAsync(subOptions);

两个 filter 有重叠时，同一条消息可能被投递两次，客户端会收到两遍。示例里同时订了demo/+/state和demo/#，如果 Topic 正好是demo/a/state，回调就会触发两次，落盘文件里出现重复行——这不是 QoS 重发，是订阅重叠，初学者经常把它误判成「Broker 重复投递」。

另外#不会匹配以$开头的话题，想看 Broker 自身的运行指标得显式订阅$SYS/#，这一点在规范里写死了，不是库的实现问题。

4.3 发布与 UI 线程：Invoke 的两种写法

// 收消息回调：跑在 MQTTnet 的 IO 线程上，不能直接碰控件 client.ApplicationMessageReceivedAsync += e => { var topic = e.ApplicationMessage.Topic; var payload = e.ApplicationMessage.ConvertPayloadToString(); // 老版本用 Payload + Encoding.UTF8.GetString // 方案一：直接切 UI 线程（消息量小时够用） if (txtRecv.InvokeRequired) txtRecv.BeginInvoke(new Action(() => txtRecv.AppendText($"{topic}: {payload}\r\n"))); else txtRecv.AppendText($"{topic}: {payload}\r\n"); // 方案二：入队，交给 UI 定时器批量刷（推荐） _uiQueue.Enqueue(new MessageRecord(topic, payload, DateTime.Now, e.ApplicationMessage.QualityOfServiceLevel)); return Task.CompletedTask; };

BeginInvoke和Invoke的区别在这里很实际：Invoke是同步等待 UI 线程处理完，消息一密集就会把 IO 线程堵住，进而影响后续消息的接收；BeginInvoke是投递后立即返回，不阻塞。但BeginInvoke投太快也有问题，UI 消息队列会堆积，表现是界面越来越卡、最后内存飙高。

我一般会改成方案二：回调里只做入队，一个 100 毫秒的Timer从队列里批量取，一次AppendText拼接多条。这样 UI 刷新频率可控，IO 线程也不会被拖住。队列要有上限，超了就丢最旧的并在界面标红提示，比无声堆积好。

4.4 QoS 0/1/2 在这个场景下的实际差别

联调时用 1 最划算：消息基本不丢，重复的概率在内网很低。但要知道它「可能重复」是协议允许的，不是 bug。真要对账，落盘记录里带上消息标识自己做去重，别指望 QoS 2 解决所有问题——QoS 2 的握手开销在设备量上来之后很明显。

5. 消息落盘避坑：写文件、格式、丢消息与重复的五类现场

落盘看起来是最没技术含量的一段，实际上翻车最集中。原因很简单：它在消息热路径上，又涉及磁盘、线程、文件句柄三个不受你控制的东西。下面这套写法是踩过之后收敛出来的，先给模型，再列五类现场。

5.1 落盘模型的两种写法：同步 AppendAllText 与单消费者队列

// 消息记录：落盘的最小字段集，多一个字段就多一份排查线索 public record MessageRecord( string Topic, string Payload, DateTime ReceivedAt, MqttQualityOfServiceLevel Qos); // 有界队列：满了就丢，避免内存无限增长 private readonly BlockingCollection<MessageRecord> _diskQueue = new(new ConcurrentQueue<MessageRecord>(), 20000); // 单消费者写盘：一个后台长任务，串行写，天然避免多线程抢文件 private void StartWriter(string dir) { Task.Factory.StartNew(() => { foreach (var rec in _diskQueue.GetConsumingEnumerable()) { try { var safeTopic = SanitizeTopic(rec.Topic); // 见 5.2 var path = Path.Combine(dir, $"{safeTopic}-{rec.ReceivedAt:yyyyMMdd}.jsonl"); var line = JsonSerializer.Serialize(rec); File.AppendAllText(path, line + Environment.NewLine, Encoding.UTF8); } catch (IOException ex) { // 写失败不能静默，计数并在界面提示，否则会以为消息没到 Interlocked.Increment(ref _writeFailCount); AppendLog($"写盘失败：{ex.Message}"); } } }, TaskCreationOptions.LongRunning); } // 入队：TryAdd 不阻塞，失败就说明消费跟不上 private void EnqueueForDisk(MessageRecord rec) { if (!_diskQueue.TryAdd(rec)) Interlocked.Increment(ref _dropCount); }

这段的关键点在三个地方。BlockingCollection有界，是为了在磁盘慢于消息速率时主动丢弃而不是把内存吃满；单消费者，是为了让文件写入天然串行，不用加锁；TryAdd而不是Add，是为了让 IO 线程永远不阻塞——Add在队列满时会挂起调用线程，MQTTnet 的回调线程被挂起，后续消息就全卡住了。

File.AppendAllText每次调用都会打开、写入、关闭文件，消息量大时这是明显的瓶颈。改善办法是缓存StreamWriter，按日期或大小切换文件句柄，但复杂度会上一个台阶。联调阶段一天几千条消息，AppendAllText完全够用，别过早优化。

5.2 文件命名、滚动与编码

// Topic 转文件名：Windows 下 \ / : * ? " < > | 都是非法字符 private static string SanitizeTopic(string topic) { var replaced = topic.Replace('/', '_'); foreach (var c in Path.GetInvalidFileNameChars()) replaced = replaced.Replace(c, '-'); // 控制长度，避免超过 260 字符路径限制 return replaced.Length > 80 ? replaced.Substring(0, 80) : replaced; }

Topic 里带/是常态，直接拼进路径会变成子目录，Path.Combine不会报错，但文件会散落到各个层级，找的时候很痛苦。更糟的是 Topic 里可能出现:或者?，在某些 Windows 版本上直接抛异常，落盘线程整个挂掉，界面上看着消息还在刷，文件里一条没有。

命名规则我一般定成「Topic 转义 + 日期 + .jsonl」，一天一个文件，Topic 多的场景再按 Topic 分目录。滚动不按大小按日期，是因为排查时人找的是「昨天下午那批」，日期比大小直观。编码统一 UTF-8 无 BOM，不然用某些编辑器打开会看到第一个字段前面多几个不可见字符。

5.3 五条踩坑记录

现象一：客户端提示连接成功，服务端在线列表也有，但订阅的消息一条都不回调。原因：订阅写在ConnectAsync之前，或者 Broker 还在StartAsync的过程中客户端就发了订阅。MQTT 的订阅是会话内的操作，连接没建立完成前发的 SUBSCRIBE 会被丢弃或直接抛异常。 解决：顺序卡死为「起 Broker → await 连接 → await 订阅」，三步都用await，中间不打岔。订阅完再在界面打一行「已订阅 xxx」，这行日志比任何断点都直观。

现象二：消息一多就抛「线程间操作无效: 从不是创建控件的线程访问它」，偶尔还伴随界面假死。原因：ApplicationMessageReceivedAsync的回调线程是 MQTTnet 内部的 IO 线程，不是 UI 线程；同时用Invoke同步切线程，在消息密集时把 IO 线程堵住了。 解决：回调里只做入队，UI 用定时器批量刷新。已经写成Invoke的项目，改成BeginInvoke能先止血，再逐步换成队列。

现象三：程序关掉再开，报「通常每个套接字地址只允许使用一次」，1883 起不来。原因：上次关闭时没走StopAsync，或者直接结束进程，端口还处在释放过程中；netstat -ano | findstr 1883能看到占用。 解决：FormClosing里按「断客户端 → 停服务端 → Dispose」的顺序走一遍并等待完成；调试期间频繁重启的话，端口改成可配置，换一个就能继续干活。

现象四：文件里出现大量重复行，topic 和 payload 完全一样，时间戳只差毫秒。原因：两个重叠的订阅过滤器同时命中，比如demo/+/state和demo/#；或者 QoS 1 下 PUBACK 丢失导致的重发。 解决：先看重复行的时间差，毫秒级且成对出现基本是订阅重叠，收敛过滤器；时间差在秒级、隔一段来一次的，是 QoS 重发，落盘时按「ClientId + 消息标识」做短窗口去重。这两种的排查方向完全不同，别一上来就怀疑 Broker。

现象五：跑了两三个小时，内存涨到几个 G，保存按钮点下去没反应。原因：界面控件里保存了全部历史消息，TextBox的行数没有上限；同时队列无界，磁盘慢的时候堆积。 解决：界面只保留最近 N 条（比如 2000 行，超了截断头部），完整数据靠落盘文件；队列设上限并统计丢弃数。界面是给人看的，不是数据存储，这条边界一定要划清。

6. 进阶：断线重连、消息补发与「消息是否真落盘」的验证手法

联调跑通之后，真正决定这套东西能不能交付的，是断线之后的行为。最小的重连实现是在DisconnectedAsync里起一个带退避的循环，退避时间从 1 秒翻倍到 30 秒封顶，重连成功后重置；不要在回调里直接while(true)重连，那会把回调线程占死。

client.DisconnectedAsync += async e => { var delay = TimeSpan.FromSeconds(1); while (!_cts.IsCancellationRequested) { try { await Task.Delay(delay, _cts.Token); await client.ConnectAsync(options, _cts.Token); await SubscribeAllAsync(); // 重连后必须重新订阅 break; } catch { delay = TimeSpan.FromSeconds(Math.Min(delay.TotalSeconds * 2, 30)); } } };

重连成功之后一定要重新订阅，CleanSession=true时服务端不会替你记着订阅关系，很多「重连后收不到消息」都是漏了这一步。

至于验证消息有没有真的落盘，我习惯用对账法：界面上维护一个发布计数器，落盘线程维护一个写入计数器，再在关闭窗口时统计文件行数，三个数对不上就说明有环节在偷偷丢。差值稳定在一个小常数，通常是队列末尾还在刷；差值持续扩大，就是写盘异常被 catch 吞掉了，回去看那行失败日志。另外隔三差五用tail看一眼文件末尾，确认最后一行是完整的 JSON 而不是写到一半——进程被强杀时，这个现象很常见。

从那以后我每次做这类落盘功能，都会在关闭流程里强制走一遍「停客户端 → 等队列清空 → 停服务端 → 打印三个计数」，哪怕多等两秒。省下的是第二天早上对着半个文件猜消息去哪的时间。希望帮到你。

本文还有配套的精品资源，点击获取