★ FEATURED ARTICLE 简介这套 C# MQTT 源码面向具备一定 .NET 基础的开发者用于在 .NET 4.5 环境下学习 MQTT 客户端实现与物联网设备通信。内容围绕发布/订阅模式展开涉及连接 broker、主题订阅与消息发布、QoS 0/1/2 服务质量、保留消息、会话持久化及断线重连等要点并包含配套 MQTT 客户端库文件读者可据此梳理协议层与业务层的分工理解连接参数、回调注册、异常重试和心跳保活的组织方式。压缩包共 278 个文件约 2.03MB以 59 个 cs 源码和 9 个 csproj、5 个 sln 工程文件为主体辅以 dll、exe、config 等运行与配置文件以及 png、html、css 等界面和说明素材可较完整地还原示例工程结构。已有 599 人学习下载适合需要参考客户端封装、消息回调、异常处理与重连写法的读者按模块查阅也可作为搭建设备通信 Demo 的对照材料。1. 一份能编译的 C# MQTT 源码值不值得拆第一次做设备接入网关我用的是一份别人给的示例连上 broker、发条消息、收条消息本地跑得挺欢。等切到真实弱网环境加上 QoS2 和断线重连半小时内必崩日志里只有一句「连接已关闭」没有任何线索。后来才发现问题不在业务代码而在我根本不知道客户端到底往 socket 里写了哪些字节。这份 C# MQTT 源码包含完整库文件和 .NET 4.5 工程价值不在于它多能打而在于把 MQTT 3.1.1 的报文编解码、会话状态、心跳保活这些平时被封装藏起来的东西全摊在明面上。它适合两类人一类是被第三方库的行为卡住、想搞清楚底下发了什么的另一类是手上还跑着一批 .NET Framework 4.5 的工控上位机没法随便升到高版本运行时只能找能在老框架里编译通过的实现。2. 工程分层C# MQTT 库的四层骨架与 .NET 4.5 编译链拿到源码先别急着 F5先把目录翻一遍。一个能长期维护的 MQTT 客户端结构上一定是分层的否则协议细节和业务逻辑会搅在一起改一个心跳参数要翻五个文件。2.1 目录拆分传输、编解码、会话、门面各管一段我拆过的这类工程目录大致长这样具体名字可能不同但分层思路基本一致MqttLab/ ├── MqttLab.Core/ │ ├── Protocol/ # 协议层只管字节不碰网络 │ │ ├── MqttPacketType.cs # 报文类型枚举 1~14 │ │ ├── MqttReader.cs # 从流里读基础类型变长整数、UTF-8 字符串 │ │ ├── MqttWriter.cs # 往流里写基础类型 │ │ ├── MqttProtocolException.cs │ │ └── Packets/ │ │ ├── ConnectPacket.cs │ │ ├── ConnAckPacket.cs │ │ ├── PublishPacket.cs │ │ ├── PubAckPacket.cs │ │ ├── SubscribePacket.cs │ │ └── PingPacket.cs │ ├── Transport/ # 传输层TCP / TLS只负责收发字节 │ │ ├── ITransport.cs │ │ ├── TcpTransport.cs │ │ └── TlsTransport.cs │ ├── Session/ # 会话层报文标识符、inflight、主题匹配 │ │ ├── PacketIdAllocator.cs │ │ ├── InflightTracker.cs │ │ └── TopicMatcher.cs │ └── MqttClient.cs # 门面层对外只暴露 Connect/Publish/Subscribe ├── MqttLab.Sample/ │ └── Program.cs # 控制台示例验证用 └── libs/ # 预编译库文件含强签名 dll四层的职责边界很清晰这也是我建议二次开发时守住的第一条线Protocol层不允许引用System.Net.Sockets它接收的是Stream输出的是byte[]这样编解码可以脱离网络单独写单元测试Transport层不认识 MQTT它只知道自己要搬运字节流Session层管状态比如一个 QoS1 消息发出去还没收到 PUBACK它得记着MqttClient是门面把异步方法、事件回调、重连策略串起来。这么拆还有一个实际好处换传输不用动协议。有些项目后期要把 TCP 换成 TLS或者在某些现场要走串口透传只要ITransport接口不变上层一行代码都不用改。// 传输层只暴露最小接口协议层和门面层都只依赖它 public interface ITransport : IDisposable { bool IsConnected { get; } Task ConnectAsync(string host, int port, CancellationToken token); void Write(byte[] buffer, int offset, int count); // 注意返回值读到的字节数可能小于请求数调用方必须自己凑满 int Read(byte[] buffer, int offset, int count); event EventHandlerException Closed; }这里的Read返回int而不是void是刻意的。很多新手写的传输层直接封装成「读满 N 字节」结果上层拿不到「对端提前关闭」这个信号遇到半包就死等。让返回值暴露出来调用方自己决定是重试还是抛异常责任更清楚。Closed事件用于把底层断开往上冒泡门面层收到后触发重连不用轮询IsConnected。2.2 在老框架下把工程编出来目标框架、语言版本与库文件引用.NET Framework 4.5 是个特殊的存在它默认对应的 C# 语言版本是 5.0没有字符串插值、没有?.、没有nameof更没有SpanT。而很多从新版本回移的源码里作者已经用上了 C# 6/7 的语法直接打开会满屏红波浪线。处理办法是在 csproj 里显式抬高语言版本把版本设置交给编译器不改目标框架Project SdkMicrosoft.NET.Sdk PropertyGroup TargetFrameworknet45/TargetFramework !-- 只抬高语法版本运行时依旧是 .NET Framework 4.5 -- LangVersion7.3/LangVersion OutputTypeLibrary/OutputType AssemblyNameMqttLab.Core/AssemblyName RootNamespaceMqttLab.Core/RootNamespace /PropertyGroup ItemGroup !-- 老式库文件用 HintPath 引用别丢到 GAC 里 -- Reference IncludeMqttLab.Legacy HintPath..\libs\MqttLab.Legacy.dll/HintPath Privatetrue/Private /Reference Reference IncludeSystem / Reference IncludeSystem.Core / /ItemGroup /ProjectLangVersion设成 7.3 而不是latest是因为 7.3 之后的语法比如switch表达式、可空引用类型会依赖新运行时的类型4.5 上编不过。Privatetrue保证编译时把 dll 拷到输出目录不然运行时报「找不到程序集」这类问题在部署到工控机上特别常见因为你本地开发机的 GAC 里可能恰好有同名程序集怎么调都是好的一到现场就崩。还有个小细节.NET 4.5 的控制台入口不支持async Task Main那是 C# 7.1 之后才有的。示例工程里要么写static void Main然后.GetAwaiter().GetResult()要么自己搭一个Task.Run(...).Wait()。前者在 UI 线程上调用有死锁风险控制台程序无所谓如果宿主是 WinForms 上位机务必用后者或者在独立线程里跑。提示编译前先确认libs/下的 dll 是不是和源码同一版本。混用过不同版本的编解码器和门面层表现是发出去的 CONNECT 报文少一个字节broker 直接断开且不回 CONNACK排查起来极其费劲。3. 报文编解码从固定头到 PUBLISH 载荷的字节级实现协议层是这份源码里最值得逐行读的部分。MQTT 的报文结构简单得有点朴素固定头 可变头 载荷三段的边界靠「剩余长度」这一个变长整数界定。恰恰是这个变长整数卡住了大半自己动手写客户端的人。3.1 剩余长度一个变长整数卡住了大半新手剩余长度用 1 到 4 个字节表示每个字节只用低 7 位最高位是「还有后续字节」的标志。这个设计的目的是让小报文只占 1 字节同时理论上支持到 256MB。编码时从低位往高位拆解码时反着累加。// 编码剩余长度每字节低 7 位有效最高位标记是否还有后续字节 public static int EncodeRemainingLength(int length, byte[] buffer, int offset) { // 协议规定上限 2684354550xFFFFFF7F超过就是非法报文 if (length 0 || length 268435455) throw new ArgumentOutOfRangeException(nameof(length)); int count 0; do { byte digit (byte)(length % 128); // 取低 7 位 length / 128; // 右移 7 位 if (length 0) digit | 0x80; // 还有后续字节就置最高位 buffer[offset count] digit; count; } while (length 0); return count; // 返回实际写入的字节数1~4 }offset参数让调用方可以复用一个 4 字节的临时缓冲区避免每条消息都 new 一个数组——在高频发布场景下这个分配开销很可观。返回值是写入字节数因为剩余长度不是定长的调用方必须知道后面要跳过几个字节。解码侧更容易出问题// 解码剩余长度最多读 4 字节超过就说明报文非法必须立刻断链 public static int DecodeRemainingLength(Stream stream) { int multiplier 1; int value 0; int bytesRead 0; byte digit; do { int b stream.ReadByte(); if (b 0) throw new EndOfStreamException(连接在报文头中途被关闭); digit (byte)b; value (digit 0x7F) * multiplier; // 累加低 7 位 multiplier * 128; bytesRead; // 关键防护不设上限的话恶意或损坏的报文会让我们死循环读下去 if (bytesRead 4) throw new MqttProtocolException(剩余长度字段超过 4 字节); } while ((digit 0x80) ! 0); return value; }bytesRead 4这个判断千万不能省。我有一次调试时抓到一个现场设备发来的畸形报文最高位一直是 1客户端就卡在ReadByte里不动了表现是「连上了但收不到任何消息」日志一片空白。加上这个防护后异常能正常往上抛重连逻辑才有机会介入。拿到剩余长度后下一步是把整段报文读进缓冲区。这里必须用「读满」的写法不能指望一次Read就拿到全部数据// 按剩余长度精确读出一整条报文读不满就继续读 private byte[] ReadBody(Stream stream, int remainingLength) { byte[] body new byte[remainingLength]; int read 0; while (read remainingLength) { int n stream.Read(body, read, remainingLength - read); if (n 0) throw new EndOfStreamException(报文未读完连接即关闭); read n; } return body; }remainingLength为 0 的报文也是合法的比如 PINGREQ 和 PINGRESP固定头之后什么都没有。如果代码里写成「读到 -1 为止」那遇到 PINGRESP 就会一直等到连接超时表现为心跳永远收不到回应。3.2 CONNECT 标志位与 PUBLISH 组包字段顺序写错就是静默断开CONNECT 是客户端发的第一条报文也是字段顺序被协议写死得最严的一条协议名 → 协议级别 → 连接标志 → 心跳秒数 → 载荷ClientId → 遗嘱主题 → 遗嘱消息 → 用户名 → 密码。这个顺序是协议规定的写反了 broker 不会给你任何错误提示直接关连接。public byte[] Build(MqttConnectOptions opt) { using (var body new MemoryStream()) { WriteString(body, MQTT); // 协议名固定 4 字节 body.WriteByte(4); // 协议级别3.1.1 用 43.1 用 3 byte flags 0; if (opt.CleanSession) flags | 0x02; // bit1清理会话 if (opt.Will ! null) { flags | 0x04; // bit2启用遗嘱 flags | (byte)((int)opt.WillQos 3); // bit3-4遗嘱 QoS if (opt.WillRetain) flags | 0x20; // bit5遗嘱保留 } if (!string.IsNullOrEmpty(opt.Password)) flags | 0x40; // bit6有密码 if (!string.IsNullOrEmpty(opt.UserName)) flags | 0x80; // bit7有用户名 body.WriteByte(flags); // 心跳秒数大端两字节写 0 表示关闭心跳实际上没人这么干 body.WriteByte((byte)(opt.KeepAliveSeconds 8)); body.WriteByte((byte)(opt.KeepAliveSeconds 0xFF)); // 以下顺序被协议写死不能调换 WriteString(body, opt.ClientId); if (opt.Will ! null) { WriteString(body, opt.Will.Topic); WriteBinary(body, opt.Will.Payload); } if (!string.IsNullOrEmpty(opt.UserName)) WriteString(body, opt.UserName); if (!string.IsNullOrEmpty(opt.Password)) WriteString(body, opt.Password); return WrapFixedHeader(MqttPacketType.Connect, 0, body.ToArray()); } }flags的第 0 位是保留位协议要求必须为 0有些 broker 会校验这一位置 1 就断开。WriteString也不是随便写的MQTT 的字符串是「2 字节大端长度 UTF-8 内容」和 .NET 里BinaryWriter.Write(string)默认的 7 位变长前缀完全是两回事// MQTT 字符串 2 字节大端长度 UTF-8 内容 public static void WriteString(Stream stream, string value) { byte[] raw Encoding.UTF8.GetBytes(value ?? string.Empty); if (raw.Length 65535) throw new MqttProtocolException(字符串长度超过 65535 字节); stream.WriteByte((byte)(raw.Length 8)); // 高字节在前 stream.WriteByte((byte)(raw.Length 0xFF)); stream.Write(raw, 0, raw.Length); }最后套上固定头。固定头的第一个字节是高 4 位的报文类型加低 4 位的标志紧随其后的是剩余长度private static byte[] WrapFixedHeader(MqttPacketType type, byte flags, byte[] body) { byte[] lenBuf new byte[4]; int lenBytes EncodeRemainingLength(body.Length, lenBuf, 0); byte[] packet new byte[1 lenBytes body.Length]; packet[0] (byte)(((int)type 4) | (flags 0x0F)); // 高 4 位类型低 4 位标志 Buffer.BlockCopy(lenBuf, 0, packet, 1, lenBytes); Buffer.BlockCopy(body, 0, packet, 1 lenBytes, body.Length); return packet; }PUBLISH 的标志位含义和别的报文不一样低 4 位在这里分别承载 DUP、QoS、Retainbit3 是重发标志bit1-2 是 QoS 等级bit0 是保留消息标志。而且只有 QoS 大于 0 的时候可变头里才带 2 字节的报文标识符。public byte[] BuildPublish(string topic, byte[] payload, int qos, bool retain, ushort packetId) { using (var body new MemoryStream()) { WriteString(body, topic); // 主题名不能含通配符 if (qos 0) { body.WriteByte((byte)(packetId 8)); // QoS0 才带报文标识符 body.WriteByte((byte)(packetId 0xFF)); } body.Write(payload, 0, payload.Length); byte flags (byte)(qos 1); // bit1-2 是 QoS if (retain) flags | 0x01; // bit0 是 Retain return WrapFixedHeader(MqttPacketType.Publish, flags, body.ToArray()); } }packetId由会话层的分配器统一发号不能在业务代码里随手new Random().Next(1, 65535)撞号会导致 broker 把两条不同的消息当成同一条去重直接丢消息。这个坑我在一个批量下发配置的场景里踩过表现是「偶尔少发一条」概率极低压测几千次才复现一次。4. 连接生命周期心跳保活、会话状态与断线重连协议层只管一条报文怎么拼。真正让客户端在现场活下来的是连接生命周期的管理——心跳、状态机、重连、重投这四件事任何一件没做好设备就会变成「上线五分钟掉线两小时」。4.1 Keep Alive 定时器与四态连接机CONNECT 里那个KeepAliveSeconds是客户端对 broker 的承诺我保证在这个时间内至少发一条报文包括 PINGREQ。broker 侧通常在 1.5 倍这个时间没收到东西就主动断开。所以客户端必须自己维护一个定时器空闲到一定程度就发心跳。private readonly object _sync new object(); private DateTime _lastSend DateTime.UtcNow; private int _pingOutstanding; // 已发未应答的 PINGREQ 数量 private int _keepAliveSeconds; private Timer _keepAliveTimer; private void StartKeepAlive(int keepAliveSeconds) { if (keepAliveSeconds 0) return; _keepAliveSeconds keepAliveSeconds; // 按半个心跳周期检查一次给自己留出重传和抖动窗口 int period Math.Max(1000, keepAliveSeconds * 1000 / 2); _keepAliveTimer new Timer(CheckKeepAlive, null, period, period); } private void CheckKeepAlive(object state) { if (_state ! MqttConnectionState.Connected) return; bool idle; lock (_sync) { // 用 0.8 倍而不是 1.0 倍抵消定时器漂移和网络抖动 idle (DateTime.UtcNow - _lastSend).TotalSeconds _keepAliveSeconds * 0.8; } if (!idle) return; // 连续两个周期没有任何回包别再等了大概率是半开连接 if (Interlocked.Increment(ref _pingOutstanding) 2) { ForceClose(new TimeoutException(心跳超时链路判定为死链)); return; } SendRaw(PingPacket.Build()); // 固定头 0xC0 0x00无载荷 }_lastSend在每次Write成功后更新不是每次发 PINGREQ 时更新这样业务消息本身就能充当心跳不必额外发探测包。_pingOutstanding收到 PINGRESP 时归零。这个计数器的存在意义是TCP 连接在某些网络设备上会出现「半开」状态——本地 socket 认为还连着实际对端早就没了如果不主动判定程序会一直挂在那儿等用户看到的就是「设备在线但不动了」属于典型的死链场景。连接状态机我自己一般是四个状态Disconnected、Connecting、Connected、Disconnecting。所有对外的方法在入口处先检查状态PublishAsync在非Connected状态下要么抛异常要么进队列不允许直接往 socket 写否则会写到一个已经 dispose 的流上抛ObjectDisposedException而且抛的位置往往在后台线程容易被吞掉。4.2 inflight 窗口、报文标识符分配与重连后的重投QoS1 和 QoS2 都是「发出去等确认」的模型中间这段等待期叫 inflight。要把这部分做对需要三样东西一个不撞号的标识符分配器、一个记录待确认报文的表、一套重连后的重投策略。// 报文标识符 1~655350 是保留值 public sealed class PacketIdAllocator { private readonly object _lock new object(); private readonly HashSetushort _inUse new HashSetushort(); private ushort _cursor; public ushort Next() { lock (_lock) { for (int i 0; i ushort.MaxValue; i) { _cursor (ushort)(_cursor % 65535 1); // 从 1 开始循环 if (_inUse.Add(_cursor)) return _cursor; // 没被占用就发出去 } // 全占满说明 inflight 窗口没限流这是设计问题不是运行问题 throw new InvalidOperationException(报文标识符耗尽请检查 inflight 上限); } } public void Release(ushort id) { lock (_lock) { _inUse.Remove(id); } } }这里用HashSet而不是bool[65536]是因为同时 inflight 的消息通常也就几十上百条遍历成本可忽略内存还省。真正需要关注的是那个throw一旦走到这里说明窗口上限设得太大或者 PUBACK 处理有泄漏与其静默复用标识符不如直接报错暴露问题。重连之后的重投是最容易写错的一段。协议规定CleanSessionfalse时客户端必须把未确认的 QoS1/QoS2 消息带上原标识符和 DUP 标志重新发出去。private void ResendInflight() { foreach (var entry in _inflight.Snapshot()) { entry.Packet.Dup true; // DUP 只是提示broker 靠 PacketId 去重 SendRaw(entry.Packet.ToBytes()); entry.Retries; if (entry.Retries 3) { // 超过重试上限就放弃交给业务层补偿别让标识符一直被占着 _inflight.Complete(entry.PacketId); _packetIds.Release(entry.PacketId); OnDeliveryFailed(entry.PacketId); } } }Dup true只在重发时置位首次发送必须为 0。有些实现图省事全程置 1短期内看不出问题但会让 broker 侧的去重逻辑多走一遍判断而且某些严格实现的 broker 会直接拒收。重连本身用指数退避加抖动别用固定间隔private async Task ReconnectLoopAsync(CancellationToken token) { int attempt 0; var rand new Random(Guid.NewGuid().GetHashCode()); while (!token.IsCancellationRequested _state ! MqttConnectionState.Connected) { attempt; // 指数退避封顶 30 秒叠加 0.75~1.25 的随机抖动 double baseDelay Math.Min(30, Math.Pow(2, Math.Min(attempt, 5))); double jitter 0.75 rand.NextDouble() * 0.5; int delayMs (int)(baseDelay * jitter * 1000); try { await ConnectCoreAsync(token).ConfigureAwait(false); } catch (Exception ex) { Log.Warn($第 {attempt} 次重连失败{ex.Message}{delayMs}ms 后重试); await Task.Delay(delayMs, token).ConfigureAwait(false); } } }抖动的必要性在批量设备场景下体现得淋漓尽致几百台设备同时掉线如果没有抖动它们会在完全相同的毫秒级时刻一起打回 broker形成一次小型的连接风暴broker 扛不住就再挂一次然后循环。加上 ±25% 的抖动之后重连请求会被摊平到几秒的窗口里这个细节在实验室环境永远测不出来。ConnectCoreAsync内部要做的事按顺序是建 TCP 连接 → 建 TLS如果配了→ 发 CONNECT → 等 CONNACK → 校验返回码 → 重投 inflight → 重新订阅CleanSessiontrue时必须重订阅→ 恢复心跳定时器。少一步都会在现场变成玄学问题。5. 避坑与排查C# MQTT 客户端最常见的五类故障下面这几条基本都是我在真实项目里熬夜定位出来的。每一条都按「看到什么现象 → 底层是什么原因 → 怎么改」写遇到类似症状可以直接对照。5.1 现象连上几秒就被踢日志里没有任何错误码原因ClientId为空或者重复。MQTT 3.1.1 允许空 ClientId但前提是CleanSessiontruebroker 会给你随机分配一个如果CleanSessionfalse又没给 ClientId多数 broker 直接拒绝。另一种更隐蔽的情况是多台设备用了同一份配置里的固定 ClientId后上线的会把前一个顶下线表现为「两个设备轮流在线」。解决ClientId 强制带唯一后缀比如设备序列号加进程 IDCleanSessionfalse时绝对不允许为空。另外注意 CONNACK 返回码返回 2 就是标识符被拒别只看「连接已关闭」这条笼统日志。5.2 现象中文主题或中文载荷出现乱码英文正常原因主题名用了Encoding.Default编码。这个属性在不同区域设置的机器上分别是 GBK 或别的本地编码而协议规定主题必须是 UTF-8。同一份代码在国内机器上跑没问题出口到别的区域就全乱。载荷乱码则是把byte[]当成string处理了协议对载荷的编码不做规定二进制数据直接按字节收发就行。解决主题、ClientId、用户名、密码统一走Encoding.UTF8载荷全程用byte[]需要序列化就在业务层做别在协议层加转换。// 主题必须 UTF-8载荷保持二进制原样不做任何字符集转换 public Task PublishAsync(string topic, byte[] payload, int qos) { if (string.IsNullOrEmpty(topic) || topic.Length 65535) throw new ArgumentException(主题为空或超长, nameof(topic)); // \( 和 # 是通配符只能出现在订阅侧 if (topic.Contains() || topic.Contains(#)) throw new ArgumentException(发布主题不允许包含通配符, nameof(topic)); return PublishCoreAsync(topic, payload, qos); }5.3 现象QoS1 消息偶发重复业务侧多下了一次配置原因MQTT 的 QoS1 语义是「至少一次」重复本来就在协议允许范围内。真正的坑在于重连后把 inflight 表清空了但又重发了报文或者业务侧把 DUP 标志当成去重依据——DUP 只是「这条可能是重发」的提示broker 和客户端都不保证它准确。解决业务侧做幂等用消息体里的业务唯一键去重不要依赖 QoS 等级或 DUP 标志。如果确实不能容忍重复那就得用 QoS2代价是四次握手吞吐会掉一大截。5.4 现象程序被结束后broker 侧还能看到设备在线遗嘱消息迟迟不触发原因进程被强制结束没走 DISCONNECT。协议规定客户端正常断开时要先发 DISCONNECTbroker 收到后知道这是主动下线不会发布遗嘱。反过来异常断线时 broker 要等心跳超时1.5 倍 keepalive才会判定离线并发布遗嘱。所以遗嘱延迟是正常的延迟时间就是心跳的 1.5 倍。解决进程退出前主动发 DISCONNECT 并优雅关闭 socket把 keepalive 设小一点能让遗嘱更快触发但也更容易误判。另外遗嘱消息在每次 CONNECT 时声明重连后必须在 CONNECT 报文里重新带上不会自动保留。5.5 现象内网一切正常一上公网或 TLS 就卡死不动原因NetworkStream.Read的返回值被忽略了。这个方法只保证「至少读到 1 字节」不保证读满你请求的长度。内网 RTT 低、MTU 内数据一次到齐所以看不出问题上了公网一个大报文被拆成多个 TCP 段第一次Read只返回了一部分剩下的代码就当整条报文去解析剩余长度对不上解析器直接跑飞。解决所有读取都走「读满」的封装并且把「对端返回 0」当成连接关闭处理// 读满 count 字节否则抛异常不能假设一次 Read 就能拿全 private static void ReadExact(Stream stream, byte[] buffer, int offset, int count) { int read 0; while (read count) { int n stream.Read(buffer, offset read, count - read); if (n 0) throw new EndOfStreamException(对端已关闭连接); read n; } }排查这类问题最有效的手段是抓包看字节。把收发的原始字节按十六进制打到日志里对着固定头的第一个字节确认报文类型再用剩余长度核对总字节数问题基本一眼就能定位。别只打「收到消息」这种业务日志那对协议层的问题毫无帮助。6. 二次开发把协议库改成能接业务的客户端真正要拿去接项目协议层基本不用动改动集中在上面的两层。我一般按这个顺序改改造点改哪个文件关键注意业务消息分发MqttClient.cs的PublishReceived用线程池处理别在 IO 回调里做耗时逻辑主题路由Session/TopicMatcher.cs匹配单层#匹配多层且必须在末尾本地离线队列新增OfflineQueue.cs断电不丢要落盘内存队列只防抖连接参数外置MqttConnectOptions从配置文件读别硬编码监控指标门面层埋点上线数、重连次数、inflight 深度主题匹配这块特别值得自己实现一遍因为订阅侧和发布侧的规则不对称// 只用于订阅侧匹配发布侧主题不允许出现通配符 public static bool Match(string filter, string topic) { if (filter topic) return true; string[] f filter.Split(/); string[] t topic.Split(/); for (int i 0; i f.Length; i) { // # 必须在最后一层匹配剩余所有层级包括零层 if (f[i] #) return true; if (i t.Length) return false; // 主题层级不够 if (f[i] ) continue; // 单层通配跳过 if (f[i] ! t[i]) return false; } return f.Length t.Length; // 层数必须完全一致 }一个容易忽略的点sport/#是否匹配sport这个主题按规范是匹配的因为#包含父层级本身。上面这个实现里f.Length是 2、t.Length是 1循环第一轮f[0] t[0]通过第二轮f[1] #直接返回 true行为正确。但如果换成sport/就不匹配sport因为至少要吃一层。这个差别在写订阅路由时经常搞混测试用例里一定要覆盖。验证改造是否到位我习惯做三件事一是断开网线三十秒再插回看 inflight 消息是否按预期重投且不重复入库二是把 broker 重启看客户端能否在退避窗口内自动恢复并重新订阅三是压一万条 QoS1 消息观察 inflight 表的深度是否稳定在设定上限附近而不是一路上涨。从那以后我每次接入一个新的 MQTT 客户端都会先把收发字节的十六进制日志打开跑一遍确认 CONNECT 报文的字段顺序和剩余长度都对得上再往上写业务。这一步花不了十分钟但能省掉后面几天的排查。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
