一个高性能的 .NET MQTT 客户端与服务器库

前言

在物联网（IoT）蓬勃发展的今天，MQTT 协议已经成为设备通信的事实标准。无论是智能家居、工业自动化还是车联网，MQTT 都扮演着至关重要的角色。今天，我要为大家介绍一个完全使用 C# 实现的高性能 MQTT 库
这个库不仅提供了完整的 MQTT 客户端实现，还包含了一个功能齐全的 Broker 服务器，支持桥接、集群等企业级特性。
核心特性

协议支持

• MQTT 3.1.1 - 完整支持
  
• MQTT 5.0 - 完整支持（包括用户属性、消息过期、主题别名等新特性）
  
• MQTT-SN - 基于 UDP 的轻量级 MQTT 变体，适合受限设备
  
• CoAP - 约束应用协议网关支持
  

性能特性

• 高性能异步实现
  
• 零不必要的内存分配
  
• 缓冲区池技术
  
• 支持 10000+ 并发连接
  

企业级功能

• Broker 桥接（多 Broker 消息同步）
  
• 集群支持（去中心化 P2P 架构）
  
• 灵活的认证与授权机制
  
• TLS/SSL 加密传输
  
• 持久会话与离线消息存储
  

框架支持

• .NET 6.0
  
• .NET 8.0
  
• .NET 10.0
  

技术实现

本项目采用了大量现代 .NET 高性能技术，下面详细介绍核心技术点。
内存管理技术

Span 和 Memory - 零拷贝处理

项目使用 ref struct 实现的二进制读写器，完全在栈上分配，避免堆内存压力：

1. // 零拷贝的二进制读取器
2. public ref struct MqttBinaryReader
3. {
4.     private readonly ReadOnlySpan<byte> _buffer;
5.     private int _position;
7.     // 零拷贝切片操作
8.     [MethodImpl(MethodImplOptions.AggressiveInlining)]
9.     public ReadOnlySpan<byte> ReadBytes(int count)
10.     {
11.         var span = _buffer.Slice(_position, count);
12.         _position += count;
13.         return span;
14.     }
15. }

复制代码

技术优势：

• ref struct 只能在栈上分配，无 GC 压力
  
• ReadOnlySpan 支持零拷贝切片
  
• 避免大量字节数组复制操作
  

ArrayPool - 缓冲区复用

使用共享内存池减少频繁的内存分配：

1. // 从共享池租借缓冲区
2. var buffer = ArrayPool<byte>.Shared.Rent(1024);
3. try
4. {
5.     await stream.ReadAsync(buffer.AsMemory(0, length), cancellationToken);
6.     // 处理数据...
7. }
8. finally
9. {
10.     ArrayPool<byte>.Shared.Return(buffer);  // 归还缓冲区
11. }

复制代码

stackalloc - 小缓冲区栈分配

对于小型临时缓冲区，直接在栈上分配：

1. // 4 字节的可变长度编码缓冲区，栈分配
2. Span<byte> remainingLengthBytes = stackalloc byte[4];
3. var size = EncodeRemainingLength(length, remainingLengthBytes);

复制代码

异步编程模型

async/await + ConfigureAwait

所有 IO 操作均采用异步模式，并使用 ConfigureAwait(false) 优化：

1. public async Task<MqttConnectResult> ConnectAsync(CancellationToken cancellationToken = default)
2. {
3.     // 建立 TCP 连接
4.     await _tcpClient.ConnectAsync(host, port, cancellationToken).ConfigureAwait(false);
6.     // TLS 握手
7.     if (Options.UseTls)
8.     {
9.         await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false);
10.     }
12.     // 发送 CONNECT 报文
13.     await SendPacketAsync(connectPacket, cancellationToken).ConfigureAwait(false);
14. }

复制代码

Channel - 高性能事件队列

Broker 使用有界通道实现非阻塞的事件分发：

