C# MQTT开发实战：MQTTnet实现服务端与客户端完整指南

简介这是一套直接可运行的 C# MQTT 通信示例基于 MQTTNET 模块进行上层封装完整包含服务端与客户端实现。服务端采用控制台工程并单独封装服务层可自行改造为 Windows 服务客户端使用 WPF 呈现内置连接封装类适合希望快速掌握 MQTT 协议落地与 C# 网络编程的开发者参考。压缩包内为完整源代码项目共 327 个文件以 cs 源码、dll 依赖库、xaml 界面文件及 json 配置为主整体体积仅 2.24MB结构清晰便于直接打开学习。目前已有 398 人学习下载适合作为 MQTT 入门与二次开发的参考资料。资源中还更新了 .NET 8.0 客户端示例开发者可对比不同框架下的实现方式同时服务端分层封装、客户端连接封装等代码能帮助理解通信模块设计、消息发布订阅流程及桌面端接入方法便于直接迁移到实际项目中。1. MQTT到底解决了什么问题MQTTMessage Queuing Telemetry Transport消息队列遥测传输是一种基于发布/订阅模式的轻量级消息协议专门为带宽有限、网络不稳定的场景设计。和传统的 HTTP 请求-响应模型不同MQTT 的通信双方不是直接点对点连接而是通过一个中间者Broker也叫消息代理来转发消息。发消息的一方叫发布者收消息的一方叫订阅者。发布者把消息发到一个“主题”上Broker 负责把这个主题下的消息分发给所有订阅了这个主题的客户端。这个设计最大的好处是解耦。发布者不需要知道哪些客户端在线订阅者也不需要知道消息是从哪台设备来的。拿停车场车牌识别相机对接来说以前用 SDK 直连或者自定义 TCP 协议每接一个新品牌相机都要重新处理一套协议细节换成 MQTT 后相机端只负责往“停车位/入口/相机A/事件”这种主题上发布识别结果后端服务订阅对应主题就行。海康、大华这些主流厂家也都支持 MQTT 上报接入成本降了一大截。在实际项目中MQTT 还有一个很关键的能力遗嘱消息Last Will and Testament和保留消息Retained Message。遗嘱消息用于在客户端异常断开时由 Broker 代替客户端发布一条预设消息让其他订阅方感知到“这台设备掉线了”。保留消息则让新订阅的客户端立刻获得主题的最后一条消息而不是等下一个发布周期。这两个机制对设备状态监控特别有用后面我会在示例里演示。那问题来了C# 这边想用 MQTT到底怎么选库、怎么写服务端和客户端我用一个开箱即用的示例逐步讲。2. C#里MQTT库怎么选C# 生态里 MQTT 相关的库我用过好几个最常被提起的是两个M2Mqtt 和 MQTTnet。M2Mqtt 是老牌库很多老项目在用但它已经很久没有更新了对现代 .NET 的支持一般。如果是维护老系统不要动它继续用没问题新项目不建议选。我推荐的是 MQTTnet理由很直接NuGet 搜索 MQTTnet一行命令就能引用支持 .NET 6/8、.NET Framework 4.6.2服务端和客户端都实现了性能在同级别的 C# 库里算很好的实现了 MQTT 3.1.1 和 5.0 协议QoS 0/1/2 都支持API 是现代异步风格和 async/await 配合很顺手。不过要提醒一句MQTTnet 从 3.x 升级到 4.x 时API 有过一次比较大的调整。网上搜到的很多老示例代码用的还是 UseApplicationMessageReceivedHandler、ServerStarted 这类回调注册方式在 4.x 里已经改成了 Async 事件形式。我下面写的代码基于当前 NuGet 上的 4.x 版本如果你用的是 3.x代码需要做少量迁移。这个事虽然不大但不搞清楚会卡很久。而且用 NuGet 装包的时候建议顺手看一下项目引用到的具体版本。因为不同大版本之间的 API 差异不是改名那么简单连事件委托签名都不一样编译期就是一堆红色报错。这点我在最后常见问题里还会再说一次。3. 服务端实现把Broker跑起来3.1 最小可用的服务端创建一个控制台项目或者 ASP.NET Core 项目NuGet 安装 MQTTnet 之后服务端代码比想象中简单。一个最小可用的 Broker 只需要这么几行using MQTTnet; using MQTTnet.Server; var factory new MqttFactory(); var server factory.CreateMqttServer(); var options new MqttServerOptionsBuilder() .WithDefaultEndpoint() // 开启默认 TCP 监听 .WithDefaultEndpointPort(1883) // MQTT 默认端口 .Build(); await server.StartAsync(options); Console.WriteLine(MQTT Broker 已启动监听端口 1883); // 阻止程序退出 Console.ReadLine();这段代码的关键点在于 MqttServerOptions。WithDefaultEndpoint() 会开启一个 TCP 监听端点默认绑定所有网卡上的 1883 端口。如果端口被占用StartAsync 会直接抛异常所以启动前先检查下端口占用命令行可以执行 netstat -ano | findstr 1883 看一下。服务端起来之后客户端就能连上来了。不过生产环境不可能允许匿名登录先加一层用户认证。3.2 用户认证与连接校验MQTTnet 的服务端可以通过 ValidatingConnectionAsync 事件做自定义校验。这个事件在客户端发起 CONNECT 报文、正式建立会话之前触发我们可以在回调里校验用户名和密码不通过就直接断开server.ValidatingConnectionAsync e { if (e.UserName ! admin || e.Password ! 123456) { e.ReasonCode MqttConnectReasonCode.BadUserNameOrPassword; Console.WriteLine($非法连接: {e.ClientId}); return Task.CompletedTask; } Console.WriteLine($客户端已通过认证: {e.ClientId}); return Task.CompletedTask; };注意这里是在服务端启动之前注册事件所以代码顺序应该是创建 server → 注册事件 → StartAsync。Validator 回调里的 e.ClientId 是客户端连接时携带的 ClientId之后排查消息是谁发的、谁掉线了主要就靠这个字段。如果你在对接设备厂商的相机有的设备不支持配置用户名密码那至少也要通过 ClientId 白名单的方式限制连接不然任何客户端都可以连上来乱发消息主题一乱整个项目就没法维护了。3.3 监听消息流向与事件为了方便调试服务端通常还会打印日志谁连上来了、谁断开了、消息发到了哪个主题。我一般会把这两个事件也注册上server.ClientConnectedAsync e { Console.WriteLine($客户端已连接: {e.ClientId}); return Task.CompletedTask; }; server.ClientDisconnectedAsync e { Console.WriteLine($客户端已断开: {e.ClientId}); return Task.CompletedTask; }; server.InterceptingPublishAsync e { Console.WriteLine($[{DateTime.Now:HH:mm:ss}] 收到消息 - Topic: {e.ApplicationMessage.Topic}, Payload: {e.ApplicationMessage.ConvertPayloadToString()}); return Task.CompletedTask; };InterceptingPublishAsync 是服务端对消息的“拦截”钩子所有客户端发布过来的消息都会经过这里适合做统一的日志记录、消息内容校验甚至可以在这一步直接丢弃不合规的消息。需要注意在这个回调里不要做耗时太长的操作比如写数据库、调第三方接口因为它是异步串行处理的一旦阻塞整个 Broker 的吞吐量都会受影响。真要落库把消息扔进队列或者 Channel让后台消费者慢慢处理。到这里一个带认证和日志的 MQTT 服务端就能用了。但它还只是最基础形态。如果部署到公网建议开启 TLS如果要支撑大规模连接可能需要集群方案或接入 EMQX 这类成熟 Broker。C# 写的 Broker 在中小规模项目、内部系统的场景下完全够用。这也是我个人的使用习惯项目前期用 MQTTnet 内嵌等连接数上来了再评估要不要换独立部署的 Broker。4. 客户端实现发布与订阅4.1 建立连接与断线重连客户端同样用 MQTTnet。连接前需要构造 MqttClientOptions常用的配置有using MQTTnet; using MQTTnet.Client; var factory new MqttFactory(); using var client factory.CreateMqttClient(); var options new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) .WithClientId(device-001) .WithCredentials(admin, 123456) .WithCleanSession() .Build(); await client.ConnectAsync(options, CancellationToken.None);几个关键点ClientId 必须全局唯一。如果两个客户端用了同一个 ClientId后连接的那个会把先连接的那个踢下线这在排查“设备经常掉线”时是首要怀疑对象。WithCleanSession 表示每次连接都重新开始一个干净的会话不保存之前的订阅记录。如果希望离线期间的消息在重连后还能补收到用 WithCleanSession(false) 配合 QoS 1 及以上。网络断开后不会自动重连需要自己处理。在长连接场景里我一般会用一个循环捕获异常后延时几秒重新 ConnectAsync同时把重连次数和日志打出来方便确认是不是网络抖动导致的。4.2 订阅主题订阅主题是收发消息的核心。客户端先注册消息接收回调再 SubscribeAsync 到目标主题client.ApplicationMessageReceivedAsync e { var payload e.ApplicationMessage.ConvertPayloadToString(); Console.WriteLine($收到主题 [{e.ApplicationMessage.Topic}] 的消息: {payload}); return Task.CompletedTask; }; await client.SubscribeAsync( new MqttTopicFilterBuilder() .WithTopic(device//status) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .Build());主题支持通配符 匹配单层比如 device//status 能匹配 device/001/status、device/002/status# 匹配多层比如 device/# 能匹配 device/001/status 以及 device/001/temperature。用通配符可以大幅度减少订阅数量但也容易在设计不善时收到意外的主题消息所以主题层级文档一定要提前定好别让开发者各自起名字。4.3 发布消息发布消息时除了指定主题和内容还需要确认 QoS 级别。示例await client.PublishAsync(new MqttApplicationMessageBuilder() .WithTopic(device/001/command) .WithPayload(open_gate) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .WithRetainFlag(false) .Build());QoS 级别是 MQTT 新手最容易迷糊的地方。我简单总结一下QoS 0最多一次消息发出后不管结果可能丢消息适合传感器周期上报、日志采集这类场景。QoS 1至少一次确保消息到达但可能出现重复需要接收方做幂等处理。QoS 2正好一次最可靠但开销最大适合金额交易、指令下发这类场景。实际开发里设备状态上报用 QoS 0 或者 1 就够了只要接收方对重复消息具备幂等性控制类指令建议至少 QoS 1重要场景用 QoS 2。4.4 遗嘱消息和保留消息怎么配遗嘱消息在连接时配置。客户端还没死Broker 就把它登记在案之后万一客户端网络断开且没有正常发送 DISCONNECT 报文Broker 会替它发布这条遗嘱消息。配置方式var options new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) .WithClientId(device-001) .WithCredentials(admin, 123456) .WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .WithWillTopic(device/001/status) .WithWillPayload(offline) .WithWillRetainFlag(true) .Build();这段配置的含义是当设备 001 异常掉线时Broker 在主题 device/001/status 上发布一条 payload 为 “offline” 的保留消息。如果伴随 WithWillRetainFlag(true)这条 offline 状态会一直被保留直到设备重新上线后发布新的状态覆盖它。保留消息的配置比较简单就是在发布时加 .WithRetainFlag(true)。它用来让新订阅的客户端立即获得主题最后一次的状态很适合设备上下线监控页面。但要注意保留消息不能滥用。如果有大量设备按分钟级发布状态每一条都保留Broker 内存里会堆很多消息建议只对“当前状态”这类主题设置保留对事件类主题不设置。5. 完整示例设备状态上报与远程指令下发这节我把服务端和两个客户端串起来跑一个完整的场景一台服务端 Broker一个设备端模拟上报状态、接收指令一个控制端订阅所有设备状态、下发开闸指令。这里的逻辑其实就是停车场道闸对接的简化版本道闸控制器作为设备端岗亭软件作为控制端。完整代码如下。服务端代码和上面类似这里只展示设备端和控制端的核心部分。设备端// 设备端上报在线状态并订阅控制指令 var factory new MqttFactory(); using var deviceClient factory.CreateMqttClient(); var willOptions new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) .WithClientId(gate-001) .WithCredentials(admin, 123456) .WithWillTopic(gate/001/status) .WithWillPayload(offline) .WithWillRetainFlag(true) .Build(); await deviceClient.ConnectAsync(willOptions, CancellationToken.None); deviceClient.ApplicationMessageReceivedAsync e { Console.WriteLine($设备端收到指令: {e.ApplicationMessage.ConvertPayloadToString()}); return Task.CompletedTask; }; await deviceClient.SubscribeAsync(gate/001/command, MqttQualityOfServiceLevel.AtLeastOnce); // 定时上报状态 while (true) { await deviceClient.PublishAsync(new MqttApplicationMessageBuilder() .WithTopic(gate/001/status) .WithPayload(online) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .WithRetainFlag(true) .Build()); await Task.Delay(TimeSpan.FromSeconds(3)); }控制端// 控制端订阅所有道闸状态并给指定道闸下发开闸指令 using var controlClient factory.CreateMqttClient(); var options new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) .WithClientId(control-center) .WithCredentials(admin, 123456) .Build(); await controlClient.ConnectAsync(options, CancellationToken.None); controlClient.ApplicationMessageReceivedAsync e { Console.WriteLine($收到道闸状态 [{e.ApplicationMessage.Topic}]: {e.ApplicationMessage.ConvertPayloadToString()}); return Task.CompletedTask; }; await controlClient.SubscribeAsync(gate//status, MqttQualityOfServiceLevel.AtLeastOnce); // 模拟岗亭软件下发开闸指令 await controlClient.PublishAsync(new MqttApplicationMessageBuilder() .WithTopic(gate/001/command) .WithPayload(open) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .Build());这里有个细节值得注意设备端订阅的是精确主题 gate/001/command控制端订阅的是通配符 gate//status。订阅粒度不同目的是把“定向控制”和“全局监控”分开。道闸只关心发给自己的指令控制中心要看到所有道闸的状态。如果架构里还有云平台、运维大屏也只需要新增订阅者不需要动设备端代码这就是发布/订阅模型的好处。Payload 格式在实际项目中不要用纯字符串我建议统一用 JSON比如 {cmd:open,ts:2025-01-01T12:00:00} 或 {status:online,uptime:86400}。这样后续加字段不用改协议解析起来也方便。C# 端用 System.Text.Json 反序列化即可。6. 常见问题与排查技巧实录我在实际项目里踩过的坑大多数都是下面这几个类型。整理成一个速查表问题现象常见原因解决办法客户端连不上服务端端口没放行、服务端没启动、防火墙拦截先在本机 telnet 127.0.0.1 1883 测试连通性再检查云服务器安全组和防火墙连接被拒绝返回 4/5 错误码ClientId 冲突或用户名密码错误换一个全局唯一的 ClientId确认认证配置能连接但收不到消息订阅主题和发布主题不一致、通配符错误把主题打印出来对比确认订阅用 还是 #设备经常掉线又自动连上ClientId 冲突多个客户端共用同一个 ID每个客户端生成唯一 ID避免硬编码重连后拿不到离线消息CleanSession 为 true 且 QoS 为 0用 CleanSession(false)QoS 设为 1 及以上设备异常断开但大屏没感知没有配置遗嘱消息连接时配置 WithWillTopic 和 WithWillPayload新页面打开后看不到设备最新状态状态主题没有使用保留消息发布状态时设置 WithRetainFlag(true)除了表格里的问题还有三个容易踩但不太好查的坑单独说一下。第一个是线程问题。在 WinForms 或 WPF 上位机里使用 MQTT 客户端时ApplicationMessageReceivedAsync 回调运行在后台线程直接在其中更新 UI 控件会抛“跨线程操作”异常。解决办法是在回调里用 Control.BeginInvoke 或 SynchronizationContext.Post 回到 UI 线程再更新界面。第二个是消息量和回调阻塞问题。如果服务端在 InterceptingPublishAsync 回调里直接做同步 IO整个 Broker 会越来越慢客户端表现为发布响应时间变长。正确做法是回调里只做轻量处理重活扔进后台队列。第三个是版本 API 差异。MQTTnet 3.x 和 4.x 的事件签名不一样网上很多老示例直接抄过来编译不过。建议参考官方 GitHub 仓库的 examples 目录或者 NuGet 包对应的 XML 文档。如果项目已经依赖了 3.x不要盲目升级到 4.xAPI 迁移成本比想象中高。最后再分享一个我认为很实用的排查思路遇到消息收发异常时先开一个 MQTT 客户端订阅 # 主题把服务端到客户端的所有消息都打出来再结合服务端的 InterceptingPublishAsync 日志基本能定位是发布端没发、Broker 没转还是订阅端没收到。这个办法比对着代码猜快得多。我自己的体会是MQTT 加上 C# 这套组合在中小规模的设备接入和上位机场景里性价比是真的高。服务端能内嵌在业务系统里客户端在 WinForms、控制台、Linux 服务里都能跑调试也方便。如果以后连接数上了几千甚至更多再往 EMQX 这类独立 Broker 迁移也不难因为 MQTT 协议本身是统一的业务代码改动很小。如果你准备在新项目里用这套方案我建议先把主题规范文档写好比如 “设备类型/设备ID/数据类型” 这种结构然后 QoS 级别、保留消息和遗嘱消息的配合早点在架构阶段定下来。等代码写完了再回头补这套约定改起来会非常痛苦。先跑通上面这个完整示例再往业务里加东西是最省力的路径。本文还有配套的精品资源点击获取