跳至主要內容

32. MQTT 集成

后端手册MQTTIoT大约 7 分钟约 2240 字

一句话定位：Admin.NET 通过 MQTTnet 在应用进程内** Hosting 一个 MQTT Broker**（不用额外装 Mosquitto 等EMQX），所有客户端的连/断/订阅/发布事件都可通过继承 MqttEventInterceptor 挂业务钩子，天然适合 IoT 设备接入场景。

一、整体结构

                       ┌─────────────────────────────────────────┐
   IoT 设备/网关 ──────►│ MqttHostedService (MQTTnet MqttServer)  │
   (TCP/MQTT)          │  端口 Mqtt.Port = 1883                    │
                       └───────────────┬─────────────────────────┘
                                       │ 事件：Started / Stopped / ValidatingConnection /
                                       │       ClientConnected / ClientDisconnected /
                                       │       Subscribed / Unsubscribed /
                                       │       InterceptingPublish / NotConsumed
                                       ▼
                       MqttHostedService.MqttEventInterceptors (按 Order 倒序)
                                       │
                                       ├── DefaultMqttEventInterceptor（框架内置）
                                       └── TestMqttEventInterceptor（应用层示例，你的钩子写这里）

   服务端主动下发：MqttHostedService.PublicMessageAsync(topic, message)
   消息轨迹查看：MqttHostedService.GetMessageRecords()（环形队列，最多 1000 条）

关键源码位置：

文件作用
Admin.NET.Core/Job/MqttHostedService.csBroker 生命周期、事件分发、拦截器注册、服务端发布、消息记录
Admin.NET.Core/Mqtt/MqttEventInterceptor.cs拦截器虚基类（9 个可重写事件）
Admin.NET.Core/Mqtt/DefaultMqttEventInterceptor.cs框架默认实现
Admin.NET.Application/Configuration/Mqtt.json配置
Admin.NET.Application/Startup.csAddMqttEventInterceptor(new TestMqttEventInterceptor())

二、服务端内嵌 Broker 配置

// Admin.NET.Application/Configuration/Mqtt.json
{
  "$schema": "https://gitee.com/dotnetchina/Furion/raw/v4/schemas/v4/furion-schema.json",
  "Mqtt": {
    "Enabled": true,              // 是否开启
    "Port": "1883",               // 端口
    "IPAddress": "",              // IP地址（空=默认）
    "ConnectionBacklog": 1000,    // 最大连接数
    "MqttServerId": "Admin.NET.MQTT", // 服务器主动发消息时的 ClientId
    "Logging": true               // 是否在控制台/日志输出消息轨迹
  }
}

注册三行（Admin.NET.Web.Core/Startup.cs）：

services.AddHostedService<StartHostedService>();
services.AddMqttTcpServerAdapter();      // MQTTnet.AspNetCore 的 TCP 适配
services.AddMqttConnectionHandler();
services.AddHostedService<MqttHostedService>();

启动流程：

public async Task StartAsync(CancellationToken cancellationToken)
{
    if (!_mqttOptions.Enabled) return;

    AddMqttEventInterceptor(new DefaultMqttEventInterceptor());     // 注册内置拦截器

    var options = new MqttServerOptionsBuilder()
        .WithDefaultEndpoint()                                      // 默认地址 127.0.0.1
        .WithDefaultEndpointPort(_mqttOptions.Port)                 // 端口
        .WithConnectionBacklog(_mqttOptions.ConnectionBacklog)      // 最大连接数
        .WithPersistentSessions()                                   // 持久化会话
        .Build();

    MqttServer = new MqttServerFactory().CreateMqttServer(options, _serverAdapters);
    MqttServer.StartedAsync   += MqttServer_StartedAsync;
    MqttServer.StoppedAsync   += MqttServer_StoppedAsync;
    MqttServer.ValidatingConnectionAsync            += MqttServer_ValidatingConnectionAsync;
    MqttServer.ClientConnectedAsync                 += MqttServer_ClientConnectedAsync;
    MqttServer.ClientDisconnectedAsync              += MqttServer_ClientDisconnectedAsync;
    MqttServer.ClientSubscribedTopicAsync           += MqttServer_ClientSubscribedTopicAsync;
    MqttServer.ClientUnsubscribedTopicAsync         += MqttServer_ClientUnsubscribedTopicAsync;
    MqttServer.InterceptingPublishAsync             += MqttServer_InterceptingPublishAsync;
    MqttServer.ApplicationMessageNotConsumedAsync   += MqttServer_ApplicationMessageNotConsumedAsync;

    await MqttServer.StartAsync();
}