1. public sealed class MqttBrokerEventDispatcher
2. {
3.     private readonly Channel<BrokerEvent> _eventChannel;
5.     public MqttBrokerEventDispatcher(int capacity = 10000)
6.     {
7.         // 有界通道，队列满时丢弃最旧事件
8.         _eventChannel = Channel.CreateBounded<BrokerEvent>(new BoundedChannelOptions(capacity)
9.         {
10.             FullMode = BoundedChannelFullMode.DropOldest,
11.             SingleReader = true,
12.             SingleWriter = false
13.         });
14.     }
16.     // 非阻塞事件发送
17.     public void Dispatch<TEventArgs>(BrokerEventType type, TEventArgs args, EventHandler<TEventArgs>? handler)
18.     {
19.         _eventChannel.Writer.TryWrite(new BrokerEvent(type, args, handler));
20.     }
21. }

复制代码

TaskCompletionSource - 请求/响应模式

实现 QoS 1/2 的确认等待机制：

1. private readonly Dictionary<ushort, TaskCompletionSource<object?>> _pendingPackets = new();
3. private async Task<object?> WaitForPacketAsync(ushort packetId, CancellationToken cancellationToken)
4. {
5.     var tcs = new TaskCompletionSource<object?>(TaskCreationOptions.RunContinuationsAsynchronously);
6.     _pendingPackets[packetId] = tcs;
8.     using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
9.     cts.CancelAfter(TimeSpan.FromSeconds(30));  // 30 秒超时
11.     using var registration = cts.Token.Register(() => tcs.TrySetCanceled());
12.     return await tcs.Task.ConfigureAwait(false);
13. }

复制代码

SemaphoreSlim - 发送同步

确保报文发送的串行化：

1. private readonly SemaphoreSlim _sendLock = new(1, 1);
3. private async Task SendPacketBytesAsync(byte[] packet, CancellationToken cancellationToken)
4. {
5.     await _sendLock.WaitAsync(cancellationToken).ConfigureAwait(false);
6.     try
7.     {
8.         await _stream.WriteAsync(packet.AsMemory(), cancellationToken).ConfigureAwait(false);
9.         await _stream.FlushAsync(cancellationToken).ConfigureAwait(false);
10.     }
11.     finally
12.     {
13.         _sendLock.Release();
14.     }
15. }

复制代码

编译器优化

MethodImpl 特性

针对不同场景使用合适的编译器优化指令：

1. // 强制内联 - 用于频繁调用的短方法
2. [MethodImpl(MethodImplOptions.AggressiveInlining)]
3. public ushort ReadUInt16()
4. {
5.     var value = (ushort)((_buffer[_position] << 8) | _buffer[_position + 1]);
6.     _position += 2;
7.     return value;
8. }
10. // 最积极的优化 - 用于热路径
11. [MethodImpl(MethodImplOptions.AggressiveOptimization)]
12. private static async Task<int> DecodeRemainingLengthAsync(Stream stream, CancellationToken ct)
13. {
14.     // 可变长度解码实现...
15. }
17. // 禁止内联 - 避免异常处理代码膨胀热路径
18. [MethodImpl(MethodImplOptions.NoInlining)]
19. private void ThrowIfDisposed()
20. {
21.     if (_disposed) throw new ObjectDisposedException(nameof(MqttClient));
22. }

复制代码

可变长度整数编码

MQTT 协议特有的可变长度编码，1-4 字节可表示 0 到 268,435,455：

1. public interface ITransportConnection : IAsyncDisposable
2. {
3.     string ConnectionId { get; }
4.     TransportType TransportType { get; }
5.     EndPoint? RemoteEndPoint { get; }
6.     bool IsConnected { get; }
8.     ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default);
9.     ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default);
10.     ValueTask FlushAsync(CancellationToken cancellationToken = default);
11. }

复制代码

并发数据结构

ConcurrentDictionary - 线程安全集合

用于管理客户端会话和订阅：

1. var sslOptions = new SslClientAuthenticationOptions
2. {
3.     TargetHost = Options.Host,
4.     EnabledSslProtocols = SslProtocols.Tls12 | SslProtocols.Tls13,
5.     ClientCertificates = Options.ClientCertificate != null
6.         ? new X509CertificateCollection { Options.ClientCertificate }
7.         : null
8. };
10. await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken);

复制代码

设计模式应用

