简介基于C#与MQTTnet模块编写的MQTT通信示例同时覆盖服务端和客户端完整源码适合需要快速搭建MQTT原型、或在.NET平台学习轻量级消息队列的开发者。服务端以控制台程序运行服务逻辑单独封装可平滑扩展为Windows服务客户端提供WPF界面示例并内置连接封装类便于在其他应用或控制台客户端中复用2026年新增了.NET 8.0客户端版本。压缩包共327个文件整体仅2.24MB以C#源代码、XAML界面、项目工程文件为主同时包含动态链接库、调试符号及JSON配置等服务端与客户端目录分离结构完整在Visual Studio中即可打开编译。从服务承载到界面交互均有可运行示例源码结构清晰便于二次开发目前已有398人学习适合具备一定C#基础、希望掌握MQTT服务端与客户端接入流程的开发者参考。 最近在弄一个停车场对接项目海康的车牌识别相机把每辆车的进出记录通过MQTT协议推出来我的任务是用C#写一个服务把这些消息稳定接住同时给相机下发开闸、补光等指令。前后折腾了一周服务端、客户端整套代码都跑通了。做完之后有个很深的感受MQTT在C#生态里的资料其实不少但大多数教程只讲客户端怎么连公共Broker自己启动一个服务端、再把客户端完整串起来的例子特别少。这篇文章就是来补这份缺口的——用C#和MQTTnet这个库从零搭一个MQTT服务端再写一个能订阅、能发布的客户端最后结合车牌识别相机的实际对接场景把Topic设计、消息可靠性和那些网上不常说的坑一次性讲透。1. 先想清楚C#项目里为什么需要自建MQTT服务端很多刚接触MQTT的人会问一个问题网上到处都是EMQX、Mosquitto、HiveMQ这些现成的Broker我直接部署一个不就行了为什么还要用C#自己写服务端这个问题本质上取决于你的部署场景。如果是企业级物联网平台的正式环境用独立Broker当然是对的因为性能、集群、监控都在那里。但C#项目里至少有三类场景自建Broker反而更省事。1.1 三种部署形态怎么选第一种是独立Broker部署。一台服务器上装Mosquitto或EMQXC#程序只做客户端连接上去。这种形态好处是Broker功能完整坏处是运维成本高要单独维护一个中间件进程而且在内网边缘端的Windows机器上装Linux的Broker并不方便。第二种是嵌入式Broker这也是我最常用的一种。把MQTTnet的服务端库直接嵌进C#程序里程序启动的同时Broker就起来了别的设备比如车牌识别相机、门禁控制器、传感器直接连接这个进程。整个系统是单进程的部署只需要一个exe特别适合做上位机、边缘网关、单机管理系统。第三种是混合模式。C#服务嵌入一个Broker的同时自己又以客户端的身份连接到上级Broker把数据上报到数据中心。这种模式说白了就是边缘网关的经典架构数据在本地做实时逻辑再向云端同步。1.2 MQTTnet这个库干了哪两件事MQTTnet是.NET生态里最常用的MQTT库而且它跟一般只做客户端的库不一样同一个包同时提供Server和Client两套API不管你是自建Broker还是写设备端一个NuGet依赖全搞定。这一点在做嵌入式Broker的时候特别舒服因为服务端和客户端用的都是同一套消息模型和编码逻辑类型转换的摩擦几乎没有。需要注意版本差异。MQTTnet 3.x时代的API是用事件委托的方式挂在client/server实例上比如UseClientConnectedHandler、UseApplicationMessageReceivedHandler。到4.x之后就改成了Async事件属性类似ClientConnectedAsync、ApplicationMessageReceivedAsync用直接挂载。网上很多老教程拿到4.x的包会编译不过多半就是API形态变了。我下面的代码统一用4.x风格如果你用3.x需要做一下适配。2. 服务端搭建用MQTTnet写一个能上线的Broker如果你只想快速起一个Broker给同事测试十行代码就够了。但要做成一个能长期跑、能定位问题、能控制权限的服务端有几个扩展点必须接上。2.1 最小服务端代码新建一个控制台项目NuGet安装MQTTnet然后写下这段using MQTTnet; using MQTTnet.Server; var options new MqttServerOptionsBuilder() .WithDefaultEndpoint() .WithDefaultEndpointPort(1883) .WithMaxPendingMessagesPerClient(1000) .Build(); var server new MqttFactory().CreateMqttServer(options); await server.StartAsync(); Console.WriteLine(MQTT 服务端已启动监听 1883 端口); Console.ReadLine();这段代码里的核心是WithDefaultEndpoint()它让Broker监听所有网卡的指定端口。端口默认就是1883除非你本机被占用否则不用改。WithMaxPendingMessagesPerClient(1000)是给每个客户端设置待处理消息队列上限防止某台设备消费太慢导致内存被撑爆。细心的朋友会发现这段代码没有做任何鉴权局域网内谁都能连还能任意订阅任何Topic。做原型可以上生产必须加校验。2.2 客户端连接管理与鉴权MQTTnet服务端提供了一系列Async事件我建议至少挂三个ValidatingConnectionAsync做连接鉴权ClientConnectedAsync记录上线ClientDisconnectedAsync记录离线。server.ValidatingConnectionAsync e { Console.WriteLine($客户端 {e.ClientId} 正在连接用户名{e.UserName}); // 这里按你的业务规则校验比如此处只允许用户名密码匹配的设备接入 if (e.UserName ! camera || e.Password ! camera123) { e.ReasonCode MqttConnectReasonCode.BadUserNameOrPassword; return Task.CompletedTask; } // 拒绝重复ClientId避免互相踢线 return Task.CompletedTask; }; server.ClientConnectedAsync e { Console.WriteLine($客户端 {e.ClientId} 已连接); return Task.CompletedTask; }; server.ClientDisconnectedAsync e { Console.WriteLine($客户端 {e.ClientId} 已断开类型{e.DisconnectType}); return Task.CompletedTask; };ValidatingConnectionAsync里把e.ReasonCode设置成非成功值MQTT握手就会失败客户端会收到5.0协议里对应的原因码。这个事件是线程并发触发的所以里面不要做阻塞型操作比如别直接查数据库最多查一下缓存字典。ClientDisconnectedAsync里的e.DisconnectType能区分是客户端主动断开还是异常掉线这对统计设备在线率很有用。我在停车场项目里就靠它判断相机网络是否稳定。2.3 服务端扩展点消息记录与发布拦截除了连接事件服务端还有两个重要扩展点。InterceptingPublishAsync可以在服务端收到任何发布消息时拦截一把你可以记录全量消息日志也可以按Topic前缀决定是否允许这个Topic在Broker上流转。InterceptingSubscriptionAsync则用来控制订阅权比如客户端试图订阅admin/#这类高权限Topic时在这里直接拒绝掉。这两个拦截器是自建Broker最值钱的地方。因为独立Broker要加这种逻辑得写插件而用MQTTnet你直接在事件回调里写业务代码整个链路都在自己进程里调试起来非常顺。server.InterceptingPublishAsync e { var topic e.ApplicationMessage.Topic; var payload System.Text.Encoding.UTF8.GetString(e.ApplicationMessage.Payload); Console.WriteLine($[消息日志] {topic} - {payload}); return Task.CompletedTask; };不过要提醒一句消息量大的时候全量打印日志对吞吐影响很明显。我一般只在调试模式开启正式环境改成按Topic过滤或者只记录异常设备的消息。3. 客户端实现从连接、订阅到发布的完整链路很多人以为写MQTT客户端就是连上Broker然后发消息但真正上手会发现连接参数、订阅通配符、回调线程这几件事如果没搞清楚项目跑起来全是坑。这部分我按完整流程拆开讲。3.1 连接参数详解客户端连服务端之前必须先想清楚ClientId、CleanSession和KeepAlive这三个参数。ClientId是客户端在Broker上的唯一身份标识。如果两个客户端用同一个ClientId连接MQTT协议规定后连的会把先连的踢下线而且踢完之后先连的客户端可能完全没有感知下次发消息才发现连接已经断了。所以同一个设备在不同进程中必须用不同的ClientId我习惯把设备编码和进程号拼在一起。WithCleanSession()表示每次连接是干净的会话Broker不会为这个客户端保留任何历史订阅和离线消息。如果希望断线后再上线还能收到离线期间的消息就要关掉CleanSession配合QoS1/2做持久会话。但持久会话对Broker内存有压力停车场这类场景一般用CleanSession就够了。KeepAlive是心跳包间隔超过这个时间Broker还没收到客户端任何数据就判定客户端失联。MQTTnet默认是30秒有些相机厂商默认值很短几秒就一次导致网络抖动时频繁掉线我通常把服务端和客户端都调到30秒以上。连接代码using MQTTnet; using MQTTnet.Client; var mqttFactory new MqttFactory(); var client mqttFactory.CreateMqttClient(); var options new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) .WithClientId(dotnet-client-001) .WithCredentials(camera, camera123) .WithCleanSession() .WithKeepAlivePeriod(TimeSpan.FromSeconds(30)) .Build(); await client.ConnectAsync(options); Console.WriteLine(客户端连接成功);3.2 订阅与接收消息订阅的时候要确定Topic和QoS。Topic支持通配符匹配一层#匹配多层。在停车场项目里我订阅vehicle/#就能收到所有vehicle开头的消息不管是哪个停车场、哪台相机上送的。await client.SubscribeAsync(new MqttClientFilterOptionsBuilder() .WithTopic(vehicle/#) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .Build());接收消息通过ApplicationMessageReceivedAsync事件这是客户端最核心的回调点client.ApplicationMessageReceivedAsync e { var topic e.ApplicationMessage.Topic; var payload e.ApplicationMessage.Payload; var msg Encoding.UTF8.GetString(payload); Console.WriteLine($收到消息主题{topic}内容{msg}); return Task.CompletedTask; };这里有几个细节。第一e.ApplicationMessage.Payload在4.x版本里是byte[]类型直接用Encoding.UTF8.GetString转字符串即可。第二这个回调跑在MQTTnet的内部线程池上不要在回调里做耗时操作比如直接查数据库、调第三方接口、读写大文件否则会阻塞后续消息的送达。我在项目里的做法是回调里只解析消息解析完丢进Channel或BlockingCollection由业务线程去消费。第三如果消息是中文一定要确认发送方用的什么编码。海康相机默认走UTF-8但有些定制固件会发GBK收到乱码时先怀疑编码别一头扎进协议解析里。3.3 发布消息发布消息和订阅一样核心是构建一个MqttApplicationMessage对象。实际业务里发布的基本都是JSON结构的数据我习惯直接用System.Text.Json序列化。using System.Text.Json; var data new { plate 京A12345, inTime DateTime.Now.ToString(yyyy-MM-dd HH:mm:ss) }; var message new MqttApplicationMessageBuilder() .WithTopic(vehicle/parking01/device01/in) .WithPayload(JsonSerializer.Serialize(data)) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .WithRetainFlag(false) .Build(); await client.PublishAsync(message);发布的时候WithTopic里的Topic路径不能为空WithPayload可以传字符串或byte[]。WithQualityOfServiceLevel在发布和订阅侧都要设置具体的质量选择我放在下一节专门讲。4. 消息靠不靠谱看这三个开关QoS、遗嘱、保留消息MQTT协议最容易被忽略的是消息可靠性但实际跑生产环境掉消息、设备失联、状态不同步这些问题基本都是没用好QoS、遗嘱消息和保留消息这三个机制。4.1 QoS 0/1/2到底怎么选QoS全称Quality of Service规定了发布者和订阅者之间消息投递的保证级别。三者对比级别名称投递保证开销适用场景QoS 0最多一次消息可能丢失最低实时性要求低、允许丢数据的环境监测QoS 1至少一次消息不丢可能重复中等业务指令、车辆记录推送QoS 2恰好一次不丢不重最高扣费、关键事务类消息我自己的选择习惯是核心业务消息用QoS1绝不用QoS0也尽量不上QoS2。原因很简单QoS0一旦网络抖动消息就没了而且没有错误反馈QoS2需要发送方和接收方两轮握手确认在弱网环境下延迟明显吞吐也会掉一截。QoS1会出现重复投递所以接收端要做好幂等比如用消息里的唯一编号判断是否处理过。在MQTTnet里订阅和发布都可以指定QoS// 订阅端 .WithTopic(vehicle/#) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) // 发布端 .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)要注意订阅的QoS表示接收方希望收到的最大级别发布的QoS表示发送方投递的级别Broker最终实际投递的QoS是两者取更低的那一个。4.2 遗嘱消息设备掉线自动通知遗嘱消息Will Message是MQTT里一个特别实用的机制。客户端在连接时可以声明一份消息如果它在没有正常发送DISCONNECT报文的情况下连接断开比如断电、断网、进程被杀Broker就会替它发布这条遗嘱消息。这相当于设备失联的最后一声通知。在MQTTnet里设置遗嘱是在构造连接参数时用WithWillvar willMsg new MqttApplicationMessageBuilder() .WithTopic(device/status) .WithPayload(offline) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce) .WithRetainFlag(true) .Build(); var options new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) .WithClientId(device-001) .WithWill(willMsg) .Build();我把遗嘱消息的QoS设成2、Retain设成true这样别的订阅端一上线马上就能拿到最新的设备状态而不是等下一次心跳超时才能判断设备掉线。4.3 保留消息新订阅者上线立即得到状态保留消息Retain是另一个关键开关。普通消息发给Broker之后后续才订阅这个Topic的客户端是收不到历史消息的。但如果发布消息时打了Retain标记Broker就会把这条消息持久保存新客户端订阅这个Topic时立即就收到这份最新的保留消息。典型的应用是设备状态、相机配置、系统版本号这些需要不断轮询获取的信息。设备上线时发布一条带Retain的消息var message new MqttApplicationMessageBuilder() .WithTopic(device/status) .WithPayload(online) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .WithRetainFlag(true) .Build();这样不管哪个客户端后来的订阅device/status马上就知道设备在线不用攒一段时间的消息才能猜状态。这在停车场场景里非常有用管理端一打开就能看到所有相机当前在线与否而不是等每台设备发消息才知道。5. 真实场景实战海康/大华车牌识别相机如何对接前面讲的是基础能力这一节我拿实际项目说事。无论海康还是大华新一代的车牌识别相机基本都内置了MQTT功能可以把识别结果主动推送到指定的Broker。但厂商之间的Topic命名、消息字段、触发条件都不一样第一次做这个对接一定会遇到各种奇奇怪怪的问题。5.1 相机侧配置在相机网页管理后台里一般会有一个“MQTT配置”或“网络服务”的入口。需要填的核心参数是Broker地址、端口、用户名密码、以及上报的Topic前缀。有些相机还允许你配置推送的消息内容模板。海康相机推送的车辆识别结果通常是一段JSON字段包括车牌号、车辆类型、车位号、抓拍时间、出入场方向等。大华的格式则不一样字段名可能是英文字母缩写。我的建议是第一次对接时先把相机的原始报文完整打出来看一眼不要上来就按文档解析因为实际固件版本经常跟文档不一致。我当时的做法是在服务端拦截器里全量打印消息日志抓了一天的包对比了几台不同固件版本的相机最后才把字段映射表定下来。5.2 Topic设计层级结构让订阅变得简单Topic命名是MQTT项目最容易被忽视的架构决策。我推荐按“业务域/地点/设备/事件类型”的层级来设计比如vehicle/{parkingLotId}/{deviceId}/in vehicle/{parkingLotId}/{deviceId}/out device/{deviceId}/status device/{deviceId}/command用这种结构的好处是订阅方可以很灵活地过滤。管理端要收整个停车场所有进口相机的事件就订阅vehicle/parking01//in要收某台相机的所有状态就订阅device/camera_0001/#要收整个系统的所有消息就订阅#。我踩过一个坑最开始把停车场ID和设备ID放在Topic的末尾比如vehicle/in/parking01/device01结果后面想订阅“所有停车场所有进车”的时候得写上vehicle/in//层级多且容易写错。调整为前面的结构之后订阅灵活度一下子提高了。5.3 C#侧解析与下发指令假设相机识别到一辆车入场推送上来的消息结构类似{ plate: 京A12345, parkingLotId: parking01, deviceId: camera_0001, inTime: 2025-01-15 08:30:00, imageUrl: http://192.168.1.20/capture/123.jpg }C#客户端订阅了vehicle/#之后收到消息先反序列化然后按业务逻辑处理public class VehicleEvent { public string Plate { get; set; } public string ParkingLotId { get; set; } public string DeviceId { get; set; } public string InTime { get; set; } public string ImageUrl { get; set; } } // 在 ApplicationMessageReceivedAsync 回调里 var vehicleEvent JsonSerializer.DeserializeVehicleEvent(msg); if (vehicleEvent ! null) { Console.WriteLine(${vehicleEvent.Plate} 进入 {vehicleEvent.ParkingLotId}); // 后续处理写数据库、推送通知、给相机回开闸指令 }如果业务规则允许车辆自动入场还要向相机的指令Topic发布一条开闸消息var command new { cmd open_gate, plate vehicleEvent.Plate }; var message new MqttApplicationMessageBuilder() .WithTopic($device/{vehicleEvent.DeviceId}/command) .WithPayload(JsonSerializer.Serialize(command)) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce) .Build(); await client.PublishAsync(message);这里我特意用了QoS2因为开闸指令不允许丢也不允许重复执行所以接收端处理时一定要做去重比如记录指令编号防止网络重发导致闸机重复抬起。6. 实测中踩过的坑以及稳定性调优建议协议层和代码层都通了之后真正的麻烦才开始。下面这几个问题我基本在头两个项目里全遇了一遍整理成清单照着排查能省很多时间。6.1 断线重连必须自己写MQTTnet的ConnectAsync失败或者连接中途断开库本身不会自动重连。如果程序只是启动时连一次网络一抖或者Broker重启客户端就再也不会连上这是最隐蔽的生产事故。我一般封装一个带指数退避的重连方法public static async Task ConnectWithRetryAsync(IMqttClient client, MqttClientOptions options) { int retry 1; while (true) { try { if (!client.IsConnected) { await client.ConnectAsync(options, CancellationToken.None); Console.WriteLine(连接成功); } break; } catch (Exception ex) { var delay TimeSpan.FromSeconds(Math.Min(Math.Pow(2, retry), 30)); Console.WriteLine($连接失败{ex.Message}{delay.TotalSeconds}秒后重试); await Task.Delay(delay); retry; } } }重试间隔从1秒开始翻倍封顶30秒这样Broker刚重启的几分钟内客户端不会用1秒一次的频率疯狂冲击它。6.2 ClientId冲突导致设备互相踢下线相机的MQTT参数里ClientId通常是设备名一旦有两台相机用了相同的名字这种情况在复制配置时太常见了它们会轮流把对方踢下线表现就是相机在线的日志断断续续每隔几分钟掉一次线非常诡异。排查方法是在服务端的ClientDisconnectedAsync里打日志如果看到同一个ClientId反复“已连接—已断开—已连接”循环基本就是ClientId冲突。解决方式是给每台相机的ClientId加上IP或编号后缀保证全局唯一。6.3 WinForm/WPF里回调线程更新控件如果用C#上位机做MQTT客户端收到的消息默认在后台线程触发直接操作TextBox、Label这些控件会抛跨线程异常。标准做法是判断InvokeRequired然后交给UI线程更新if (textBox1.InvokeRequired) { textBox1.Invoke(new Action(() textBox1.AppendText(msg Environment.NewLine))); } else { textBox1.AppendText(msg Environment.NewLine); }如果消息频率很高频繁Invoke会让UI卡顿。我的经验是先在回调里把消息累积到一个队列UI定时器每100毫秒刷新一次界面既流畅又不丢消息。6.4 消息体大小、防火墙与默认配置MQTT对单个消息体大小是有上限的MQTTnet服务端默认可以改但很多网络环境会限制单包大小。相机抓拍图片通常不会通过MQTT传输但有些业务会把Base64编码的小图塞进Payload一塞就超过1MB导致消息怎么发都发不出去或者被Broker静默丢弃。我的底线是MQTT只传结构化数据和短文本图片、视频一律走HTTP或者文件通道消息里只放URL。防火墙这块最容易被忽略。Broker装好了客户端在同机连得上换一台机器就连接超时十有八九是1883端口没放行。Windows平台记得在高级防火墙里添加入站规则Linux生产环境用firewall-cmd或ufw放行TCP 1883。最后再看一下参数调优。网络环境差的话把KeepAlive从默认30秒适当调大减少无效心跳订阅数量大的客户端可以调大MaxPendingMessagesPerClient对实时性要求高的场景把Broker的默认消息大小上限调上去同时对超大Payload做拦截。这些坑没有一个是高深的原理但每一个都能让线上系统莫名其妙地出问题。做MQTT对接先把这些底层的稳定性问题解决掉再去调业务逻辑整个项目会顺畅很多。本文还有配套的精品资源点击获取
MQTTnet实战：C#搭建MQTT服务端与客户端，对接车牌识别相机
简介基于C#与MQTTnet模块编写的MQTT通信示例同时覆盖服务端和客户端完整源码适合需要快速搭建MQTT原型、或在.NET平台学习轻量级消息队列的开发者。服务端以控制台程序运行服务逻辑单独封装可平滑扩展为Windows服务客户端提供WPF界面示例并内置连接封装类便于在其他应用或控制台客户端中复用2026年新增了.NET 8.0客户端版本。压缩包共327个文件整体仅2.24MB以C#源代码、XAML界面、项目工程文件为主同时包含动态链接库、调试符号及JSON配置等服务端与客户端目录分离结构完整在Visual Studio中即可打开编译。从服务承载到界面交互均有可运行示例源码结构清晰便于二次开发目前已有398人学习适合具备一定C#基础、希望掌握MQTT服务端与客户端接入流程的开发者参考。 最近在弄一个停车场对接项目海康的车牌识别相机把每辆车的进出记录通过MQTT协议推出来我的任务是用C#写一个服务把这些消息稳定接住同时给相机下发开闸、补光等指令。前后折腾了一周服务端、客户端整套代码都跑通了。做完之后有个很深的感受MQTT在C#生态里的资料其实不少但大多数教程只讲客户端怎么连公共Broker自己启动一个服务端、再把客户端完整串起来的例子特别少。这篇文章就是来补这份缺口的——用C#和MQTTnet这个库从零搭一个MQTT服务端再写一个能订阅、能发布的客户端最后结合车牌识别相机的实际对接场景把Topic设计、消息可靠性和那些网上不常说的坑一次性讲透。1. 先想清楚C#项目里为什么需要自建MQTT服务端很多刚接触MQTT的人会问一个问题网上到处都是EMQX、Mosquitto、HiveMQ这些现成的Broker我直接部署一个不就行了为什么还要用C#自己写服务端这个问题本质上取决于你的部署场景。如果是企业级物联网平台的正式环境用独立Broker当然是对的因为性能、集群、监控都在那里。但C#项目里至少有三类场景自建Broker反而更省事。1.1 三种部署形态怎么选第一种是独立Broker部署。一台服务器上装Mosquitto或EMQXC#程序只做客户端连接上去。这种形态好处是Broker功能完整坏处是运维成本高要单独维护一个中间件进程而且在内网边缘端的Windows机器上装Linux的Broker并不方便。第二种是嵌入式Broker这也是我最常用的一种。把MQTTnet的服务端库直接嵌进C#程序里程序启动的同时Broker就起来了别的设备比如车牌识别相机、门禁控制器、传感器直接连接这个进程。整个系统是单进程的部署只需要一个exe特别适合做上位机、边缘网关、单机管理系统。第三种是混合模式。C#服务嵌入一个Broker的同时自己又以客户端的身份连接到上级Broker把数据上报到数据中心。这种模式说白了就是边缘网关的经典架构数据在本地做实时逻辑再向云端同步。1.2 MQTTnet这个库干了哪两件事MQTTnet是.NET生态里最常用的MQTT库而且它跟一般只做客户端的库不一样同一个包同时提供Server和Client两套API不管你是自建Broker还是写设备端一个NuGet依赖全搞定。这一点在做嵌入式Broker的时候特别舒服因为服务端和客户端用的都是同一套消息模型和编码逻辑类型转换的摩擦几乎没有。需要注意版本差异。MQTTnet 3.x时代的API是用事件委托的方式挂在client/server实例上比如UseClientConnectedHandler、UseApplicationMessageReceivedHandler。到4.x之后就改成了Async事件属性类似ClientConnectedAsync、ApplicationMessageReceivedAsync用直接挂载。网上很多老教程拿到4.x的包会编译不过多半就是API形态变了。我下面的代码统一用4.x风格如果你用3.x需要做一下适配。2. 服务端搭建用MQTTnet写一个能上线的Broker如果你只想快速起一个Broker给同事测试十行代码就够了。但要做成一个能长期跑、能定位问题、能控制权限的服务端有几个扩展点必须接上。2.1 最小服务端代码新建一个控制台项目NuGet安装MQTTnet然后写下这段using MQTTnet; using MQTTnet.Server; var options new MqttServerOptionsBuilder() .WithDefaultEndpoint() .WithDefaultEndpointPort(1883) .WithMaxPendingMessagesPerClient(1000) .Build(); var server new MqttFactory().CreateMqttServer(options); await server.StartAsync(); Console.WriteLine(MQTT 服务端已启动监听 1883 端口); Console.ReadLine();这段代码里的核心是WithDefaultEndpoint()它让Broker监听所有网卡的指定端口。端口默认就是1883除非你本机被占用否则不用改。WithMaxPendingMessagesPerClient(1000)是给每个客户端设置待处理消息队列上限防止某台设备消费太慢导致内存被撑爆。细心的朋友会发现这段代码没有做任何鉴权局域网内谁都能连还能任意订阅任何Topic。做原型可以上生产必须加校验。2.2 客户端连接管理与鉴权MQTTnet服务端提供了一系列Async事件我建议至少挂三个ValidatingConnectionAsync做连接鉴权ClientConnectedAsync记录上线ClientDisconnectedAsync记录离线。server.ValidatingConnectionAsync e { Console.WriteLine($客户端 {e.ClientId} 正在连接用户名{e.UserName}); // 这里按你的业务规则校验比如此处只允许用户名密码匹配的设备接入 if (e.UserName ! camera || e.Password ! camera123) { e.ReasonCode MqttConnectReasonCode.BadUserNameOrPassword; return Task.CompletedTask; } // 拒绝重复ClientId避免互相踢线 return Task.CompletedTask; }; server.ClientConnectedAsync e { Console.WriteLine($客户端 {e.ClientId} 已连接); return Task.CompletedTask; }; server.ClientDisconnectedAsync e { Console.WriteLine($客户端 {e.ClientId} 已断开类型{e.DisconnectType}); return Task.CompletedTask; };ValidatingConnectionAsync里把e.ReasonCode设置成非成功值MQTT握手就会失败客户端会收到5.0协议里对应的原因码。这个事件是线程并发触发的所以里面不要做阻塞型操作比如别直接查数据库最多查一下缓存字典。ClientDisconnectedAsync里的e.DisconnectType能区分是客户端主动断开还是异常掉线这对统计设备在线率很有用。我在停车场项目里就靠它判断相机网络是否稳定。2.3 服务端扩展点消息记录与发布拦截除了连接事件服务端还有两个重要扩展点。InterceptingPublishAsync可以在服务端收到任何发布消息时拦截一把你可以记录全量消息日志也可以按Topic前缀决定是否允许这个Topic在Broker上流转。InterceptingSubscriptionAsync则用来控制订阅权比如客户端试图订阅admin/#这类高权限Topic时在这里直接拒绝掉。这两个拦截器是自建Broker最值钱的地方。因为独立Broker要加这种逻辑得写插件而用MQTTnet你直接在事件回调里写业务代码整个链路都在自己进程里调试起来非常顺。server.InterceptingPublishAsync e { var topic e.ApplicationMessage.Topic; var payload System.Text.Encoding.UTF8.GetString(e.ApplicationMessage.Payload); Console.WriteLine($[消息日志] {topic} - {payload}); return Task.CompletedTask; };不过要提醒一句消息量大的时候全量打印日志对吞吐影响很明显。我一般只在调试模式开启正式环境改成按Topic过滤或者只记录异常设备的消息。3. 客户端实现从连接、订阅到发布的完整链路很多人以为写MQTT客户端就是连上Broker然后发消息但真正上手会发现连接参数、订阅通配符、回调线程这几件事如果没搞清楚项目跑起来全是坑。这部分我按完整流程拆开讲。3.1 连接参数详解客户端连服务端之前必须先想清楚ClientId、CleanSession和KeepAlive这三个参数。ClientId是客户端在Broker上的唯一身份标识。如果两个客户端用同一个ClientId连接MQTT协议规定后连的会把先连的踢下线而且踢完之后先连的客户端可能完全没有感知下次发消息才发现连接已经断了。所以同一个设备在不同进程中必须用不同的ClientId我习惯把设备编码和进程号拼在一起。WithCleanSession()表示每次连接是干净的会话Broker不会为这个客户端保留任何历史订阅和离线消息。如果希望断线后再上线还能收到离线期间的消息就要关掉CleanSession配合QoS1/2做持久会话。但持久会话对Broker内存有压力停车场这类场景一般用CleanSession就够了。KeepAlive是心跳包间隔超过这个时间Broker还没收到客户端任何数据就判定客户端失联。MQTTnet默认是30秒有些相机厂商默认值很短几秒就一次导致网络抖动时频繁掉线我通常把服务端和客户端都调到30秒以上。连接代码using MQTTnet; using MQTTnet.Client; var mqttFactory new MqttFactory(); var client mqttFactory.CreateMqttClient(); var options new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) .WithClientId(dotnet-client-001) .WithCredentials(camera, camera123) .WithCleanSession() .WithKeepAlivePeriod(TimeSpan.FromSeconds(30)) .Build(); await client.ConnectAsync(options); Console.WriteLine(客户端连接成功);3.2 订阅与接收消息订阅的时候要确定Topic和QoS。Topic支持通配符匹配一层#匹配多层。在停车场项目里我订阅vehicle/#就能收到所有vehicle开头的消息不管是哪个停车场、哪台相机上送的。await client.SubscribeAsync(new MqttClientFilterOptionsBuilder() .WithTopic(vehicle/#) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .Build());接收消息通过ApplicationMessageReceivedAsync事件这是客户端最核心的回调点client.ApplicationMessageReceivedAsync e { var topic e.ApplicationMessage.Topic; var payload e.ApplicationMessage.Payload; var msg Encoding.UTF8.GetString(payload); Console.WriteLine($收到消息主题{topic}内容{msg}); return Task.CompletedTask; };这里有几个细节。第一e.ApplicationMessage.Payload在4.x版本里是byte[]类型直接用Encoding.UTF8.GetString转字符串即可。第二这个回调跑在MQTTnet的内部线程池上不要在回调里做耗时操作比如直接查数据库、调第三方接口、读写大文件否则会阻塞后续消息的送达。我在项目里的做法是回调里只解析消息解析完丢进Channel或BlockingCollection由业务线程去消费。第三如果消息是中文一定要确认发送方用的什么编码。海康相机默认走UTF-8但有些定制固件会发GBK收到乱码时先怀疑编码别一头扎进协议解析里。3.3 发布消息发布消息和订阅一样核心是构建一个MqttApplicationMessage对象。实际业务里发布的基本都是JSON结构的数据我习惯直接用System.Text.Json序列化。using System.Text.Json; var data new { plate 京A12345, inTime DateTime.Now.ToString(yyyy-MM-dd HH:mm:ss) }; var message new MqttApplicationMessageBuilder() .WithTopic(vehicle/parking01/device01/in) .WithPayload(JsonSerializer.Serialize(data)) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .WithRetainFlag(false) .Build(); await client.PublishAsync(message);发布的时候WithTopic里的Topic路径不能为空WithPayload可以传字符串或byte[]。WithQualityOfServiceLevel在发布和订阅侧都要设置具体的质量选择我放在下一节专门讲。4. 消息靠不靠谱看这三个开关QoS、遗嘱、保留消息MQTT协议最容易被忽略的是消息可靠性但实际跑生产环境掉消息、设备失联、状态不同步这些问题基本都是没用好QoS、遗嘱消息和保留消息这三个机制。4.1 QoS 0/1/2到底怎么选QoS全称Quality of Service规定了发布者和订阅者之间消息投递的保证级别。三者对比级别名称投递保证开销适用场景QoS 0最多一次消息可能丢失最低实时性要求低、允许丢数据的环境监测QoS 1至少一次消息不丢可能重复中等业务指令、车辆记录推送QoS 2恰好一次不丢不重最高扣费、关键事务类消息我自己的选择习惯是核心业务消息用QoS1绝不用QoS0也尽量不上QoS2。原因很简单QoS0一旦网络抖动消息就没了而且没有错误反馈QoS2需要发送方和接收方两轮握手确认在弱网环境下延迟明显吞吐也会掉一截。QoS1会出现重复投递所以接收端要做好幂等比如用消息里的唯一编号判断是否处理过。在MQTTnet里订阅和发布都可以指定QoS// 订阅端 .WithTopic(vehicle/#) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) // 发布端 .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)要注意订阅的QoS表示接收方希望收到的最大级别发布的QoS表示发送方投递的级别Broker最终实际投递的QoS是两者取更低的那一个。4.2 遗嘱消息设备掉线自动通知遗嘱消息Will Message是MQTT里一个特别实用的机制。客户端在连接时可以声明一份消息如果它在没有正常发送DISCONNECT报文的情况下连接断开比如断电、断网、进程被杀Broker就会替它发布这条遗嘱消息。这相当于设备失联的最后一声通知。在MQTTnet里设置遗嘱是在构造连接参数时用WithWillvar willMsg new MqttApplicationMessageBuilder() .WithTopic(device/status) .WithPayload(offline) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce) .WithRetainFlag(true) .Build(); var options new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) .WithClientId(device-001) .WithWill(willMsg) .Build();我把遗嘱消息的QoS设成2、Retain设成true这样别的订阅端一上线马上就能拿到最新的设备状态而不是等下一次心跳超时才能判断设备掉线。4.3 保留消息新订阅者上线立即得到状态保留消息Retain是另一个关键开关。普通消息发给Broker之后后续才订阅这个Topic的客户端是收不到历史消息的。但如果发布消息时打了Retain标记Broker就会把这条消息持久保存新客户端订阅这个Topic时立即就收到这份最新的保留消息。典型的应用是设备状态、相机配置、系统版本号这些需要不断轮询获取的信息。设备上线时发布一条带Retain的消息var message new MqttApplicationMessageBuilder() .WithTopic(device/status) .WithPayload(online) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .WithRetainFlag(true) .Build();这样不管哪个客户端后来的订阅device/status马上就知道设备在线不用攒一段时间的消息才能猜状态。这在停车场场景里非常有用管理端一打开就能看到所有相机当前在线与否而不是等每台设备发消息才知道。5. 真实场景实战海康/大华车牌识别相机如何对接前面讲的是基础能力这一节我拿实际项目说事。无论海康还是大华新一代的车牌识别相机基本都内置了MQTT功能可以把识别结果主动推送到指定的Broker。但厂商之间的Topic命名、消息字段、触发条件都不一样第一次做这个对接一定会遇到各种奇奇怪怪的问题。5.1 相机侧配置在相机网页管理后台里一般会有一个“MQTT配置”或“网络服务”的入口。需要填的核心参数是Broker地址、端口、用户名密码、以及上报的Topic前缀。有些相机还允许你配置推送的消息内容模板。海康相机推送的车辆识别结果通常是一段JSON字段包括车牌号、车辆类型、车位号、抓拍时间、出入场方向等。大华的格式则不一样字段名可能是英文字母缩写。我的建议是第一次对接时先把相机的原始报文完整打出来看一眼不要上来就按文档解析因为实际固件版本经常跟文档不一致。我当时的做法是在服务端拦截器里全量打印消息日志抓了一天的包对比了几台不同固件版本的相机最后才把字段映射表定下来。5.2 Topic设计层级结构让订阅变得简单Topic命名是MQTT项目最容易被忽视的架构决策。我推荐按“业务域/地点/设备/事件类型”的层级来设计比如vehicle/{parkingLotId}/{deviceId}/in vehicle/{parkingLotId}/{deviceId}/out device/{deviceId}/status device/{deviceId}/command用这种结构的好处是订阅方可以很灵活地过滤。管理端要收整个停车场所有进口相机的事件就订阅vehicle/parking01//in要收某台相机的所有状态就订阅device/camera_0001/#要收整个系统的所有消息就订阅#。我踩过一个坑最开始把停车场ID和设备ID放在Topic的末尾比如vehicle/in/parking01/device01结果后面想订阅“所有停车场所有进车”的时候得写上vehicle/in//层级多且容易写错。调整为前面的结构之后订阅灵活度一下子提高了。5.3 C#侧解析与下发指令假设相机识别到一辆车入场推送上来的消息结构类似{ plate: 京A12345, parkingLotId: parking01, deviceId: camera_0001, inTime: 2025-01-15 08:30:00, imageUrl: http://192.168.1.20/capture/123.jpg }C#客户端订阅了vehicle/#之后收到消息先反序列化然后按业务逻辑处理public class VehicleEvent { public string Plate { get; set; } public string ParkingLotId { get; set; } public string DeviceId { get; set; } public string InTime { get; set; } public string ImageUrl { get; set; } } // 在 ApplicationMessageReceivedAsync 回调里 var vehicleEvent JsonSerializer.DeserializeVehicleEvent(msg); if (vehicleEvent ! null) { Console.WriteLine(${vehicleEvent.Plate} 进入 {vehicleEvent.ParkingLotId}); // 后续处理写数据库、推送通知、给相机回开闸指令 }如果业务规则允许车辆自动入场还要向相机的指令Topic发布一条开闸消息var command new { cmd open_gate, plate vehicleEvent.Plate }; var message new MqttApplicationMessageBuilder() .WithTopic($device/{vehicleEvent.DeviceId}/command) .WithPayload(JsonSerializer.Serialize(command)) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce) .Build(); await client.PublishAsync(message);这里我特意用了QoS2因为开闸指令不允许丢也不允许重复执行所以接收端处理时一定要做去重比如记录指令编号防止网络重发导致闸机重复抬起。6. 实测中踩过的坑以及稳定性调优建议协议层和代码层都通了之后真正的麻烦才开始。下面这几个问题我基本在头两个项目里全遇了一遍整理成清单照着排查能省很多时间。6.1 断线重连必须自己写MQTTnet的ConnectAsync失败或者连接中途断开库本身不会自动重连。如果程序只是启动时连一次网络一抖或者Broker重启客户端就再也不会连上这是最隐蔽的生产事故。我一般封装一个带指数退避的重连方法public static async Task ConnectWithRetryAsync(IMqttClient client, MqttClientOptions options) { int retry 1; while (true) { try { if (!client.IsConnected) { await client.ConnectAsync(options, CancellationToken.None); Console.WriteLine(连接成功); } break; } catch (Exception ex) { var delay TimeSpan.FromSeconds(Math.Min(Math.Pow(2, retry), 30)); Console.WriteLine($连接失败{ex.Message}{delay.TotalSeconds}秒后重试); await Task.Delay(delay); retry; } } }重试间隔从1秒开始翻倍封顶30秒这样Broker刚重启的几分钟内客户端不会用1秒一次的频率疯狂冲击它。6.2 ClientId冲突导致设备互相踢下线相机的MQTT参数里ClientId通常是设备名一旦有两台相机用了相同的名字这种情况在复制配置时太常见了它们会轮流把对方踢下线表现就是相机在线的日志断断续续每隔几分钟掉一次线非常诡异。排查方法是在服务端的ClientDisconnectedAsync里打日志如果看到同一个ClientId反复“已连接—已断开—已连接”循环基本就是ClientId冲突。解决方式是给每台相机的ClientId加上IP或编号后缀保证全局唯一。6.3 WinForm/WPF里回调线程更新控件如果用C#上位机做MQTT客户端收到的消息默认在后台线程触发直接操作TextBox、Label这些控件会抛跨线程异常。标准做法是判断InvokeRequired然后交给UI线程更新if (textBox1.InvokeRequired) { textBox1.Invoke(new Action(() textBox1.AppendText(msg Environment.NewLine))); } else { textBox1.AppendText(msg Environment.NewLine); }如果消息频率很高频繁Invoke会让UI卡顿。我的经验是先在回调里把消息累积到一个队列UI定时器每100毫秒刷新一次界面既流畅又不丢消息。6.4 消息体大小、防火墙与默认配置MQTT对单个消息体大小是有上限的MQTTnet服务端默认可以改但很多网络环境会限制单包大小。相机抓拍图片通常不会通过MQTT传输但有些业务会把Base64编码的小图塞进Payload一塞就超过1MB导致消息怎么发都发不出去或者被Broker静默丢弃。我的底线是MQTT只传结构化数据和短文本图片、视频一律走HTTP或者文件通道消息里只放URL。防火墙这块最容易被忽略。Broker装好了客户端在同机连得上换一台机器就连接超时十有八九是1883端口没放行。Windows平台记得在高级防火墙里添加入站规则Linux生产环境用firewall-cmd或ufw放行TCP 1883。最后再看一下参数调优。网络环境差的话把KeepAlive从默认30秒适当调大减少无效心跳订阅数量大的客户端可以调大MaxPendingMessagesPerClient对实时性要求高的场景把Broker的默认消息大小上限调上去同时对超大Payload做拦截。这些坑没有一个是高深的原理但每一个都能让线上系统莫名其妙地出问题。做MQTT对接先把这些底层的稳定性问题解决掉再去调业务逻辑整个项目会顺畅很多。本文还有配套的精品资源点击获取