文档/中间件/MQTT

📨 MQTT 中间件

包名: Snet.Mqtt | 类: MqttClientOperate, MqttServiceOperate, MqttWebSocketServiceOperate | 基类: MqAbstract<TOperate, TData>

提供 MQTT 3.1.1 客户端连接、内置独立 MQTT 代理以及基于 WebSocket 的 MQTT 服务，适用于工业物联网场景中的实时发布/订阅消息。

概述

Snet.Mqtt 提供三种操作模式：

MqttClientOperate -- 作为客户端连接到外部 MQTT 代理（如 Mosquitto、EMQX、HiveMQ）
MqttServiceOperate -- 在应用程序内运行嵌入式 MQTT 代理，无需外部代理
MqttWebSocketServiceOperate -- 为基于浏览器的客户端提供基于 WebSocket 的 MQTT 服务

这三个类均继承自 MqAbstract<TOperate, TData.Basics> 并实现了 IMq（IProducer + IConsumer）。

安装

dotnet add package Snet.Mqtt

快速开始

using Snet.Mqtt.client;

var mqttClient = new MqttClientOperate(new MqttClientData.Basics
{
    IpAddress = "127.0.0.1",
    Port = 6688
});

await mqttClient.OnAsync();

// 绑定事件再消费
mqttClient.OnDataEventAsync += async (sender, e) =>
{
    if (e.Status)
        Console.WriteLine($"接收到数据: {e.ResultData}");
    else
        Console.WriteLine($"消费失败: {e.Message}");
};

await mqttClient.ConsumeAsync("snet/temperature");

// 生产消息（可选）
await mqttClient.ProduceAsync("snet/temperature", "hello mqtt", System.Text.Encoding.UTF8);

// 后续：取消消费
// await mqttClient.UnConsumeAsync("snet/temperature");
// await mqttClient.DisposeAsync();

MqttClientOperate 配置

参数	类型	默认值	描述
`SN`	string	—	序列号 / 设备标识符
`IpAddress`	string	`"127.0.0.1"`	MQTT 代理 IP 地址
`Port`	int	6688	MQTT 代理端口（默认 6688，TLS 为 8883）
`UserName`	string	`"shunnet"`	认证用户名
`Password`	string	`"shunnet"`	认证密码
`ClientID`	string	自动	唯一客户端标识符
`MessageExpirationTime`	int	86400000	消息过期时间（毫秒）
`QualityOfServiceLevel`	enum	AtMostOnce	QoS 级别：AtMostOnce, AtLeastOnce, ExactlyOnce
`ResponseType`	enum	Content	响应类型

MqttServiceOperate 配置

参数	类型	默认值	描述
`Port`	int	6688	内置代理监听端口
`MaxNumber`	int	10000	最大并发连接数
`UserName`	string	`"shunnet"`	代理认证用户名
`Password`	string	`"shunnet"`	代理认证密码

MqttWebSocketServiceOperate 配置

参数	类型	默认值	描述
`Port`	int	6688	MQTT 代理端口
`WsPort`	int	8866	WebSocket 服务端口
`Uri`	string	`"shun"`	WebSocket 路径段
`UserName`	string	`"shunnet"`	认证用户名
`Password`	string	`"shunnet"`	认证密码

支持的操作

操作	方法	描述
连接 / 启动	`OnAsync()`	连接到代理或启动嵌入式服务
断开 / 停止	`OffAsync()`	断开连接或停止服务
发布	`ProduceAsync(topic, string, Encoding?)`	发布字符串消息
发布	`ProduceAsync(topic, byte[])`	发布原始字节
订阅	`ConsumeAsync(topic)`	订阅主题
取消订阅	`UnConsumeAsync(topic)`	取消订阅主题

QoS 级别

MQTT 支持三种服务质量级别：

QoS	名称	描述
0	最多一次	发后即忘，无确认
1	至少一次	保证送达，可能重复
2	恰好一次	保证送达且无重复

事件

事件	签名	描述
`OnDataEvent`	`EventHandler<EventDataResult>`	收到订阅主题数据时触发
`OnDataEventAsync`	`EventHandlerAsync<EventDataResult>`	OnDataEvent 的异步变体
`OnInfoEvent`	`EventHandler<EventInfoResult>`	收到信息通知时触发
`OnInfoEventAsync`	`EventHandlerAsync<EventInfoResult>`	OnInfoEvent 的异步变体
`OnLanguageEvent`	`EventHandler<EventLanguageResult>`	语言变更通知时触发
`OnLanguageEventAsync`	`EventHandlerAsync<EventLanguageResult>`	OnLanguageEvent 的异步变体