模式应用场景示例工厂模式协议处理器创建MqttProtocolHandlerFactory策略模式不同协议版本实现V311ProtocolHandler / V500ProtocolHandler建造者模式报文构建IPublishPacketBuilder / IConnectPacketBuilder观察者模式事件系统MessageReceived / ClientConnected装饰器模式传输层 TLSSslStream 装饰 NetworkStream单例模式协议处理器缓存全局共享的处理器实例技术栈总结

类别技术作用内存管理Span, Memory, ref struct, ArrayPool, stackalloc零拷贝、栈分配、缓冲区复用异步编程async/await, Channel, TaskCompletionSource, SemaphoreSlim高效并发、非阻塞事件处理编译优化AggressiveInlining, AggressiveOptimization, NoInliningJIT 编译器优化提示网络层TcpClient, TcpListener, SslStream, 传输抽象多协议支持、安全传输并发集合ConcurrentDictionary, ConcurrentQueue线程安全的数据结构序列化自定义二进制读写器、可变长度编码高效的协议解析性能优化建议

客户端优化

• 选择合适的 QoS：大多数场景 QoS 1 就足够了，QoS 2 开销较大
  
• 批量发送：如果有大量消息，考虑合并后发送
  
• 合理设置 KeepAlive：根据网络环境调整，一般 60 秒即可
  
• 使用持久会话：如果需要接收离线消息，设置 CleanSession = false
  

Broker 优化

• 调整最大连接数：根据服务器性能设置 MaxConnections
  
• 限制消息大小：设置 MaxMessageSize 防止恶意大消息
  
• 离线消息限制：设置 MaxOfflineMessagesPerClient 防止内存溢出
  
• 使用集群：高可用场景使用集群部署
  

客户端使用指南

基础连接

1. public static class MqttProtocolHandlerFactory
2. {
3.     private static readonly Lazy<IMqttProtocolHandler> _v311Handler =
4.         new(() => new V311ProtocolHandler());
5.     private static readonly Lazy<IMqttProtocolHandler> _v500Handler =
6.         new(() => new V500ProtocolHandler());
8.     public static IMqttProtocolHandler GetHandler(MqttProtocolVersion version)
9.     {
10.         return version switch
11.         {
12.             MqttProtocolVersion.V311 => _v311Handler.Value,
13.             MqttProtocolVersion.V500 => _v500Handler.Value,
14.             _ => throw new NotSupportedException()
15.         };
16.     }
17. }

复制代码

订阅主题

1. public uint ReadVariableByteInteger()
2. {
3.     uint value = 0;
4.     int multiplier = 1;
5.     byte encodedByte;
7.     do
8.     {
9.         encodedByte = _buffer[_position++];
10.         value += (uint)((encodedByte & 0x7F) * multiplier);
11.         multiplier *= 128;
12.     } while ((encodedByte & 0x80) != 0);
14.     return value;
15. }

复制代码

接收消息

1. private readonly ConcurrentDictionary<string, MqttClientSession> _sessions = new();
2. private readonly ConcurrentDictionary<string, MqttApplicationMessage> _retainedMessages = new();
4. public int ConnectedClients => _sessions.Count;
5. public IEnumerable<MqttClientSession> Sessions => _sessions.Values;

复制代码

发布消息

1. using System.Net.MQTT;
3. // 配置客户端选项
4. var options = new MqttClientOptions
5. {
6.     Host = "localhost",
7.     Port = 1883,
8.     ClientId = "my-iot-device",
9.     CleanSession = true
10. };
12. // 创建客户端
13. using var client = new MqttClient(options);
15. // 连接到 Broker
16. var result = await client.ConnectAsync();
18. if (result.IsSuccess)
19. {
20.     Console.WriteLine("连接成功！");
21. }

复制代码

遗嘱消息（Last Will）

遗嘱消息会在客户端异常断开时自动发布：

1. // 订阅单个主题
2. await client.SubscribeAsync("sensors/temperature", MqttQualityOfService.AtLeastOnce);
4. // 使用通配符订阅多个主题
5. await client.SubscribeAsync("sensors/#", MqttQualityOfService.AtLeastOnce);  // 多级通配符
6. await client.SubscribeAsync("sensors/+/status", MqttQualityOfService.AtMostOnce);  // 单级通配符

