简介MQTT客户端C#版是一份面向物联网开发者与C#学习者的实战项目源码基于M2Mqtt.Net.dll实现MQTT协议通信可用于智能家居、工业自动化、远程监测等场景的上位机开发。资源包共35个文件约223KB以cs源码、dll类库、resx资源、exe可执行文件及config配置为主另含sln解决方案与csproj工程文件结构完整便于直接编译运行与二次开发。项目涵盖连接管理、主题订阅、消息发布、心跳保活与异常重连等核心逻辑并配有WinForm界面设计文件可帮助读者理解C#如何与MQTT库集成、构建图形化上位机及处理MQTT事件。目前已有4171人学习下载适合希望掌握MQTT协议原理与C#客户端实现方法的开发者参考借鉴。1. 从一次设备离线告警说起MQTT 客户端 C# 版到底能干什么产线上十几台工控机跑着数据采集上位机需要实时拿到每台设备的温度、转速和报警状态。最初用 HTTP 轮询2 秒一次设备一多服务端就喘网络抖一下还容易丢状态。后来换成 MQTT发布订阅模型一上带宽和实时性都顺了但新的问题来了——服务端是 C# 写的得有一个能长期稳定跑在 Windows 服务里的 MQTT 客户端。这就是这份「MQTT 客户端 C# 版」资源要解决的事它不是一个玩具 Demo而是一个能订阅、发布、断线重连、处理 QoS 的客户端实现适合做工业上位机、物联网网关、数据采集后台的 .NET 开发者。MQTT 本身是轻量级的消息传输协议基于 TCP走发布/订阅模式核心概念就几个Broker消息代理、Topic主题、QoS服务质量等级、Client ID客户端标识。C# 这边主流用 MQTTnet 这个库跨平台、异步、支持 MQTT 3.1.1 和 5.0。这份资源的价值在于把「连上、订阅、收消息、发消息、断线重连」这条链路用可运行的代码串起来而不是让你对着官方文档一行行猜。下面从环境搭建讲到参数配置再到实际会翻车的地方照着走能少踩不少坑。2. 环境准备与 MQTTnet 选型为什么不是别的库2.1 三个 C# MQTT 库的取舍C# 生态里能用的 MQTT 客户端库不多常见的有 MQTTnet、M2MqttGnatMQ 系、以及一些封装了 Paho 的绑定。选型时我一般看四点是否支持 MQTT 5.0、异步 API 是否干净、断线重连是否内置、社区是否还在维护。库MQTT 5.0异步支持断线重连维护状态适用场景MQTTnet支持原生 async/await需自己写活跃新项目首选M2Mqtt不支持回调为主部分基本停更老项目维护Paho 绑定部分一般需封装一般跨语言统一结论很直接新项目用 MQTTnet。它的 API 设计贴近 .NET 习惯MqttFactory创建客户端ConnectAsync、SubscribeAsync、PublishAsync都是异步的配合CancellationToken能干净地退出。M2Mqtt 那套事件回调在 .NET Core 里用起来别扭而且不支持 5.0 的新特性。2.2 建项目、装包、确认版本先建一个 .NET 控制台或 Worker Service 项目。工业场景我倾向 Worker Service因为它天生适合跑后台长驻任务。# 建一个 Worker Service 项目适合做后台常驻客户端 dotnet new worker -n MqttClientDemo cd MqttClientDemo # 安装 MQTTnet注意版本4.x 和 3.x 的 API 差异较大 dotnet add package MQTTnet --version 4.3.7.1207装完确认一下csproj里的引用别装成 3.x 了3.x 的MqttClientOptionsBuilder部分方法签名和 4.x 不一样照着 4.x 的代码写会编译不过。这是第一个容易翻车的地方。提示MQTTnet 4.x 把很多扩展方法挪到了MQTTnet.Extensions命名空间下WithTcpServer、WithCredentials这些如果找不到先检查 using 是否齐全。2.3 Broker 从哪来客户端要连一个 Broker。测试阶段常见做法是本地起一个用 EMQX 或 Mosquitto 都行Docker 一条命令就能跑起来# 本地起一个 Mosquitto映射 1883 端口适合开发调试 docker run -d --name mosquitto -p 1883:1883 eclipse-mosquitto:2生产环境 Broker 通常是独立部署的客户端只需要拿到地址、端口、用户名密码。这里要注意Broker 的allow_anonymous配置如果开着本地测试能连上一上生产关了匿名就连接失败别到那时候才查。3. 连接、订阅、发布把核心链路跑通3.1 构建客户端与连接参数连接是整条链路的地基参数设错后面全白搭。下面这段是连接的核心代码我把它拆开讲。using MQTTnet; using MQTTnet.Client; // 用工厂创建客户端实例一个客户端对应一个连接 var factory new MqttFactory(); var client factory.CreateMqttClient(); // 构建连接选项这里每一项都影响连接行为 var options new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) // Broker 地址和端口 .WithClientId(gateway-001) // 客户端唯一标识重复会顶掉前一个 .WithCredentials(user, pass) // 用户名密码匿名 Broker 可省略 .WithCleanSession(true) // 是否清除会话见下方说明 .WithKeepAlivePeriod(TimeSpan.FromSeconds(30)) // 心跳间隔 .Build(); // 连接失败会抛异常生产环境要 try-catch var result await client.ConnectAsync(options, CancellationToken.None); Console.WriteLine($连接结果: {result.ResultCode});WithClientId是重点。同一个 Client ID 同时连两次Broker 会把前一个踢掉现象就是「客户端莫名其妙掉线」。工业场景里每台设备用固定且唯一的 ID比如用设备序列号拼出来。WithCleanSession(true)表示每次连接都是干净会话Broker 不保留订阅关系和离线消息如果要收离线消息得设成false并且用固定的 Client ID否则会话对不上。KeepAlivePeriod是心跳设太短网络开销大设太长 Broker 可能判定你掉线30 秒是个稳妥值。3.2 订阅主题与通配符订阅要理解 Topic 层级和通配符。匹配单层#匹配多层。比如factory/line1//temp能匹配factory/line1/machineA/temp但匹配不了factory/line1/machineA/room1/temp。// 订阅一个带通配符的主题QoS 用 AtLeastOnce var subOptions new MqttClientSubscribeOptionsBuilder() .WithTopicFilter(f f .WithTopic(factory/line1//status) // 匹配单层 .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce)) .Build(); await client.SubscribeAsync(subOptions, CancellationToken.None);订阅完要挂消息接收事件否则消息来了你收不到// 消息到达时触发e.ApplicationMessage 里是负载 client.ApplicationMessageReceivedAsync e { var topic e.ApplicationMessage.Topic; var payload System.Text.Encoding.UTF8.GetString( e.ApplicationMessage.PayloadSegment); // 4.x 用 PayloadSegment Console.WriteLine($收到 [{topic}]: {payload}); return Task.CompletedTask; };注意 4.x 里取负载用的是PayloadSegment不是老版本的Payload这个改动坑过不少人编译报错时先看这里。3.3 发布消息与 QoS 选择发布消息时 QoS 决定可靠性。QoS 0 最多一次可能丢QoS 1 至少一次可能重复QoS 2 恰好一次开销最大。工业数据采集里状态上报用 QoS 1 比较平衡控制指令如果要求不重复可以用 QoS 2但别滥用QoS 2 的握手流程会拖慢吞吐。// 构建并发布一条消息 var message new MqttApplicationMessageBuilder() .WithTopic(factory/line1/machineA/status) .WithPayload({\temp\:36.5,\rpm\:1200}) // 负载建议用 JSON .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce) .WithRetainFlag(false) // 是否保留最后一条消息 .Build(); await client.PublishAsync(message, CancellationToken.None);WithRetainFlag值得单独说。设成 trueBroker 会保留这条消息新订阅者一连上立刻收到最后一条。适合「设备当前状态」这种场景不适合「一次报警事件」否则新订阅者会收到一条早就过期的报警。3.4 断线重连别指望库帮你全搞定MQTTnet 不会自动重连得自己监听DisconnectedAsync事件然后重连。这是生产环境必须写的部分。// 断线事件里做重连加退避避免疯狂重试 client.DisconnectedAsync async e { Console.WriteLine($断开: {e.Reason}); // 简单退避实际项目建议指数退避 await Task.Delay(TimeSpan.FromSeconds(5)); try { await client.ConnectAsync(options, CancellationToken.None); Console.WriteLine(重连成功); } catch (Exception ex) { Console.WriteLine($重连失败: {ex.Message}); } };这段代码有个隐患如果重连也失败不会再次触发重连客户端就永久掉线了。稳妥做法是套一个循环加指数退避或者用while直到连上为止。这是新手最容易忽略的地方测试时拔网线再插上往往就发现客户端再也连不回来了。4. 避坑与排查那些让你加班到半夜的问题4.1 现象客户端频繁掉线日志里全是重连原因通常有三个。一是 Client ID 重复两个进程用了同一个 ID互相顶。二是 KeepAlive 设得太短网络稍有延迟 Broker 就判定超时。三是 Broker 端有max_keepalive限制客户端设的值超过了它。排查时先看 Broker 日志它会记录踢掉客户端的原因再确认 Client ID 是否唯一最后把 KeepAlive 调到 60 秒试试。4.2 现象订阅了但收不到消息先确认 Topic 拼写和通配符层级对不对factory//status和factory/line1/status是两回事。再确认发布方用的 Topic 是否真的匹配。还有一个隐蔽原因订阅时 QoS 和发布时 QoS 不一致某些 Broker 在特定配置下会降级投递。排查手段是用一个通用的#订阅所有主题看消息到底有没有到 Broker能到就是订阅过滤的问题不能到就是发布端的问题。4.3 现象消息重复收到QoS 1 的语义就是「至少一次」重复是正常的不是 bug。解决方式是在应用层做幂等比如消息里带一个唯一 ID收到后去重。别指望把 QoS 改成 2 就万事大吉QoS 2 开销大而且实现不当一样可能出问题。工业场景里我一般用 QoS 1 加业务层去重比 QoS 2 更可控。4.4 现象程序退出时连接没断干净client.DisconnectAsync()没调或者调了没 await。进程退出时 TCP 连接可能还挂着Broker 要等 KeepAlive 超时才清理。正确做法是在StopAsync或退出钩子里 await 断开并且给一个超时别让断开卡住整个退出流程。// 优雅断开带超时保护 using var cts new CancellationTokenSource(TimeSpan.FromSeconds(3)); await client.DisconnectAsync(new MqttClientDisconnectOptions(), cts.Token);4.5 现象大负载消息导致内存飙升MQTT 单条消息默认有大小限制Broker 和客户端都有。发大 JSON 或二进制时如果超过限制会被断开。排查时看 Broker 的message_size_limit配置客户端这边也要注意别把整个文件塞进一条消息。常见做法是大数据分片或者只发一个引用地址让接收方自己去拉。5. 进阶把客户端做成能长期跑的服务5.1 用 Worker Service 托管生命周期控制台程序一关就没了生产环境得用 Worker Service 或 Windows 服务。把 MQTT 客户端放进BackgroundService的ExecuteAsync里配合IHostApplicationLifetime处理退出。public class MqttWorker : BackgroundService { private readonly IMqttClient _client; private readonly MqttClientOptions _options; public MqttWorker(IMqttClient client, MqttClientOptions options) { _client client; _options options; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { // 挂事件、连接、订阅都在这里 _client.ApplicationMessageReceivedAsync OnMessageAsync; await _client.ConnectAsync(_options, stoppingToken); // 阻塞直到服务停止 await Task.Delay(Timeout.Infinite, stoppingToken); } private Task OnMessageAsync(MqttApplicationMessageReceivedEventArgs e) { // 处理消息注意别在这里做耗时操作会阻塞接收 return Task.CompletedTask; } }关键点OnMessageAsync里别做耗时操作MQTTnet 的消息处理是串行的你在这里卡住后面的消息全堵着。要处理耗时逻辑就丢到队列里用另一个线程消费。5.2 消息处理与背压高吞吐场景下消息来得比处理快内存会涨。常见做法是引入Channel做缓冲接收事件只负责往 Channel 里写后台任务慢慢消费。// 有界 Channel满了就丢或阻塞防止内存无限增长 private readonly ChannelMqttApplicationMessage _channel Channel.CreateBoundedMqttApplicationMessage(1000); // 接收事件里只写 Channel不做业务 private async Task OnMessageAsync(MqttApplicationMessageReceivedEventArgs e) { await _channel.Writer.WriteAsync(e.ApplicationMessage); }CreateBounded的容量根据业务定1000 是个起点。满了之后WriteAsync会等待等于给上游一个背压信号。如果不想等可以用TryWrite满了就丢但要记录丢弃数量否则数据丢了都不知道。5.3 验证客户端是否真的稳写完别急着上线做几个验证。拔网线 30 秒再插上看是否自动重连并恢复订阅。用mosquitto_pub手动发消息确认能收到。把 Broker 重启看客户端行为。压测时用脚本每秒发几百条观察内存和 CPU。我一般还会在客户端里加一个心跳主题定期发布自己的存活状态这样监控端能一眼看出哪个客户端掉线了。从那以后我每次写 MQTT 客户端都会先把断线重连和优雅退出这两段代码写死再动业务逻辑——这两个地方翻车后面查起来就是黑匣子。希望帮到你。本文还有配套的精品资源点击获取 MQTT客户端C#版实战：从连接到断线重连的完整指南
简介MQTT客户端C#版是一份面向物联网开发者与C#学习者的实战项目源码基于M2Mqtt.Net.dll实现MQTT协议通信可用于智能家居、工业自动化、远程监测等场景的上位机开发。资源包共35个文件约223KB以cs源码、dll类库、resx资源、exe可执行文件及config配置为主另含sln解决方案与csproj工程文件结构完整便于直接编译运行与二次开发。项目涵盖连接管理、主题订阅、消息发布、心跳保活与异常重连等核心逻辑并配有WinForm界面设计文件可帮助读者理解C#如何与MQTT库集成、构建图形化上位机及处理MQTT事件。目前已有4171人学习下载适合希望掌握MQTT协议原理与C#客户端实现方法的开发者参考借鉴。1. 从一次设备离线告警说起MQTT 客户端 C# 版到底能干什么产线上十几台工控机跑着数据采集上位机需要实时拿到每台设备的温度、转速和报警状态。最初用 HTTP 轮询2 秒一次设备一多服务端就喘网络抖一下还容易丢状态。后来换成 MQTT发布订阅模型一上带宽和实时性都顺了但新的问题来了——服务端是 C# 写的得有一个能长期稳定跑在 Windows 服务里的 MQTT 客户端。这就是这份「MQTT 客户端 C# 版」资源要解决的事它不是一个玩具 Demo而是一个能订阅、发布、断线重连、处理 QoS 的客户端实现适合做工业上位机、物联网网关、数据采集后台的 .NET 开发者。MQTT 本身是轻量级的消息传输协议基于 TCP走发布/订阅模式核心概念就几个Broker消息代理、Topic主题、QoS服务质量等级、Client ID客户端标识。C# 这边主流用 MQTTnet 这个库跨平台、异步、支持 MQTT 3.1.1 和 5.0。这份资源的价值在于把「连上、订阅、收消息、发消息、断线重连」这条链路用可运行的代码串起来而不是让你对着官方文档一行行猜。下面从环境搭建讲到参数配置再到实际会翻车的地方照着走能少踩不少坑。2. 环境准备与 MQTTnet 选型为什么不是别的库2.1 三个 C# MQTT 库的取舍C# 生态里能用的 MQTT 客户端库不多常见的有 MQTTnet、M2MqttGnatMQ 系、以及一些封装了 Paho 的绑定。选型时我一般看四点是否支持 MQTT 5.0、异步 API 是否干净、断线重连是否内置、社区是否还在维护。库MQTT 5.0异步支持断线重连维护状态适用场景MQTTnet支持原生 async/await需自己写活跃新项目首选M2Mqtt不支持回调为主部分基本停更老项目维护Paho 绑定部分一般需封装一般跨语言统一结论很直接新项目用 MQTTnet。它的 API 设计贴近 .NET 习惯MqttFactory创建客户端ConnectAsync、SubscribeAsync、PublishAsync都是异步的配合CancellationToken能干净地退出。M2Mqtt 那套事件回调在 .NET Core 里用起来别扭而且不支持 5.0 的新特性。2.2 建项目、装包、确认版本先建一个 .NET 控制台或 Worker Service 项目。工业场景我倾向 Worker Service因为它天生适合跑后台长驻任务。# 建一个 Worker Service 项目适合做后台常驻客户端 dotnet new worker -n MqttClientDemo cd MqttClientDemo # 安装 MQTTnet注意版本4.x 和 3.x 的 API 差异较大 dotnet add package MQTTnet --version 4.3.7.1207装完确认一下csproj里的引用别装成 3.x 了3.x 的MqttClientOptionsBuilder部分方法签名和 4.x 不一样照着 4.x 的代码写会编译不过。这是第一个容易翻车的地方。提示MQTTnet 4.x 把很多扩展方法挪到了MQTTnet.Extensions命名空间下WithTcpServer、WithCredentials这些如果找不到先检查 using 是否齐全。2.3 Broker 从哪来客户端要连一个 Broker。测试阶段常见做法是本地起一个用 EMQX 或 Mosquitto 都行Docker 一条命令就能跑起来# 本地起一个 Mosquitto映射 1883 端口适合开发调试 docker run -d --name mosquitto -p 1883:1883 eclipse-mosquitto:2生产环境 Broker 通常是独立部署的客户端只需要拿到地址、端口、用户名密码。这里要注意Broker 的allow_anonymous配置如果开着本地测试能连上一上生产关了匿名就连接失败别到那时候才查。3. 连接、订阅、发布把核心链路跑通3.1 构建客户端与连接参数连接是整条链路的地基参数设错后面全白搭。下面这段是连接的核心代码我把它拆开讲。using MQTTnet; using MQTTnet.Client; // 用工厂创建客户端实例一个客户端对应一个连接 var factory new MqttFactory(); var client factory.CreateMqttClient(); // 构建连接选项这里每一项都影响连接行为 var options new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) // Broker 地址和端口 .WithClientId(gateway-001) // 客户端唯一标识重复会顶掉前一个 .WithCredentials(user, pass) // 用户名密码匿名 Broker 可省略 .WithCleanSession(true) // 是否清除会话见下方说明 .WithKeepAlivePeriod(TimeSpan.FromSeconds(30)) // 心跳间隔 .Build(); // 连接失败会抛异常生产环境要 try-catch var result await client.ConnectAsync(options, CancellationToken.None); Console.WriteLine($连接结果: {result.ResultCode});WithClientId是重点。同一个 Client ID 同时连两次Broker 会把前一个踢掉现象就是「客户端莫名其妙掉线」。工业场景里每台设备用固定且唯一的 ID比如用设备序列号拼出来。WithCleanSession(true)表示每次连接都是干净会话Broker 不保留订阅关系和离线消息如果要收离线消息得设成false并且用固定的 Client ID否则会话对不上。KeepAlivePeriod是心跳设太短网络开销大设太长 Broker 可能判定你掉线30 秒是个稳妥值。3.2 订阅主题与通配符订阅要理解 Topic 层级和通配符。匹配单层#匹配多层。比如factory/line1//temp能匹配factory/line1/machineA/temp但匹配不了factory/line1/machineA/room1/temp。// 订阅一个带通配符的主题QoS 用 AtLeastOnce var subOptions new MqttClientSubscribeOptionsBuilder() .WithTopicFilter(f f .WithTopic(factory/line1//status) // 匹配单层 .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce)) .Build(); await client.SubscribeAsync(subOptions, CancellationToken.None);订阅完要挂消息接收事件否则消息来了你收不到// 消息到达时触发e.ApplicationMessage 里是负载 client.ApplicationMessageReceivedAsync e { var topic e.ApplicationMessage.Topic; var payload System.Text.Encoding.UTF8.GetString( e.ApplicationMessage.PayloadSegment); // 4.x 用 PayloadSegment Console.WriteLine($收到 [{topic}]: {payload}); return Task.CompletedTask; };注意 4.x 里取负载用的是PayloadSegment不是老版本的Payload这个改动坑过不少人编译报错时先看这里。3.3 发布消息与 QoS 选择发布消息时 QoS 决定可靠性。QoS 0 最多一次可能丢QoS 1 至少一次可能重复QoS 2 恰好一次开销最大。工业数据采集里状态上报用 QoS 1 比较平衡控制指令如果要求不重复可以用 QoS 2但别滥用QoS 2 的握手流程会拖慢吞吐。// 构建并发布一条消息 var message new MqttApplicationMessageBuilder() .WithTopic(factory/line1/machineA/status) .WithPayload({\temp\:36.5,\rpm\:1200}) // 负载建议用 JSON .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce) .WithRetainFlag(false) // 是否保留最后一条消息 .Build(); await client.PublishAsync(message, CancellationToken.None);WithRetainFlag值得单独说。设成 trueBroker 会保留这条消息新订阅者一连上立刻收到最后一条。适合「设备当前状态」这种场景不适合「一次报警事件」否则新订阅者会收到一条早就过期的报警。3.4 断线重连别指望库帮你全搞定MQTTnet 不会自动重连得自己监听DisconnectedAsync事件然后重连。这是生产环境必须写的部分。// 断线事件里做重连加退避避免疯狂重试 client.DisconnectedAsync async e { Console.WriteLine($断开: {e.Reason}); // 简单退避实际项目建议指数退避 await Task.Delay(TimeSpan.FromSeconds(5)); try { await client.ConnectAsync(options, CancellationToken.None); Console.WriteLine(重连成功); } catch (Exception ex) { Console.WriteLine($重连失败: {ex.Message}); } };这段代码有个隐患如果重连也失败不会再次触发重连客户端就永久掉线了。稳妥做法是套一个循环加指数退避或者用while直到连上为止。这是新手最容易忽略的地方测试时拔网线再插上往往就发现客户端再也连不回来了。4. 避坑与排查那些让你加班到半夜的问题4.1 现象客户端频繁掉线日志里全是重连原因通常有三个。一是 Client ID 重复两个进程用了同一个 ID互相顶。二是 KeepAlive 设得太短网络稍有延迟 Broker 就判定超时。三是 Broker 端有max_keepalive限制客户端设的值超过了它。排查时先看 Broker 日志它会记录踢掉客户端的原因再确认 Client ID 是否唯一最后把 KeepAlive 调到 60 秒试试。4.2 现象订阅了但收不到消息先确认 Topic 拼写和通配符层级对不对factory//status和factory/line1/status是两回事。再确认发布方用的 Topic 是否真的匹配。还有一个隐蔽原因订阅时 QoS 和发布时 QoS 不一致某些 Broker 在特定配置下会降级投递。排查手段是用一个通用的#订阅所有主题看消息到底有没有到 Broker能到就是订阅过滤的问题不能到就是发布端的问题。4.3 现象消息重复收到QoS 1 的语义就是「至少一次」重复是正常的不是 bug。解决方式是在应用层做幂等比如消息里带一个唯一 ID收到后去重。别指望把 QoS 改成 2 就万事大吉QoS 2 开销大而且实现不当一样可能出问题。工业场景里我一般用 QoS 1 加业务层去重比 QoS 2 更可控。4.4 现象程序退出时连接没断干净client.DisconnectAsync()没调或者调了没 await。进程退出时 TCP 连接可能还挂着Broker 要等 KeepAlive 超时才清理。正确做法是在StopAsync或退出钩子里 await 断开并且给一个超时别让断开卡住整个退出流程。// 优雅断开带超时保护 using var cts new CancellationTokenSource(TimeSpan.FromSeconds(3)); await client.DisconnectAsync(new MqttClientDisconnectOptions(), cts.Token);4.5 现象大负载消息导致内存飙升MQTT 单条消息默认有大小限制Broker 和客户端都有。发大 JSON 或二进制时如果超过限制会被断开。排查时看 Broker 的message_size_limit配置客户端这边也要注意别把整个文件塞进一条消息。常见做法是大数据分片或者只发一个引用地址让接收方自己去拉。5. 进阶把客户端做成能长期跑的服务5.1 用 Worker Service 托管生命周期控制台程序一关就没了生产环境得用 Worker Service 或 Windows 服务。把 MQTT 客户端放进BackgroundService的ExecuteAsync里配合IHostApplicationLifetime处理退出。public class MqttWorker : BackgroundService { private readonly IMqttClient _client; private readonly MqttClientOptions _options; public MqttWorker(IMqttClient client, MqttClientOptions options) { _client client; _options options; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { // 挂事件、连接、订阅都在这里 _client.ApplicationMessageReceivedAsync OnMessageAsync; await _client.ConnectAsync(_options, stoppingToken); // 阻塞直到服务停止 await Task.Delay(Timeout.Infinite, stoppingToken); } private Task OnMessageAsync(MqttApplicationMessageReceivedEventArgs e) { // 处理消息注意别在这里做耗时操作会阻塞接收 return Task.CompletedTask; } }关键点OnMessageAsync里别做耗时操作MQTTnet 的消息处理是串行的你在这里卡住后面的消息全堵着。要处理耗时逻辑就丢到队列里用另一个线程消费。5.2 消息处理与背压高吞吐场景下消息来得比处理快内存会涨。常见做法是引入Channel做缓冲接收事件只负责往 Channel 里写后台任务慢慢消费。// 有界 Channel满了就丢或阻塞防止内存无限增长 private readonly ChannelMqttApplicationMessage _channel Channel.CreateBoundedMqttApplicationMessage(1000); // 接收事件里只写 Channel不做业务 private async Task OnMessageAsync(MqttApplicationMessageReceivedEventArgs e) { await _channel.Writer.WriteAsync(e.ApplicationMessage); }CreateBounded的容量根据业务定1000 是个起点。满了之后WriteAsync会等待等于给上游一个背压信号。如果不想等可以用TryWrite满了就丢但要记录丢弃数量否则数据丢了都不知道。5.3 验证客户端是否真的稳写完别急着上线做几个验证。拔网线 30 秒再插上看是否自动重连并恢复订阅。用mosquitto_pub手动发消息确认能收到。把 Broker 重启看客户端行为。压测时用脚本每秒发几百条观察内存和 CPU。我一般还会在客户端里加一个心跳主题定期发布自己的存活状态这样监控端能一眼看出哪个客户端掉线了。从那以后我每次写 MQTT 客户端都会先把断线重连和优雅退出这两段代码写死再动业务逻辑——这两个地方翻车后面查起来就是黑匣子。希望帮到你。本文还有配套的精品资源点击获取