端口是字符串类型的 "1883"，注意 MqttOptions.Port 的实际类型与 int.Parse 时机，改配置别写成数字以外的值。

三、MqttEventInterceptor 拦截器写法

基类提供 9 个虚方法，默认全是空实现（await Task.CompletedTask）：

// Admin.NET.Core/Mqtt/MqttEventInterceptor.cs
public class MqttEventInterceptor
{
    public int Order = 0;                                            // 数值越大越先执行

    public virtual async Task StartedAsync(EventArgs arg);
    public virtual async Task StoppedAsync(EventArgs arg);
    public virtual async Task ValidatingConnectionAsync(ValidatingConnectionEventArgs arg);
    public virtual async Task ClientConnectedAsync(ClientConnectedEventArgs arg);
    public virtual async Task ClientDisconnectedAsync(ClientDisconnectedEventArgs arg);
    public virtual async Task ClientSubscribedTopicAsync(ClientSubscribedTopicEventArgs arg);
    public virtual async Task ClientUnsubscribedTopicAsync(ClientUnsubscribedTopicEventArgs arg);
    public virtual async Task InterceptingPublishAsync(InterceptingPublishEventArgs arg);
    public virtual async Task ApplicationMessageNotConsumedAsync(ApplicationMessageNotConsumedEventArgs arg);

    protected static void Logging(string msg)                        // 受 Mqtt.Logging 开关控制
    protected static ISqlSugarClient CreateDbClient()                // 独立连接：CopyNew()
}

注册方式：

// Admin.NET.Application/Startup.cs
public void ConfigureServices(IServiceCollection services)
{
    ...
    // 添加自定义 MQTT 事件拦截器
    MqttHostedService.AddMqttEventInterceptor(new TestMqttEventInterceptor());
}

排序规则源码：

public static void AddMqttEventInterceptor(MqttEventInterceptor mqttEventInterceptor, int order = 0)
{
    mqttEventInterceptor.Order = order;
    MqttEventInterceptors.Add(mqttEventInterceptor);
    MqttEventInterceptors.Sort((a, b) => b.Order - a.Order); // 降序：Order 越大越先执行
}

完整示例：设备鉴权 + 遥测入库 + 指令下发

using MQTTnet.Protocol;
using MQTTnet.Server;

namespace Admin.NET.Application;

/// <summary>
/// 设备接入事件拦截器：鉴权、上下线记录、遥测入库
/// </summary>
public class DeviceMqttEventInterceptor : MqttEventInterceptor
{
    public override Task ValidatingConnectionAsync(ValidatingConnectionEventArgs arg)
    {
        // 约定：ClientId = device:{设备编号}；UserName = 产品Key；Password = 设备密钥
        if (!arg.ClientId.StartsWith("device:", StringComparison.OrdinalIgnoreCase))
        {
            arg.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid;
            return Task.CompletedTask;
        }

        var secret = arg.Password;
        // TODO: 换成你自己的设备密钥校验（如查 BizDevice 表比对 Secret）
        if (string.IsNullOrWhiteSpace(secret) || secret != "预设的设备密钥")
        {
            arg.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;   // 鉴权失败
            Logging($"设备鉴权失败：{arg.ClientId}");
            return Task.CompletedTask;
        }

        arg.ReasonCode = MqttConnectReasonCode.Success;                      // 必须显式置成功
        return Task.CompletedTask;
    }