复制代码

TLS 加密连接

1. client.MessageReceived += (sender, e) =>
2. {
3.     Console.WriteLine($"收到消息:");
4.     Console.WriteLine($"  主题: {e.Message.Topic}");
5.     Console.WriteLine($"  内容: {e.Message.PayloadAsString}");
6.     Console.WriteLine($"  QoS: {e.Message.QualityOfService}");
7. };

复制代码

自动重连

1. // 简单发布
2. await client.PublishAsync("sensors/temperature", "25.5");
4. // 指定 QoS 发布
5. await client.PublishAsync("sensors/humidity", "60%", MqttQualityOfService.AtLeastOnce);
7. // 发布保留消息
8. await client.PublishAsync("device/status", "online", MqttQualityOfService.AtLeastOnce, retain: true);
10. // 使用完整的消息对象
11. var message = MqttApplicationMessage.Create(
12.     topic: "sensors/data",
13.     payload: "{"temp": 25.5, "humidity": 60}",
14.     qos: MqttQualityOfService.ExactlyOnce,
15.     retain: false
16. );
17. await client.PublishAsync(message);

复制代码

完整客户端示例

1. var options = new MqttClientOptions
2. {
3.     Host = "localhost",
4.     ClientId = "my-device",
5.     WillMessage = MqttApplicationMessage.Create(
6.         topic: "devices/my-device/status",
7.         payload: "offline",
8.         qos: MqttQualityOfService.AtLeastOnce,
9.         retain: true
10.     )
11. };

复制代码

服务器（Broker）使用指南

启动基础 Broker

1. var options = new MqttClientOptions
2. {
3.     Host = "secure-broker.example.com",
4.     Port = 8883,
5.     UseTls = true,
6.     // 可选：客户端证书
7.     ClientCertificate = new X509Certificate2("client.pfx", "password")
8. };

复制代码

配置认证

1. var options = new MqttClientOptions
2. {
3.     Host = "localhost",
4.     AutoReconnect = true,
5.     ReconnectDelayMs = 5000  // 5秒后重连
6. };
8. client.Connected += (s, e) => Console.WriteLine("已连接");
9. client.Disconnected += (s, e) => Console.WriteLine("连接断开，正在重连...");

复制代码

自定义认证器

1. using System.Net.MQTT;
3. var options = new MqttClientOptions
4. {
5.     Host = "localhost",
6.     Port = 1883,
7.     ClientId = $"client-{Guid.NewGuid():N}",
8.     Username = "user",
9.     Password = "password",
10.     CleanSession = true,
11.     KeepAliveSeconds = 60,
12.     AutoReconnect = true
13. };
15. using var client = new MqttClient(options);
17. // 设置事件处理
18. client.Connected += (s, e) => Console.WriteLine("[事件] 已连接到 Broker");
19. client.Disconnected += (s, e) => Console.WriteLine("[事件] 连接已断开");
20. client.MessageReceived += (s, e) =>
21. {
22.     Console.WriteLine($"[消息] {e.Message.Topic}: {e.Message.PayloadAsString}");
23. };
25. // 连接
26. var result = await client.ConnectAsync();
27. if (!result.IsSuccess)
28. {
29.     Console.WriteLine($"连接失败: {result.ReasonCode}");
30.     return;
31. }
33. // 订阅
34. await client.SubscribeAsync("test/#", MqttQualityOfService.AtLeastOnce);
36. // 发布测试消息
37. for (int i = 0; i < 10; i++)
38. {
39.     await client.PublishAsync("test/counter", i.ToString());
40.     await Task.Delay(1000);
41. }
43. // 断开连接
44. await client.DisconnectAsync();

复制代码

Broker 事件处理

1. using System.Net.MQTT.Broker;
3. var options = new MqttBrokerOptions
4. {
5.     Port = 1883,
6.     AllowAnonymous = true,
7.     EnableRetainedMessages = true,
8.     MaxConnections = 10000
9. };
11. using var broker = new MqttBroker(options);
13. // 启动服务器
14. await broker.StartAsync();
15. Console.WriteLine("MQTT Broker 已启动，监听端口 1883");
17. // 保持运行
18. await Task.Delay(Timeout.Infinite);
20. // 停止服务器
21. await broker.StopAsync();

