简介MQTT C# demo测试案例是一套基于MQTTnet库的完整示例工程涵盖服务端与客户端两大模块面向C#开发者和物联网初学者帮助理解MQTT发布/订阅模型在实际项目中的落地方式。资源包共2000个文件以XML文档、CS源代码、DLL程序集和项目配置文件为主同时包含测试用例、依赖包及运行所需资源整体压缩后约42.99MB。已有688人学习适合作为入门实践与参考蓝本。案例覆盖服务端启动监听、客户端连接配置、主题订阅、消息发布与回调接收等关键环节并展示QoS不同等级下的消息交互行为。通过运行所提供的demo读者可快速搭建本地Broker验证端到端通信过程并学习如何配置连接参数、处理连接事件以及编写可复用的消息处理逻辑。对于物联网设备接入、远程监控和实时数据推送等场景这套代码提供了清晰的目录结构与可直接扩展的代码模板能显著缩短项目前期调研与原型开发时间。1. 这套MQTT C# Demo到底解决什么问题MQTT这个东西我在项目里用了好几年了。最开始接触是因为一个停车场道闸对接项目需要在现场部署边缘计算盒子跟海康、大华的车牌识别相机做数据交互又要跟云端平台保持长连接。当时调研了一圈HTTP轮询太浪费带宽TCP裸协议又得自己设计报文格式翻来覆去就MQTT最合适。从那以后MQTT就成了我嵌入式设备和云平台通信的首选方案。如果你搜到这个标题大概率也是遇到类似情况要在C#环境里快速验证MQTT通信需要一个能跑起来的demo最好服务端客户端都有可以直接测试。这个需求我太懂了。很多初学者或者半路出家的开发者一上来就去啃MQTT协议规范结果被那些控制报文、QoS级别、遗嘱消息搞得晕头转向。其实最有效的入门方式就是先跑起来一个demo看到消息从客户端A发到Broker再从Broker推给客户端B整个链路通了回头再看协议细节就轻松多了。这套demo代码核心包含三部分MQTT Broker服务端用MQTTnet库实现启动后监听1883端口接受客户端连接和消息转发MQTT Publisher发布端模拟设备上报数据周期性发送JSON格式的遥测数据MQTT Subscriber订阅端订阅指定Topic实时接收并展示消息内容对于刚接触MQTT的C#开发者来说这套demo的价值在于把协议最核心的发布订阅模型完整跑通了还附带服务端实现。这意味着你不需要额外部署Mosquitto或EMQX这类独立Broker直接用Visual Studio跑起来就能全链路测试断网也能用。对于要评估MQTT方案可行性的团队来说这套demo能在半小时内帮你确认QoS等级选择、Topic设计、断线重连机制、遗嘱消息这些关键点。那为什么不直接用现成的开源Broker非要自己写一个C#服务端呢两种场景需求不同。如果你做的是生产环境比如连接数千台设备那肯定要上EMQX、EMQ X或者Mosquitto这类经过大规模验证的Broker。但如果你只是做功能验证、原型开发或者想把MQTT协议原理弄清楚用C#自己写一个Broker反而能让你把协议的每个细节都吃透。而且有些场景比较特殊比如在离线环境下测试或者需要定制Broker行为做特殊业务逻辑自己实现一个还是很有必要的。这套demo的架构不复杂但麻雀虽小五脏俱全。接下来我把整个实现过程拆解开从环境准备到服务端客户端具体代码再到常见问题排查都过一遍。你可以直接照着敲一遍也可以clone下来跑起来再回头研究细节。2. 环境准备与库选型5分钟搭建开发环境2.1 开发环境与依赖安装先交代一下我的开发环境方便你对照。我用的是Visual Studio 2022.NET版本选的是.NET 6.0LTS版本稳定且生命周期长。如果你用的是VS 2019或者更早版本建议至少保证.NET Core 3.1以上因为MQTTnet从4.x版本开始对旧框架的支持有变化。创建项目的时候我建议直接创建控制台应用项目不需要额外引入复杂的UI框架。虽然WinForm或者WPF做界面更方便看效果但对于原理验证阶段控制台程序足够直观而且省事。创建好项目后使用NuGet包管理器安装MQTTnet库Install-Package MQTTnet我写这套demo的时候用的是MQTTnet 4.1.4版本也是目前用得最广泛的稳定版。如果你使用的是5.x版本要注意API有一些变更具体差异后面我会提到。这里多说一句为什么选MQTTnet而不是别的库。.NET生态里MQTT客户端库有好几个比如MQTTnet、M2Mqtt、uPLibrary等。M2Mqtt算是老牌库了但维护不活跃而且API设计偏老旧。MQTTnet是社区最活跃、性能最好的库之一支持.NET Standard 2.0意味着在.NET Framework 4.6.1以上、.NET Core、UWP甚至Xamarin里都能用。它同时包含客户端和服务端的实现正好满足我们一个demo搞定服务端客户端的需求。2.2 项目结构设计整个解决方案我分成了三个项目解耦清晰也方便后续扩展成独立服务MqttDemo/ ├── MqttBroker/ // 服务端Broker控制台程序 ├── MqttPublisher/ // 客户端A模拟设备上报发布端 └── MqttSubscriber/ // 客户端B模拟平台接收订阅端如果你觉得三个项目太繁琐也可以只建两个项目一个Broker、一个ClientClient里同时跑发布和订阅逻辑。但我个人建议分开因为实际生产环境里发布端和订阅端往往是不同的服务分开更贴近真实场景。为什么不把服务端和客户端合并到一个进程里其实有的场景确实可以合并比如本机测试。但我还是推荐拆开原因很简单MQTT本身就是为分布式场景设计的当你把发布端和订阅端分开在不同进程甚至不同机器运行时才能真切体会到解耦这两个字的含义——发布端根本不需要知道订阅端的存在反之亦然。这种架构上的优势只有跑起来分布式才能感受到。2.3 库选型背后的考量我拿MQTTnet 4.x版本举例你后续看版本差异时会更容易理解。MQTTnet 4.x的核心理念是Everything is async。所有API都是异步的没有同步版本。这跟老一代库不太一样刚开始可能觉得麻烦但用久了会发现异步模型在处理并发连接、消息吞吐时优势明显。另外它的Broker端设计成了高度可扩展的你可以通过拦截器Interceptors来实现自定义认证逻辑、消息过滤等高级功能。然后是最纠结的类型问题在MQTTnet 4.x里项目主要分两大块MqttServer负责服务端MqttClient负责客户端。但到了5.x版本引入了一个统一的MqttFactory类来管理所有生命周期API风格更统一。我用4.x是因为网上资料多、踩坑记录全、跟旧项目兼容性好如果你是新项目也可以直接上5.x只是要注意API有调整。3. 服务端实现自己动手写一个Broker3.1 服务端核心代码与说明Broker的本质很简单接收客户端连接、接收订阅请求、接收发布消息、把消息转发给所有匹配的订阅者。MQTTnet把这些都封装好了我们只需要几十行代码就能实现一个可用的Broker。using MQTTnet; using MQTTnet.Server; var mqttFactory new MqttFactory(); var mqttServerOptions new MqttServerOptionsBuilder() .WithDefaultEndpoint() .WithDefaultEndpointPort(1883) .Build(); var mqttServer mqttFactory.CreateMqttServer(mqttServerOptions); // 注册客户端连接事件 mqttServer.ClientConnectedAsync e { Console.WriteLine($客户端连接{e.ClientId}时间{DateTime.Now}); return Task.CompletedTask; }; // 注册客户端断开事件 mqttServer.ClientDisconnectedAsync e { Console.WriteLine($客户端断开{e.ClientId}原因{e.Reason}); return Task.CompletedTask; }; await mqttServer.StartAsync(); Console.WriteLine(MQTT Broker 已启动监听端口1883); Console.ReadLine();这段代码就这么短核心就四步创建工厂、配置选项、创建服务、启动。但这里有几个细节值得注意第一WithDefaultEndpoint()表示同时监听TCP 1883端口和WebSocket 8083端口。如果你只需要TCP可以改成WithDefaultEndpoint()然后只监听TCP。WebSocket端口主要是给Web端连接用的如果前端要用MQTT.js做浏览器客户端就需要WebSocket支持。第二ClientConnectedAsync是异步事件。MQTTnet 4.x中事件处理器的签名是返回Task的这意味着你可以在事件处理器里做异步操作比如向数据库写入一条日志、调用外部API等。这个设计比传统的EventHandler模式灵活得多。3.2 安全认证与消息拦截基础版Broker跑通后下一步肯定要加上安全控制。毕竟生产环境里不可能允许任何人随便连接你的Broker。MQTTnet支持两种认证方式用户名密码认证和客户端证书认证。用户名密码认证最简单在MqttServerOptions里配置var mqttServerOptions new MqttServerOptionsBuilder() .WithDefaultEndpoint() .WithDefaultEndpointPort(1883) .WithConnectionValidator(context { if (context.Username ! admin || context.Password ! 123456) { context.ReasonCode MqttConnectReasonCode.BadUserNameOrPassword; Console.WriteLine($认证失败{context.ClientId}); } else { Console.WriteLine($认证通过{context.ClientId}); } }) .Build();这里WithConnectionValidator的用法尤其值得一提。4.x版本里不管是内置密码校验、自定义认证、IP白名单都是在这个地方做。5.x版本虽然API改了但思路一致都是通过拦截器实现。除了认证还有个很有用的功能是消息拦截。你可以把Broker当成一个中间人对经过的所有消息做过滤、审计甚至篡改。这个功能在调试和合规审计场景下非常好用mqttServer.InterceptingPublishAsync e { Console.WriteLine($收到消息Topic{e.ApplicationMessage.Topic}, Payload{e.ApplicationMessage.Payload}); return Task.CompletedTask; };3.3 服务端的实际运行效果启动Broker后控制台会输出监听信息和所有接入客户端的连接日志。我用这个Broker连着跑了三天测试稳定性和性能表现都不错。大概测过50个客户端同时连接、每秒500条消息的负载CPU占用率不到10%内存占用不到100MB。这个量级对于原型验证和中小规模项目来说完全够用。不过有一点必须说明白自己写的Broker和生产级Broker在架构上有本质差异。EMQX这类专业Broker支持集群、规则引擎、数据持久化、多协议网关等能力是自己简单的实现完全不能比的。我们的服务端只能算是教学版和内网快速部署版如果要在公网上承载大量设备建议还是用EMQX或Mosquitto来承载。4. 客户端实现发布与订阅双向打通4.1 客户端连接与Topic设计客户端这边我准备了两个程序一个发布一个订阅模拟设备上报和平台接收的完整链路。先看订阅端代码它的核心逻辑是连接Broker、订阅Topic、等待消息、处理消息。using MQTTnet; using MQTTnet.Client; var mqttFactory new MqttFactory(); var mqttClient mqttFactory.CreateMqttClient(); var mqttClientOptions new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) .WithClientId(subscriber-client) .WithCredentials(admin, 123456) .WithCleanSession(true) .Build(); await mqttClient.ConnectAsync(mqttClientOptions); Console.WriteLine(订阅端已连接); await mqttClient.SubscribeAsync(device/{deviceId}/telemetry); Console.WriteLine(已订阅主题device//telemetry);Topic的设计是MQTT里最有讲究的地方。我见过很多初学者把Topic设计成扁平结构比如temp、humid、status这样在设备多了以后根本没法管理。正确的做法是设计成层级结构类似文件系统的目录device/{deviceId}/telemetry设备遥测数据温度、湿度、电压等device/{deviceId}/status设备上下线状态device/{deviceId}/command下发命令device/{deviceId}/response命令响应其中{deviceId}可以用MQTT的通配符替代。匹配单层#匹配多层。比如订阅端用device//telemetry可以接收所有设备的遥测数据用device/#可以接收某个设备的所有消息。这个特性在监控、日志采集场景下非常灵活。4.2 订阅接收与消息处理MQTTnet接收消息有两种方式事件回调和ApplicationMessageReceivedAsync回调。我用的是后者代码更清晰mqttClient.ApplicationMessageReceivedAsync e { var topic e.ApplicationMessage.Topic; var payload System.Text.Encoding.UTF8.GetString(e.ApplicationMessage.Payload); Console.WriteLine($收到消息Topic{topic}, Payload{payload}, QoS{e.ApplicationMessage.QualityOfServiceLevel}); // 这里可以解析JSON写入数据库或者做业务逻辑 return Task.CompletedTask; };实际项目里收到消息后的处理逻辑往往比较复杂。比如需要把设备数据写入时序数据库、做告警判断、更新设备状态缓存等。所有这些耗时操作不应该阻塞消息接收回调。建议在回调里只做入队操作真正处理逻辑放到后台线程池里执行。我踩过这个坑当时直接把写入数据库的操作放在回调里在设备量上来后出现了消息堆积和回调整常阻塞的问题原因是回调是单线程串行执行的一个消息处理慢了会拖慢所有消息的接收后来改成生产者消费者模型才算解决。4.3 发布端的完整实现发布端逻辑更简单就是定时构造消息然后发布using MQTTnet; using MQTTnet.Client; var clientId $device-{Guid.NewGuid():N}; var mqttFactory new MqttFactory(); var mqttClient mqttFactory.CreateMqttClient(); var options new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) .WithClientId(clientId) .WithCredentials(admin, 123456) .WithWillTopic(device/ clientId /status) .WithWillPayload(offline) .WithWillRetain(true) .Build(); var connectResult await mqttClient.ConnectAsync(options); if (connectResult.ResultCode ! MqttClientConnectResultCode.Success) { Console.WriteLine($连接失败{connectResult.ResultCode}); return; } Console.WriteLine($设备 {clientId} 已连接); // 发送遗嘱消息表示设备上线 await mqttClient.PublishAsync(new MqttApplicationMessageBuilder() .WithTopic($device/{clientId}/status) .WithPayload(online) .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce) .WithRetainFlag(true) // 保留消息 .Build()); // 模拟遥测数据上报 var random new Random(); while (true) { var telemetry new { deviceId clientId, temperature random.Next(20, 35), humidity random.Next(40, 80), timestamp DateTimeOffset.UtcNow.ToUnixTimeSeconds() }; var payload System.Text.Json.JsonSerializer.Serialize(telemetry); await mqttClient.PublishAsync(new MqttApplicationMessageBuilder() .WithTopic($device/{clientId}/telemetry) .WithPayload(payload) .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce) .Build()); Console.WriteLine($已发布遥测数据{payload}); await Task.Delay(3000); // 每3秒上报一次 }这里有三个很重要的设计点需要展开讲。第一个是遗嘱消息Will Message这个小而美的特性解决了一个真实痛点当设备异常断电时怎么让平台及时感知。普通TCP连接断开其实平台也能感知到但通过遗嘱消息可以更优雅地处理状态变更——不是简单断开了而是有语义地通知到对应Topic。上面代码里设备连接时带了遗嘱Topic是device/{id}/status内容是offline。如果设备正常断开代码里调用DisconnectAsync遗嘱消息会被移除但如果设备异常崩溃、断网Broker会替它发布遗嘱消息订阅了这个Topic的平台就能立刻感知设备掉线。我在一个充电桩项目里就是靠这个机制实现了1秒内的设备掉线检测比心跳超时快得多。第二个是保留消息Retain Flag。有了它新的客户端一订阅就能立即收到该Topic上最新的一条消息而不是傻等设备下一次上报。比如新部署的监控平台一启动就能立刻看到所有设备的最新状态不需要等一个上报周期。这个在展示大屏、看板类场景下特别实用不用等数据刷新。第三个是QoS级别选择。我这套demo里用的是QoS 1AtLeastOnce即保证消息至少到达一次可能重复。QoS 0是最多一次适合环境传感器数据丢了就丢了。QoS 2是恰好一次性能开销大适合指令下发场景比如远程控制设备开关。大多数业务场景QoS 1就够了要配合去重逻辑处理可能的重复消息。4.4 完整交互演示与测试跑起来的效果非常直观。启动顺序是先启动Broker再启动订阅端最后启动发布端。Broker那边会依次打出日志MQTT Broker 已启动监听端口1883 客户端连接subscriber-client时间2025/1/12 14:23:01 客户端连接device-3f9a2b1c时间2025/1/12 14:23:05 收到消息Topicdevice-3f9a2b1c/telemetry订阅端会实时打印收到的设备数据我用简单的JSON格式化后展示。你可以清楚地看到一条消息从发布端发出、经过Broker中转、最终到达订阅端的全过程。这时再配合Wireshark抓包看MQTT的报文结构整个协议就彻底明白了。5. 常见问题排查与避坑指南5.1 连接失败80%是地址或防火墙问题本地测试最常见的问题是客户端连接不到Broker。排查思路按顺序来先用telnet 127.0.0.1 1883确认端口通不通。如果不通大概率是Broker没启动或者端口配错。如果本机通但局域网其他机器连不上检查Windows防火墙有没有放行1883端口。命令行执行netsh advfirewall firewall add rule nameMQTT dirin actionallow protocolTCP localport1883还有一个隐蔽的坑如果把Broker部署在云服务器上除了系统防火墙还要在云控制台安全组里放行端口。我在腾讯云上就踩过这个坑系统防火墙放行了但安全组没放行客户端一直连接超时。如果网络通但连接还是失败关掉Broker的认证试一下。有些时候问题出在用户名密码不匹配。MQTTnet的客户端日志不显示具体认证失败原因只返回一个BadUserNameOrPassword排查起来确实费时间但逻辑上就是这两端配置没对齐。5.2 消息收不到通配符和QoS的坑订阅成功了发布也成功了但订阅端就是收不到消息这个问题我见过太多次了。排查路径就三条第一Topic是否完全匹配。MQTT的Topic是大小写敏感的Device/Telemetry和device/telemetry是两个完全不同的Topic。发布端和订阅端的Topic字符串要逐字节对比。第二通配符是否用对。device//telemetry只能匹配中间一层device/#可以匹配后续所有层级。注意#必须是Topic最后一个字符device/#/telemetry是非法用法。第三QoS是否匹配。发布消息的QoS是发布端定的订阅的QoS是订阅端定的最终消息实际使用的QoS取两者最小值。如果发布端用QoS 0、订阅端用QoS 2实际消息是QoS 0可能丢失。5.3 其他值得注意的坑客户端ID冲突是个高频问题。MQTT协议要求每个客户端的ClientId必须唯一如果两个客户端用相同的ClientId连接同一个Broker旧的连接会被强制断开。我测试的时候经常因为忘记改ClientId导致神秘掉线排查了半天才发现是这个原因。生产环境建议用设备MAC地址、SN号或者UUID做ClientId。我强烈建议在正式环境关闭Broker的匿名访问启用WithConnectionValidator做用户名密码校验。公网上的MQTT Broker默认端口长期暴露扫描机器人几分钟就能发现你然后开始暴力破解或者刷流量。这东西我在生产环境遇到过当时一台阿里云服务器裸奔一天就被刷了几十万条消息。另外提醒一下我的demo里为了演示方便所有代码都写在Program.cs的Main方法里实际项目一定要做分层服务层负责MQTT通信、业务层负责数据处理。我后来在另一个项目里把MQTT代码直接写进了UI按钮事件里结果界面卡死的惨状至今记忆犹新。MQTT消息回调是在后台线程执行的操作UI控件要使用Invoke或BeginInvoke切回UI线程这在WinForm/WPF开发里是个必须过的坎。6. 从Demo到生产的三个升级方向6.1 数据持久化与消息记录Demo版本的Broker收到消息就转发转发完就丢弃。生产环境通常需要把消息内容存下来方便回溯分析。MQTTnet提供了InterceptingPublishAsync事件消息经过Broker时可以同时做持久化。我在一个环境监测项目里就是这么做的把每台设备上报的温湿度数据实时写入InfluxDB时序数据库方便后续做趋势分析和告警。具体实现是在拦截事件里把消息解析成结构化数据然后异步写入数据库。6.2 证书加密与TLS1883端口是明文传输生产环境公网传输必须用TLS加密。MQTTnet支持TLS加密和双向认证。需要生成CA证书、服务器证书和客户端证书配置比明文复杂得多但这是工业场景的刚需。// 服务端启用TLS var options new MqttServerOptionsBuilder() .WithEncryptedEndpoint() .WithEncryptedEndpointPort(8883) .WithEncryptionCertificate(certificate) .Build();6.3 设备接入网关的完善跟真实的设备接入系统相比这套demo更像是一个实验室里的毛坯房。真实设备接入时格式差异、时钟同步、指令队列都是绕不开的问题而MQTT协议本身并不帮你解决这些需要自己在网关层做适配。我在停车场项目里对接海康相机时就是因为相机端的MQTT实现不规范Topic格式和消息格式都跟标准有差异最后只好在网关上做了一层消息转换才搞定。这些升级方向每个都可以单独写一篇了但基本功还是这套demo里跑通的那些核心逻辑连接、发布、订阅、消息流转。把这套链路吃透了后面不管接什么设备、上什么方案心里都有底。我在实际项目里的体会是MQTT真正的威力不是协议本身有多复杂而是它把设备与平台之间的通信关系彻底解耦了。设备只管上报平台只管订阅谁上线谁离线都不影响对方。这种松耦合的架构设计跟面向对象里依赖倒置的思想其实是一脉相承的。把demo跑通之后建议你再花点时间研究一下遗嘱消息、保留消息和Topic通配符在真实场景下的用法这几个特性用好了能解决很多看似棘手的问题。本文还有配套的精品资源点击获取 MQTT C# Demo实战：基于MQTTnet搭建Broker与发布订阅通信
简介MQTT C# demo测试案例是一套基于MQTTnet库的完整示例工程涵盖服务端与客户端两大模块面向C#开发者和物联网初学者帮助理解MQTT发布/订阅模型在实际项目中的落地方式。资源包共2000个文件以XML文档、CS源代码、DLL程序集和项目配置文件为主同时包含测试用例、依赖包及运行所需资源整体压缩后约42.99MB。已有688人学习适合作为入门实践与参考蓝本。案例覆盖服务端启动监听、客户端连接配置、主题订阅、消息发布与回调接收等关键环节并展示QoS不同等级下的消息交互行为。通过运行所提供的demo读者可快速搭建本地Broker验证端到端通信过程并学习如何配置连接参数、处理连接事件以及编写可复用的消息处理逻辑。对于物联网设备接入、远程监控和实时数据推送等场景这套代码提供了清晰的目录结构与可直接扩展的代码模板能显著缩短项目前期调研与原型开发时间。1. 这套MQTT C# Demo到底解决什么问题MQTT这个东西我在项目里用了好几年了。最开始接触是因为一个停车场道闸对接项目需要在现场部署边缘计算盒子跟海康、大华的车牌识别相机做数据交互又要跟云端平台保持长连接。当时调研了一圈HTTP轮询太浪费带宽TCP裸协议又得自己设计报文格式翻来覆去就MQTT最合适。从那以后MQTT就成了我嵌入式设备和云平台通信的首选方案。如果你搜到这个标题大概率也是遇到类似情况要在C#环境里快速验证MQTT通信需要一个能跑起来的demo最好服务端客户端都有可以直接测试。这个需求我太懂了。很多初学者或者半路出家的开发者一上来就去啃MQTT协议规范结果被那些控制报文、QoS级别、遗嘱消息搞得晕头转向。其实最有效的入门方式就是先跑起来一个demo看到消息从客户端A发到Broker再从Broker推给客户端B整个链路通了回头再看协议细节就轻松多了。这套demo代码核心包含三部分MQTT Broker服务端用MQTTnet库实现启动后监听1883端口接受客户端连接和消息转发MQTT Publisher发布端模拟设备上报数据周期性发送JSON格式的遥测数据MQTT Subscriber订阅端订阅指定Topic实时接收并展示消息内容对于刚接触MQTT的C#开发者来说这套demo的价值在于把协议最核心的发布订阅模型完整跑通了还附带服务端实现。这意味着你不需要额外部署Mosquitto或EMQX这类独立Broker直接用Visual Studio跑起来就能全链路测试断网也能用。对于要评估MQTT方案可行性的团队来说这套demo能在半小时内帮你确认QoS等级选择、Topic设计、断线重连机制、遗嘱消息这些关键点。那为什么不直接用现成的开源Broker非要自己写一个C#服务端呢两种场景需求不同。如果你做的是生产环境比如连接数千台设备那肯定要上EMQX、EMQ X或者Mosquitto这类经过大规模验证的Broker。但如果你只是做功能验证、原型开发或者想把MQTT协议原理弄清楚用C#自己写一个Broker反而能让你把协议的每个细节都吃透。而且有些场景比较特殊比如在离线环境下测试或者需要定制Broker行为做特殊业务逻辑自己实现一个还是很有必要的。这套demo的架构不复杂但麻雀虽小五脏俱全。接下来我把整个实现过程拆解开从环境准备到服务端客户端具体代码再到常见问题排查都过一遍。你可以直接照着敲一遍也可以clone下来跑起来再回头研究细节。2. 环境准备与库选型5分钟搭建开发环境2.1 开发环境与依赖安装先交代一下我的开发环境方便你对照。我用的是Visual Studio 2022.NET版本选的是.NET 6.0LTS版本稳定且生命周期长。如果你用的是VS 2019或者更早版本建议至少保证.NET Core 3.1以上因为MQTTnet从4.x版本开始对旧框架的支持有变化。创建项目的时候我建议直接创建控制台应用项目不需要额外引入复杂的UI框架。虽然WinForm或者WPF做界面更方便看效果但对于原理验证阶段控制台程序足够直观而且省事。创建好项目后使用NuGet包管理器安装MQTTnet库Install-Package MQTTnet我写这套demo的时候用的是MQTTnet 4.1.4版本也是目前用得最广泛的稳定版。如果你使用的是5.x版本要注意API有一些变更具体差异后面我会提到。这里多说一句为什么选MQTTnet而不是别的库。.NET生态里MQTT客户端库有好几个比如MQTTnet、M2Mqtt、uPLibrary等。M2Mqtt算是老牌库了但维护不活跃而且API设计偏老旧。MQTTnet是社区最活跃、性能最好的库之一支持.NET Standard 2.0意味着在.NET Framework 4.6.1以上、.NET Core、UWP甚至Xamarin里都能用。它同时包含客户端和服务端的实现正好满足我们一个demo搞定服务端客户端的需求。2.2 项目结构设计整个解决方案我分成了三个项目解耦清晰也方便后续扩展成独立服务MqttDemo/ ├── MqttBroker/ // 服务端Broker控制台程序 ├── MqttPublisher/ // 客户端A模拟设备上报发布端 └── MqttSubscriber/ // 客户端B模拟平台接收订阅端如果你觉得三个项目太繁琐也可以只建两个项目一个Broker、一个ClientClient里同时跑发布和订阅逻辑。但我个人建议分开因为实际生产环境里发布端和订阅端往往是不同的服务分开更贴近真实场景。为什么不把服务端和客户端合并到一个进程里其实有的场景确实可以合并比如本机测试。但我还是推荐拆开原因很简单MQTT本身就是为分布式场景设计的当你把发布端和订阅端分开在不同进程甚至不同机器运行时才能真切体会到解耦这两个字的含义——发布端根本不需要知道订阅端的存在反之亦然。这种架构上的优势只有跑起来分布式才能感受到。2.3 库选型背后的考量我拿MQTTnet 4.x版本举例你后续看版本差异时会更容易理解。MQTTnet 4.x的核心理念是Everything is async。所有API都是异步的没有同步版本。这跟老一代库不太一样刚开始可能觉得麻烦但用久了会发现异步模型在处理并发连接、消息吞吐时优势明显。另外它的Broker端设计成了高度可扩展的你可以通过拦截器Interceptors来实现自定义认证逻辑、消息过滤等高级功能。然后是最纠结的类型问题在MQTTnet 4.x里项目主要分两大块MqttServer负责服务端MqttClient负责客户端。但到了5.x版本引入了一个统一的MqttFactory类来管理所有生命周期API风格更统一。我用4.x是因为网上资料多、踩坑记录全、跟旧项目兼容性好如果你是新项目也可以直接上5.x只是要注意API有调整。3. 服务端实现自己动手写一个Broker3.1 服务端核心代码与说明Broker的本质很简单接收客户端连接、接收订阅请求、接收发布消息、把消息转发给所有匹配的订阅者。MQTTnet把这些都封装好了我们只需要几十行代码就能实现一个可用的Broker。using MQTTnet; using MQTTnet.Server; var mqttFactory new MqttFactory(); var mqttServerOptions new MqttServerOptionsBuilder() .WithDefaultEndpoint() .WithDefaultEndpointPort(1883) .Build(); var mqttServer mqttFactory.CreateMqttServer(mqttServerOptions); // 注册客户端连接事件 mqttServer.ClientConnectedAsync e { Console.WriteLine($客户端连接{e.ClientId}时间{DateTime.Now}); return Task.CompletedTask; }; // 注册客户端断开事件 mqttServer.ClientDisconnectedAsync e { Console.WriteLine($客户端断开{e.ClientId}原因{e.Reason}); return Task.CompletedTask; }; await mqttServer.StartAsync(); Console.WriteLine(MQTT Broker 已启动监听端口1883); Console.ReadLine();这段代码就这么短核心就四步创建工厂、配置选项、创建服务、启动。但这里有几个细节值得注意第一WithDefaultEndpoint()表示同时监听TCP 1883端口和WebSocket 8083端口。如果你只需要TCP可以改成WithDefaultEndpoint()然后只监听TCP。WebSocket端口主要是给Web端连接用的如果前端要用MQTT.js做浏览器客户端就需要WebSocket支持。第二ClientConnectedAsync是异步事件。MQTTnet 4.x中事件处理器的签名是返回Task的这意味着你可以在事件处理器里做异步操作比如向数据库写入一条日志、调用外部API等。这个设计比传统的EventHandler模式灵活得多。3.2 安全认证与消息拦截基础版Broker跑通后下一步肯定要加上安全控制。毕竟生产环境里不可能允许任何人随便连接你的Broker。MQTTnet支持两种认证方式用户名密码认证和客户端证书认证。用户名密码认证最简单在MqttServerOptions里配置var mqttServerOptions new MqttServerOptionsBuilder() .WithDefaultEndpoint() .WithDefaultEndpointPort(1883) .WithConnectionValidator(context { if (context.Username ! admin || context.Password ! 123456) { context.ReasonCode MqttConnectReasonCode.BadUserNameOrPassword; Console.WriteLine($认证失败{context.ClientId}); } else { Console.WriteLine($认证通过{context.ClientId}); } }) .Build();这里WithConnectionValidator的用法尤其值得一提。4.x版本里不管是内置密码校验、自定义认证、IP白名单都是在这个地方做。5.x版本虽然API改了但思路一致都是通过拦截器实现。除了认证还有个很有用的功能是消息拦截。你可以把Broker当成一个中间人对经过的所有消息做过滤、审计甚至篡改。这个功能在调试和合规审计场景下非常好用mqttServer.InterceptingPublishAsync e { Console.WriteLine($收到消息Topic{e.ApplicationMessage.Topic}, Payload{e.ApplicationMessage.Payload}); return Task.CompletedTask; };3.3 服务端的实际运行效果启动Broker后控制台会输出监听信息和所有接入客户端的连接日志。我用这个Broker连着跑了三天测试稳定性和性能表现都不错。大概测过50个客户端同时连接、每秒500条消息的负载CPU占用率不到10%内存占用不到100MB。这个量级对于原型验证和中小规模项目来说完全够用。不过有一点必须说明白自己写的Broker和生产级Broker在架构上有本质差异。EMQX这类专业Broker支持集群、规则引擎、数据持久化、多协议网关等能力是自己简单的实现完全不能比的。我们的服务端只能算是教学版和内网快速部署版如果要在公网上承载大量设备建议还是用EMQX或Mosquitto来承载。4. 客户端实现发布与订阅双向打通4.1 客户端连接与Topic设计客户端这边我准备了两个程序一个发布一个订阅模拟设备上报和平台接收的完整链路。先看订阅端代码它的核心逻辑是连接Broker、订阅Topic、等待消息、处理消息。using MQTTnet; using MQTTnet.Client; var mqttFactory new MqttFactory(); var mqttClient mqttFactory.CreateMqttClient(); var mqttClientOptions new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) .WithClientId(subscriber-client) .WithCredentials(admin, 123456) .WithCleanSession(true) .Build(); await mqttClient.ConnectAsync(mqttClientOptions); Console.WriteLine(订阅端已连接); await mqttClient.SubscribeAsync(device/{deviceId}/telemetry); Console.WriteLine(已订阅主题device//telemetry);Topic的设计是MQTT里最有讲究的地方。我见过很多初学者把Topic设计成扁平结构比如temp、humid、status这样在设备多了以后根本没法管理。正确的做法是设计成层级结构类似文件系统的目录device/{deviceId}/telemetry设备遥测数据温度、湿度、电压等device/{deviceId}/status设备上下线状态device/{deviceId}/command下发命令device/{deviceId}/response命令响应其中{deviceId}可以用MQTT的通配符替代。匹配单层#匹配多层。比如订阅端用device//telemetry可以接收所有设备的遥测数据用device/#可以接收某个设备的所有消息。这个特性在监控、日志采集场景下非常灵活。4.2 订阅接收与消息处理MQTTnet接收消息有两种方式事件回调和ApplicationMessageReceivedAsync回调。我用的是后者代码更清晰mqttClient.ApplicationMessageReceivedAsync e { var topic e.ApplicationMessage.Topic; var payload System.Text.Encoding.UTF8.GetString(e.ApplicationMessage.Payload); Console.WriteLine($收到消息Topic{topic}, Payload{payload}, QoS{e.ApplicationMessage.QualityOfServiceLevel}); // 这里可以解析JSON写入数据库或者做业务逻辑 return Task.CompletedTask; };实际项目里收到消息后的处理逻辑往往比较复杂。比如需要把设备数据写入时序数据库、做告警判断、更新设备状态缓存等。所有这些耗时操作不应该阻塞消息接收回调。建议在回调里只做入队操作真正处理逻辑放到后台线程池里执行。我踩过这个坑当时直接把写入数据库的操作放在回调里在设备量上来后出现了消息堆积和回调整常阻塞的问题原因是回调是单线程串行执行的一个消息处理慢了会拖慢所有消息的接收后来改成生产者消费者模型才算解决。4.3 发布端的完整实现发布端逻辑更简单就是定时构造消息然后发布using MQTTnet; using MQTTnet.Client; var clientId $device-{Guid.NewGuid():N}; var mqttFactory new MqttFactory(); var mqttClient mqttFactory.CreateMqttClient(); var options new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) .WithClientId(clientId) .WithCredentials(admin, 123456) .WithWillTopic(device/ clientId /status) .WithWillPayload(offline) .WithWillRetain(true) .Build(); var connectResult await mqttClient.ConnectAsync(options); if (connectResult.ResultCode ! MqttClientConnectResultCode.Success) { Console.WriteLine($连接失败{connectResult.ResultCode}); return; } Console.WriteLine($设备 {clientId} 已连接); // 发送遗嘱消息表示设备上线 await mqttClient.PublishAsync(new MqttApplicationMessageBuilder() .WithTopic($device/{clientId}/status) .WithPayload(online) .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce) .WithRetainFlag(true) // 保留消息 .Build()); // 模拟遥测数据上报 var random new Random(); while (true) { var telemetry new { deviceId clientId, temperature random.Next(20, 35), humidity random.Next(40, 80), timestamp DateTimeOffset.UtcNow.ToUnixTimeSeconds() }; var payload System.Text.Json.JsonSerializer.Serialize(telemetry); await mqttClient.PublishAsync(new MqttApplicationMessageBuilder() .WithTopic($device/{clientId}/telemetry) .WithPayload(payload) .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce) .Build()); Console.WriteLine($已发布遥测数据{payload}); await Task.Delay(3000); // 每3秒上报一次 }这里有三个很重要的设计点需要展开讲。第一个是遗嘱消息Will Message这个小而美的特性解决了一个真实痛点当设备异常断电时怎么让平台及时感知。普通TCP连接断开其实平台也能感知到但通过遗嘱消息可以更优雅地处理状态变更——不是简单断开了而是有语义地通知到对应Topic。上面代码里设备连接时带了遗嘱Topic是device/{id}/status内容是offline。如果设备正常断开代码里调用DisconnectAsync遗嘱消息会被移除但如果设备异常崩溃、断网Broker会替它发布遗嘱消息订阅了这个Topic的平台就能立刻感知设备掉线。我在一个充电桩项目里就是靠这个机制实现了1秒内的设备掉线检测比心跳超时快得多。第二个是保留消息Retain Flag。有了它新的客户端一订阅就能立即收到该Topic上最新的一条消息而不是傻等设备下一次上报。比如新部署的监控平台一启动就能立刻看到所有设备的最新状态不需要等一个上报周期。这个在展示大屏、看板类场景下特别实用不用等数据刷新。第三个是QoS级别选择。我这套demo里用的是QoS 1AtLeastOnce即保证消息至少到达一次可能重复。QoS 0是最多一次适合环境传感器数据丢了就丢了。QoS 2是恰好一次性能开销大适合指令下发场景比如远程控制设备开关。大多数业务场景QoS 1就够了要配合去重逻辑处理可能的重复消息。4.4 完整交互演示与测试跑起来的效果非常直观。启动顺序是先启动Broker再启动订阅端最后启动发布端。Broker那边会依次打出日志MQTT Broker 已启动监听端口1883 客户端连接subscriber-client时间2025/1/12 14:23:01 客户端连接device-3f9a2b1c时间2025/1/12 14:23:05 收到消息Topicdevice-3f9a2b1c/telemetry订阅端会实时打印收到的设备数据我用简单的JSON格式化后展示。你可以清楚地看到一条消息从发布端发出、经过Broker中转、最终到达订阅端的全过程。这时再配合Wireshark抓包看MQTT的报文结构整个协议就彻底明白了。5. 常见问题排查与避坑指南5.1 连接失败80%是地址或防火墙问题本地测试最常见的问题是客户端连接不到Broker。排查思路按顺序来先用telnet 127.0.0.1 1883确认端口通不通。如果不通大概率是Broker没启动或者端口配错。如果本机通但局域网其他机器连不上检查Windows防火墙有没有放行1883端口。命令行执行netsh advfirewall firewall add rule nameMQTT dirin actionallow protocolTCP localport1883还有一个隐蔽的坑如果把Broker部署在云服务器上除了系统防火墙还要在云控制台安全组里放行端口。我在腾讯云上就踩过这个坑系统防火墙放行了但安全组没放行客户端一直连接超时。如果网络通但连接还是失败关掉Broker的认证试一下。有些时候问题出在用户名密码不匹配。MQTTnet的客户端日志不显示具体认证失败原因只返回一个BadUserNameOrPassword排查起来确实费时间但逻辑上就是这两端配置没对齐。5.2 消息收不到通配符和QoS的坑订阅成功了发布也成功了但订阅端就是收不到消息这个问题我见过太多次了。排查路径就三条第一Topic是否完全匹配。MQTT的Topic是大小写敏感的Device/Telemetry和device/telemetry是两个完全不同的Topic。发布端和订阅端的Topic字符串要逐字节对比。第二通配符是否用对。device//telemetry只能匹配中间一层device/#可以匹配后续所有层级。注意#必须是Topic最后一个字符device/#/telemetry是非法用法。第三QoS是否匹配。发布消息的QoS是发布端定的订阅的QoS是订阅端定的最终消息实际使用的QoS取两者最小值。如果发布端用QoS 0、订阅端用QoS 2实际消息是QoS 0可能丢失。5.3 其他值得注意的坑客户端ID冲突是个高频问题。MQTT协议要求每个客户端的ClientId必须唯一如果两个客户端用相同的ClientId连接同一个Broker旧的连接会被强制断开。我测试的时候经常因为忘记改ClientId导致神秘掉线排查了半天才发现是这个原因。生产环境建议用设备MAC地址、SN号或者UUID做ClientId。我强烈建议在正式环境关闭Broker的匿名访问启用WithConnectionValidator做用户名密码校验。公网上的MQTT Broker默认端口长期暴露扫描机器人几分钟就能发现你然后开始暴力破解或者刷流量。这东西我在生产环境遇到过当时一台阿里云服务器裸奔一天就被刷了几十万条消息。另外提醒一下我的demo里为了演示方便所有代码都写在Program.cs的Main方法里实际项目一定要做分层服务层负责MQTT通信、业务层负责数据处理。我后来在另一个项目里把MQTT代码直接写进了UI按钮事件里结果界面卡死的惨状至今记忆犹新。MQTT消息回调是在后台线程执行的操作UI控件要使用Invoke或BeginInvoke切回UI线程这在WinForm/WPF开发里是个必须过的坎。6. 从Demo到生产的三个升级方向6.1 数据持久化与消息记录Demo版本的Broker收到消息就转发转发完就丢弃。生产环境通常需要把消息内容存下来方便回溯分析。MQTTnet提供了InterceptingPublishAsync事件消息经过Broker时可以同时做持久化。我在一个环境监测项目里就是这么做的把每台设备上报的温湿度数据实时写入InfluxDB时序数据库方便后续做趋势分析和告警。具体实现是在拦截事件里把消息解析成结构化数据然后异步写入数据库。6.2 证书加密与TLS1883端口是明文传输生产环境公网传输必须用TLS加密。MQTTnet支持TLS加密和双向认证。需要生成CA证书、服务器证书和客户端证书配置比明文复杂得多但这是工业场景的刚需。// 服务端启用TLS var options new MqttServerOptionsBuilder() .WithEncryptedEndpoint() .WithEncryptedEndpointPort(8883) .WithEncryptionCertificate(certificate) .Build();6.3 设备接入网关的完善跟真实的设备接入系统相比这套demo更像是一个实验室里的毛坯房。真实设备接入时格式差异、时钟同步、指令队列都是绕不开的问题而MQTT协议本身并不帮你解决这些需要自己在网关层做适配。我在停车场项目里对接海康相机时就是因为相机端的MQTT实现不规范Topic格式和消息格式都跟标准有差异最后只好在网关上做了一层消息转换才搞定。这些升级方向每个都可以单独写一篇了但基本功还是这套demo里跑通的那些核心逻辑连接、发布、订阅、消息流转。把这套链路吃透了后面不管接什么设备、上什么方案心里都有底。我在实际项目里的体会是MQTT真正的威力不是协议本身有多复杂而是它把设备与平台之间的通信关系彻底解耦了。设备只管上报平台只管订阅谁上线谁离线都不影响对方。这种松耦合的架构设计跟面向对象里依赖倒置的思想其实是一脉相承的。把demo跑通之后建议你再花点时间研究一下遗嘱消息、保留消息和Topic通配符在真实场景下的用法这几个特性用好了能解决很多看似棘手的问题。本文还有配套的精品资源点击获取