    public override async Task ClientConnectedAsync(ClientConnectedEventArgs arg)
    {
        var db = CreateDbClient();                                           // ← 关键：独立连接
        await db.Insertable(new BizDeviceOnlineLog
        {
            ClientId = arg.ClientId, UserName = arg.UserName,
            EventType = "Online", CreateTime = DateTime.Now
        }).ExecuteCommandAsync();

        Logging($"设备上线：{arg.ClientId}");
    }

    public override async Task ClientDisconnectedAsync(ClientDisconnectedEventArgs arg)
    {
        var db = CreateDbClient();
        await db.Updateable<BizDevice>()
                .SetColumns(u => u.Online == false)
                .Where(u => u.ClientId == arg.ClientId)
                .ExecuteCommandAsync();
        Logging($"设备离线：{arg.ClientId}");
    }

    public override async Task InterceptingPublishAsync(InterceptingPublishEventArgs arg)
    {
        // 主题约定：device/{productKey}/{deviceNo}/telemetry
        var topic = arg.ApplicationMessage.Topic;
        if (!topic.EndsWith("/telemetry", StringComparison.OrdinalIgnoreCase)) return;

        var payload = Encoding.UTF8.GetString(arg.ApplicationMessage.Payload);
        var db = CreateDbClient();
        await db.Insertable(new BizDeviceTelemetry
        {
            Topic = topic, Payload = payload, CreateTime = DateTime.Now
        }).ExecuteCommandAsync();
    }
}

注册：

MqttHostedService.AddMqttEventInterceptor(new DeviceMqttEventInterceptor(), order: 100);

拦截器执行策略（源码）：

private static async Task InvokeInterceptorsAsync(Func<MqttEventInterceptor, Task> invoke)
{
    foreach (var eh in MqttEventInterceptors.ToArray())
    {
        try { await invoke(eh); }
        catch (Exception ex) { Log.Error($"【MQTT】事件拦截器执行失败：{eh.GetType().Name}", ex); }
    }
}

即：单个拦截器抛错只记日志，不影响其它拦截器，也不会打断 MQTT 服务线程。唯一的例外是 ValidatingConnectionAsync，它有自己的特殊处理：

private async Task MqttServer_ValidatingConnectionAsync(ValidatingConnectionEventArgs arg)
{
    foreach (var eh in MqttEventInterceptors.ToArray())
    {
        try { await eh.ValidatingConnectionAsync(arg); }
        catch (Exception ex)
        {
            arg.ReasonCode = MqttConnectReasonCode.ServerUnavailable;
            Log.Error(...);
        }
        if (arg.ReasonCode != MqttConnectReasonCode.Success) break;   // 任一拦截器不通过即中断
    }
}

四、发布 / 订阅示例

4.1 服务端下发（Broker 主动推）

public async Task<bool> PublicMessageAsync(string topic, string message)
{
    var applicationMessage = new MqttApplicationMessageBuilder()
       .WithTopic(topic)
       .WithPayload(message)
       .Build();

    await MqttServer.InjectApplicationMessage(new InjectedMqttApplicationMessage(applicationMessage)
    {
        SenderClientId  = _mqttOptions.MqttServerId,
        SenderUserName  = _mqttOptions.MqttServerId,
    });

    AddMessageRecord("ServerPublish", ...);
    Logging($"服务器发布主题：{topic}, 内容：{message}");
}

业务调用：

public class BizDeviceCmdService : IDynamicApiController, ITransient
{
    private readonly MqttHostedService _mqttHostedService;

    public BizDeviceCmdService(MqttHostedService mqttHostedService) => _mqttHostedService = mqttHostedService;