复制代码

TLS 配置

1. // 使用简单认证器
2. broker.Authenticator = new SimpleAuthenticator()
3.     .AddUser("admin", "admin123")
4.     .AddUser("device1", "device1pass")
5.     .AddUser("device2", "device2pass");
7. var options = new MqttBrokerOptions
8. {
9.     Port = 1883,
10.     AllowAnonymous = false  // 禁用匿名访问
11. };

复制代码

高级功能

Broker 桥接

桥接功能允许将多个 Broker 连接起来，实现消息的跨 Broker 同步。

1. public class MyAuthenticator : IMqttAuthenticator
2. {
3.     public Task<MqttAuthenticationResult> AuthenticateAsync(
4.         MqttAuthenticationContext context,
5.         CancellationToken cancellationToken)
6.     {
7.         // 从数据库验证用户
8.         if (ValidateFromDatabase(context.Username, context.Password))
9.         {
10.             return Task.FromResult(MqttAuthenticationResult.Success());
11.         }
13.         return Task.FromResult(MqttAuthenticationResult.Failure(
14.             MqttConnectReasonCode.BadUserNameOrPassword));
15.     }
16. }
18. broker.Authenticator = new MyAuthenticator();

复制代码

集群部署

集群功能实现了去中心化的 P2P 架构，任何节点都可以独立运行，支持自动故障检测和恢复。

1. // 客户端连接事件
2. broker.ClientConnected += (s, e) =>
3. {
4.     Console.WriteLine($"[连接] 客户端 {e.Session.ClientId} 已连接");
5.     Console.WriteLine($"  地址: {e.Session.RemoteEndpoint}");
6.     Console.WriteLine($"  当前连接数: {broker.ConnectedClients}");
7. };
9. // 客户端断开事件
10. broker.ClientDisconnected += (s, e) =>
11. {
12.     Console.WriteLine($"[断开] 客户端 {e.Session.ClientId} 已断开");
13. };
15. // 客户端订阅事件
16. broker.ClientSubscribed += (s, e) =>
17. {
18.     Console.WriteLine($"[订阅] {e.Session.ClientId} 订阅了 {e.TopicFilter}");
19. };
21. // 消息发布事件
22. broker.MessagePublished += (s, e) =>
23. {
24.     Console.WriteLine($"[消息] {e.Message.Topic}: {e.Message.PayloadAsString}");
25.     Console.WriteLine($"  来自: {e.SourceClientId}");
26. };
28. // 消息发布前拦截（可以阻止消息发布）
29. broker.MessagePublishing += (s, e) =>
30. {
31.     // 检查敏感主题
32.     if (e.Message.Topic.StartsWith("admin/") && e.SourceClientId != "admin")
33.     {
34.         e.Cancel = true;  // 阻止非管理员发布到 admin 主题
35.     }
36. };

复制代码

MQTT-SN 网关

MQTT-SN 是基于 UDP 的轻量级协议，适合资源受限的嵌入式设备：

1. var options = new MqttBrokerOptions
2. {
3.     // 普通端口
4.     Port = 1883,
6.     // TLS 端口
7.     UseTls = true,
8.     TlsPort = 8883,
9.     ServerCertificate = new X509Certificate2("server.pfx", "password"),
10.     RequireClientCertificate = false
11. };

复制代码

CoAP 网关

CoAP 网关允许 CoAP 设备与 MQTT 生态系统互通：

1. var broker = new MqttBroker(new MqttBrokerOptions { Port = 2883 });
3. // 添加桥接到父 Broker
4. var bridge = broker.AddBridge(new MqttBridgeOptions
5. {
6.     Name = "parent-bridge",
7.     RemoteHost = "parent-broker.example.com",
8.     RemotePort = 1883,
9.     ClientId = "bridge-client-1",
11.     // 上行规则：本地消息 -> 远程 Broker
12.     UpstreamRules =
13.     {
14.         new MqttBridgeRule { LocalTopicFilter = "sensor/#", Enabled = true },
15.         new MqttBridgeRule { LocalTopicFilter = "device/+/data", Enabled = true }
16.     },
18.     // 下行规则：远程消息 -> 本地
19.     DownstreamRules =
20.     {
21.         new MqttBridgeRule { LocalTopicFilter = "commands/#", Enabled = true },
22.         new MqttBridgeRule { LocalTopicFilter = "config/#", Enabled = true }
23.     }
24. });
26. // 桥接事件
27. bridge.Connected += (s, e) => Console.WriteLine("桥接已连接");
28. bridge.MessageForwarded += (s, e) =>
29. {
30.     var direction = e.Direction == BridgeDirection.Upstream ? "上行" : "下行";
31.     Console.WriteLine($"[桥接-{direction}] {e.OriginalTopic}");
32. };
34. // 获取统计信息
35. var stats = bridge.GetStatistics();
36. Console.WriteLine($"上行消息: {stats.UpstreamMessageCount}");
37. Console.WriteLine($"下行消息: {stats.DownstreamMessageCount}");

复制代码

QoS 服务质量

MQTT 定义了三种服务质量级别：
QoS名称说明适用场景0At Most Once最多一次，不保证送达传感器数据，丢失可接受1At Least Once至少一次，可能重复重要数据，可处理重复2Exactly Once恰好一次，保证送达且不重复计费、订单等关键数据

1. var broker = new MqttBroker(new MqttBrokerOptions { Port = 1883 });
3. // 启用集群
4. broker.EnableCluster(new MqttClusterOptions
5. {
6.     NodeId = "node-1",
7.     ClusterName = "my-cluster",
8.     ClusterPort = 11883,
9.     SeedNodes = new List<string>
10.     {
11.         "node2.example.com:11883",
12.         "node3.example.com:11883"
13.     },
14.     HeartbeatIntervalMs = 5000,
15.     NodeTimeoutMs = 15000,
16.     EnableDeduplication = true  // 防止消息重复
17. });
19. // 集群事件
20. broker.Cluster!.PeerJoined += (s, e) =>
21.     Console.WriteLine($"节点加入: {e.Peer.NodeId}");
23. broker.Cluster!.PeerLeft += (s, e) =>
24.     Console.WriteLine($"节点离开: {e.Peer.NodeId}");
26. broker.Cluster!.MessageForwarded += (s, e) =>
27.     Console.WriteLine($"消息转发: {e.Topic}");
29. await broker.StartAsync();

复制代码

主题通配符

通配符说明示例+匹配单个层级sensor/+/temp 匹配 sensor/room1/temp#匹配多个层级sensor/# 匹配 sensor/room1/temp 和 sensor/room1/humidity

1. var options = new MqttBrokerOptions
2. {
3.     Port = 1883,
4.     EnableMqttSn = true,
5.     MqttSnPort = 1885
6. };

复制代码

MQTT 5.0 新特性

如果你使用 MQTT 5.0 协议，可以利用以下新特性：

1. var options = new MqttBrokerOptions
2. {
3.     Port = 1883,
4.     EnableCoAP = true,
5.     CoapPort = 5683,
6.     CoapMqttPrefix = "mqtt"
7. };
9. // CoAP 客户端可以通过以下方式访问 MQTT 主题：
10. // GET coap://broker:5683/mqtt/sensors/temperature
11. // PUT coap://broker:5683/mqtt/sensors/temperature (发布消息)

复制代码

总结

是一个功能完整、性能优秀的 .NET MQTT 库，具有以下优势：

• 完整的协议支持：MQTT 3.1.1、MQTT 5.0、MQTT-SN、CoAP 全覆盖
  
• 高性能设计：异步 IO、零分配、缓冲区池
  
• 企业级特性：桥接、集群、认证授权
  
• 易于使用：简洁的 API 设计，丰富的示例代码
  
• 现代化：支持最新的 .NET 版本
  

无论你是构建 IoT 平台、实现设备通信，还是搭建消息中间件，这个库都能满足你的需求。
相关链接

源码地址：https://github.com/hnlyf1688/mqtt

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！