    /// <summary>
    /// 下发重启指令 🔖
    /// </summary>
    [DisplayName("下发重启指令")]
    public async Task Reboot(string deviceNo)
    {
        var topic = $"device/telemetry/{deviceNo}/cmd";
        await _mqttHostedService.PublicMessageAsync(topic, JsonSerializer.Serialize(new { cmd = "reboot", ts = DateTimeOffset.Now.ToUnixTimeSeconds() }));
    }
}

4.2 服务端作为客户端订阅（连自己的 Broker 或外部 Broker）

using MQTTnet;

var factory = new MqttClientFactory();
using var client = factory.CreateMqttClient();

var options = new MqttClientOptionsBuilder()
    .WithTcpServer("127.0.0.1", 1883)
    .WithClientId(Guid.NewGuid().ToString())
    .WithCredentials("admin", "pwd")
    .WithCleanSession()
    .Build();

client.ApplicationMessageReceivedAsync += e =>
{
    var topic = e.ApplicationMessage.Topic;
    var payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
    Console.WriteLine($"收到 {topic}: {payload}");
    return Task.CompletedTask;
};

await client.ConnectAsync(options);
await client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("device/#").Build());
await client.PublishStringAsync("device/telemetry/001/cmd", "{\"cmd\":\"reboot\"}");

4.3 设备侧（ESP32 Arduino 风格伪代码）

client.setServer("192.168.1.100", 1883);
client.setCallback([](char* topic, byte* payload, unsigned int len){
    // 处理下发指令
});
client.connect("device:DEV001", "productKey", "deviceSecret");
client.subscribe("device/telemetry/DEV001/cmd");
client.publish("device/telemetry/DEV001/telemetry", "{\"temp\":26.5,\"hum\":61}");

五、消息轨迹查看

MqttHostedService 内部维护了一个环形队列，最多保留最近 1000 条：

private const int MaxMessageRecordCount = 1000;
private static readonly ConcurrentQueue<MqttMessageRecordOutput> MessageRecords = new();

public static List<MqttMessageRecordOutput> GetMessageRecords()   // 按时间倒序返回
public static int GetMessageRecordCount()
public static void ClearMessageRecords()

MqttMessageRecordOutput 字段：Id / Time / EventType / ClientId / UserName / RemoteEndPoint / Topic / Payload / PayloadLength / QualityOfServiceLevel / Retain / Remark。

EventType 取值（源码写入点）：ClientConnected、ClientDisconnected、Subscribed、Unsubscribed、ClientPublish、ServerPublish、NotConsumed。

Payload 超过 4096 字符会被截断（payload[..4096]），PayloadLength 保存的是真实字节数。

把它接成一个管理端接口（示例）：

[ApiDescriptionSettings(Order = 950, Description = "MQTT监控")]
public class SysMqttMonitorService : IDynamicApiController, ITransient
{
    [DisplayName("MQTT消息轨迹")]
    public List<MqttMessageRecordOutput> Records() => MqttHostedService.GetMessageRecords();

    [DisplayName("清空MQTT消息轨迹")]
    public void Clear() => MqttHostedService.ClearMessageRecords();
}

六、与 IoT 场景结合的典型设计

┌────────┐  CONNECT device:{no} / productKey / deviceSecret   ┌──────────────┐
│ 设备   │ ─────────────────────────────────────────────────► │              │
│        │  PUBLISH device/{pk}/{no}/telemetry                │  Admin.NET   │
└────────┘ ◄───────────────────────────────────────────────── │  Broker      │
            PUBLISH device/{pk}/{no}/cmd  (服务端下发)         └──────┬───────┘
                                                                     │ InterceptingPublishAsync
                                                                     ▼
                                                           BizDeviceTelemetry 落库
                                                                     │ 超过阈值
                                                                     ▼
                                            事件总线发 MQTT 告警 → SignalR 推给 Web 端

落地建议：

需求推荐做法
海量遥测InterceptingPublishAsync 里只写遥测表或转发 RabbitMQ/Kafka，别做重业务逻辑
设备影子ClientDisconnectedAsync 更新设备在线状态 + Retain 消息保存最后状态
指令下发PublicMessageAsync + QoS 1；重要指令落 SysLogMsg/业务指令表等待 ACK
OTA 升级下发 URL，设备自行拉取；进度用 .../ota/progress 主题回报
多租户ClientId 或 UserName 带租户标识，ValidatingConnectionAsync 里解析并落后续业务实体
离线消息开 WithPersistentSessions() + QoS≥1 + CleanSession=false

七、步骤清单

  1. Mqtt.json 打开 Enabled，改端口（生产建议不要用明文 1883 直曝公网，用 8883 + TLS）
  2. 写自己的 XxxMqttEventInterceptor : MqttEventInterceptor
  3. 在 Admin.NET.Application/Startup.cs 的 ConfigureServices 里 MqttHostedService.AddMqttEventInterceptor(...)
  4. 拦截器里访问数据库必须用 CreateDbClient()（CopyNew()）
  5. 需要下发就注入 MqttHostedService 调 PublicMessageAsync
  6. 设备侧按 device:{no} + productKey + deviceSecret 约定接入
  7. 调试期打开 Logging: true，生产可视情况关闭（每条消息都会 Console.WriteLine）

⚠️ 注意事项 / 常见坑

  1. 绝不要在拦截器里用作用域仓储：MQTT 回调跑在线程池线程，没有 HttpContext，注入 SqlSugarRepository<T> 会抛"无法访问已释放的作用域"。框架已提供 CreateDbClient()（App.GetRequiredService<ISqlSugarClient>().CopyNew()），照用即可。
  2. 单机 Broker 不能集群：Broker 在应用进程内，多副本部署时设备分散连到不同节点，节点之间没有消息同步。要做集群请网关 Broker（EMQX/Mosquitto bridge）或让设备固定连某一节点。
  3. ClientId 冲突会互踢：同一 ClientId 二次连接会断开前一个，生产请保证全局唯一（建议 device:{no}:{随机后缀}）。
  4. ValidatingConnectionAsync 必须显式置 Success：拦截器链以 ReasonCode != Success 为中断条件（源码里的 break），所以通过时一定要写 arg.ReasonCode = MqttConnectReasonCode.Success，校验失败则置对应的错误码。
  5. 消息记录是进程内存：重启即丢，GetMessageRecords() 只能当调试/监控用，不能当审计。
  6. Payload 被截断到 4096：大 payload 业务请以 PayloadLength 判定，别信任前端显示内容长度。
  7. 端口是字符串配置：改配置时写成 "1883" 而非 1883，否则绑定 Options 可能失败。
  8. WithPersistentSessions() 会吃内存：大量离线设备 + CleanSession=false 会让 Broker 保留会话数据，要评估。
  9. 拦截器异常被静默吞掉：InvokeInterceptorsAsync catch 后只写日志。调试时记得看日志，别以为功能没执行。
  10. 公网裸奔风险：MQTT 默认无鉴权，DefaultMqttEventInterceptor 只是默认放行。上生产前一定在 ValidatingConnectionAsync 里做鉴权，并启用 TLS。

小结

  • Broker 内嵌在应用进程，Mqtt.json 一个开关搞定，默认端口 1883
  • 业务钩子继承 MqttEventInterceptor，在 Admin.NET.Application/Startup.cs 用 AddMqttEventInterceptor 注册，Order 越大越先执行
  • 9 个事件里最常用：ValidatingConnectionAsync（鉴权）、ClientConnected/Disconnected（上下线）、InterceptingPublishAsync（收数）
  • 服务端下发走 MqttHostedService.PublicMessageAsync(topic, message)
  • 拦截器里访问数据库一律 CreateDbClient()（CopyNew()）
  • 最近 1000 条消息轨迹可用 GetMessageRecords() 查看，仅用于调试监控