Hyz.MqttClient 1.1.1

Hyz.MqttClient

基于 Roslyn 源码生成器的 MQTT 客户端库，编译时自动生成订阅代码，零运行时反射。

安装

dotnet add package Hyz.MqttClient

快速开始

1. 注册服务

一行完成「连接配置 + helper + handler 扫描」：

using Hyz.MqttClient.Extensions;

builder.Services.AddHyzMqtt(options =>
{
    options.Server = "localhost";
    options.Port = 1883;
    options.ClientId = "MyMqttClient";
});

2. 定义消息处理器

消息处理器类必须声明为 partial（源码生成器会在编译期合并 IMqttSubscriber 实现）。IMqttMessageHandler 接口不是必需的——只有手写订阅路径才需要；纯特性订阅路径只需 partial class + [MqttSubscribe] 即可。

支持任意 namespace（含 C# 10+ 文件级 namespace X; 语法）与任意方法名——SG 始终按 MethodName 字典分发，不会写死调 HandleMessageAsync。

using Hyz.MqttClient.Core.Attributes;

namespace MyApp.Handlers;  // 文件级 namespace 也可

public partial class MqttHandlers
{
    // 处理字符串消息
    [MqttSubscribe("topic/string")]
    public async Task HandleStringMessage(string payload)
    {
        Console.WriteLine($"收到: {payload}");
        await Task.CompletedTask;
    }

    // 处理强类型消息（自动 JSON 反序列化）
    // 第一个参数类型非 string 时生成器按 JSON 反序列化处理
    [MqttSubscribe("topic/json")]
    public async Task HandleJsonMessage(MyMessage message)
    {
        Console.WriteLine($"收到: {message.Content}");
        await Task.CompletedTask;
    }
}

public class MyMessage
{
    public string Content { get; set; }
}

若需在方法内直接处理原始事件参数（如访问 MQTT 5.0 用户属性），可将首参数类型改为 MqttApplicationMessageReceivedEventArgs。

强类型处理器也可以派生 MqttMessageHandlerBase<TMessage>，自动获得 JSON 反序列化与错误处理模板方法，参见下方 消息处理器接口。

何时 HandleMessageAsync 是真实入口？

HandleMessageAsync 仅在「手写订阅」路径下被调用：

写法 A：直接实现 IMqttMessageHandler（无需 partial，无需 SG）：

// 路径 2：手写订阅 — 类不必为 partial，SG 不参与
public class Foo : IMqttMessageHandler
{
    public Task HandleMessageAsync(string topic, string payload, CancellationToken ct = default)
    {
        // payload 是原始字符串；JSON 反序列化由你自行处理
        Console.WriteLine($"{topic}: {payload}");
        return Task.CompletedTask;
    }
}

// 运行时手动订阅（不走 SG 路径，写入 MessageHandlers 字典）
var handler = new Foo();
await mqttClient.SubscribeAsync("test/topic", handler, MqttQualityOfServiceLevel.AtLeastOnce);

写法 B：派生 MqttMessageHandlerBase<T> 做 JSON 反序列化：

public class TypedHandler : MqttMessageHandlerBase<TelemetryMessage>
{
    public override Task HandleMessageAsync(string topic, TelemetryMessage message, CancellationToken ct = default)
    {
        // 基类已做 JSON 反序列化与错误处理
        Console.WriteLine($"{topic}: {message.DeviceId}");
        return Task.CompletedTask;
    }
}

var handler = new TypedHandler(...);
await mqttClient.SubscribeAsync<TelemetryMessage>("sensors/+/data", handler, MqttQualityOfServiceLevel.AtLeastOnce);

这条路径适合：

• 不希望 SG 在编译期生成 partial 代码的强类型订阅
• 运行时动态创建 handler 实例
• 需要在订阅时携带上下文对象（如 logger、配置）的场景

⚠️ 空实现会被静默调用：如果你写 => Task.CompletedTask;，消息抵达后会被默默丢弃——broker 已收到、应用已接收，但你的业务逻辑没跑。路径 2 想要有意义的处理，必须在 HandleMessageAsync 里写实际逻辑。

如果你的类只用 [MqttSubscribe] 特性订阅，可以彻底删除 IMqttMessageHandler 接口与 HandleMessageAsync 方法——源生成器路径完全不依赖它们。

3. 连接并订阅

// 一行完成「连接 + 启动所有订阅」
await sp.StartHyzMqttAsync();

// 发布消息(组合方法结束后 helper 停在最后处理的连接上,Publish 前按需 Use 切换)
var mqttClient = sp.GetRequiredService<IMqttClientHelper>();
await mqttClient.Use("default").PublishAsync("topic/string", "Hello MQTT!");

核心特性

特性详解

MqttSubscribeAttribute

[MqttSubscribe(
    topic: "device/+/data",      // 必填：订阅主题
    qos: MqttQoS.AtLeastOnce,    // QoS 等级，默认 AtMostOnce
    connectionName: "server1",   // 连接名称，默认使用默认连接
    messageType: typeof(MyMessage),  // 消息类型，用于自动反序列化
    enableAutoDeserialization: true  // 是否启用自动反序列化
)]
public async Task HandleMessage(MyMessage message) { }

方法参数类型：

消息处理器接口

库提供三层消息处理器抽象，按需选择：

[MqttSubscribe] 标记的方法会被源码生成器收集并由 StartHyzMqttAsync 触发订阅； 也可以通过 IMqttClientHelper.SubscribeAsync(topic, handler) 手动注册接口实现的处理器（不走生成器路径，写入 MessageHandlers 字典）。

订阅启动入口指南

StartHyzMqttAsync 是推荐的统一启动入口，覆盖 90% 场景：

StartHyzMqttAsync 内部：固定三步顺序 Use(name) → ConnectFromConfigAsync → 启动订阅，消除「调用方需先自行 Use(name)」的隐含不变量；任一步失败立即返回 false，不再静默吞错。

QoS 等级

public enum MqttQoS
{
    AtMostOnce = 0,   // 最多一次
    AtLeastOnce = 1,  // 至少一次
    ExactlyOnce = 2   // 恰好一次
}

完整示例

基本发布/订阅

using Hyz.MqttClient.Core.Attributes;
using Hyz.MqttClient.Extensions;
using MQTTnet.Protocol;

var builder = WebApplication.CreateBuilder(args);

// 一站式注册：连接配置 + helper + handler 扫描
builder.Services.AddHyzMqtt(options =>
{
    options.Server = "localhost";
    options.Port = 1883;
});

var app = builder.Build();

// 一行 connect + 启动所有订阅
await app.Services.StartHyzMqttAsync();

// 发布(组合方法结束后 helper 停在最后处理的连接,Publish 前按需 Use)
var mqttClient = app.Services.GetRequiredService<IMqttClientHelper>();
await mqttClient.Use("default").PublishAsync("topic/test", "Hello!", MqttQualityOfServiceLevel.AtMostOnce);

app.Run();

// 处理器（特性订阅路径：仅需 partial + [MqttSubscribe]，无须实现 IMqttMessageHandler）
public partial class MyHandler
{
    [MqttSubscribe("topic/test")]
    public async Task Handle(string payload)
    {
        Console.WriteLine(payload);
        await Task.CompletedTask;
    }
}

强类型消息

派生 MqttMessageHandlerBase<TMessage> 可复用默认 JSON 反序列化与错误处理：

public class SensorData
{
    public string DeviceId { get; set; }
    public double Temperature { get; set; }
}

public partial class SensorHandler : MqttMessageHandlerBase<SensorData>
{
    public override Task HandleMessageAsync(string topic, SensorData data, CancellationToken ct = default)
    {
        Console.WriteLine($"设备 {data.DeviceId}: {data.Temperature}°C");
        return Task.CompletedTask;
    }

    [MqttSubscribe("sensor/+/data")]
    public Task OnTelemetry(SensorData data) => HandleMessageAsync(data.DeviceId, data, default);
}

// 发布时需要传递 JSON 字符串
var json = JsonSerializer.Serialize(new SensorData { DeviceId = "001", Temperature = 25.5 });
await mqttClient.PublishAsync("sensor/001/data", json);

多连接

// 一站式注册：多连接 builder
builder.Services.AddHyzMqtt(mqtt =>
{
    mqtt.AddConnection("server1", options => options.Server = "mqtt1.example.com");
    mqtt.AddConnection("server2", options => options.Server = "mqtt2.example.com");
});

// 处理器指定连接（仅 partial + [MqttSubscribe]）
public partial class MultiHandler
{
    [MqttSubscribe("topic/a", ConnectionName = "server1")]
    public Task HandleA(string payload) => Task.CompletedTask;

    [MqttSubscribe("topic/b", ConnectionName = "server2")]
    public Task HandleB(string payload) => Task.CompletedTask;
}

// 发布到指定连接
mqttClient.Use("server1").PublishAsync("topic/a", "message");

// 一行 connect + 启动该连接订阅
await sp.StartHyzMqttAsync("server1");
// 或：批量 connect + 启动所有连接
await sp.StartHyzMqttAsync();

WebSocket 连接

builder.Services.AddHyzMqtt(options =>
{
    options.WebSocketUrl = "wss://mqtt.example.com/mqtt";
});

MQTT 5.0 用户属性

var properties = new List<MqttUserProperty>
{
    new("correlation-id", Guid.NewGuid().ToString())
};

await mqttClient.PublishAsync(
    "topic/data",
    jsonPayload,
    MqttQualityOfServiceLevel.AtLeastOnce,
    retain: false,
    properties);

HyzMqttOptions 配置

IMqttClientHelper 主要方法

属性： IsConnected、IsReconnecting、CurrentReconnectAttempts、ClientId、CurrentConnectionName

事件：

伴生扩展方法（Hyz.MqttClient.Extensions.MqttServiceExtensions）：

监控与诊断

helper 内置三类只读诊断方法，可在不侵入业务代码的前提下输出运行时状态。

连接健康度

GetConnectionHealth() 返回 ConnectionHealthInfo，适合做心跳探针 / 健康检查端点：

var health = mqttClient.GetConnectionHealth();
// health.IsConnected           : 是否已连接
// health.ConnectedDuration     : 当前连接已持续时长（TimeSpan）
// health.TotalMessagesReceived : 累计接收消息数
// health.TotalMessagesPublished: 累计发布消息数
// health.LastMessageReceivedAt : 上次收到消息的 UTC 时间
// health.ReconnectAttempts     : 当前连接上的累计重连次数

自动重连状态

GetReconnectInfo() 返回 ReconnectInfo，对接入层异常检测很有用——可直接暴露给监控系统：

var info = mqttClient.GetReconnectInfo();
// info.IsReconnecting : 是否正在自动重连
// info.Attempts       : 已尝试次数
// info.MaxAttempts    : 配置的最大尝试次数（0 表示无限）
// info.NextDelay      : 下次重连前的等待时长
// info.LastAttempt    : 上次尝试的 UTC 时间
// info.Strategy       : FixedDelay / LinearBackoff / ExponentialBackoff

已订阅主题清单

GetSubscribedTopics() 返回当前连接实际下发给 broker 的主题列表（去重），可用于：

• 启动后断言订阅是否符合预期（QA 场景）
• 动态主题管理界面展示

foreach (var topic in mqttClient.GetSubscribedTopics())
{
    Console.WriteLine($"已订阅: {topic}");
}

事件订阅

若需要细粒度响应（写入日志、上报监控），可直接订阅 helper 暴露的事件：

mqttClient.ConnectionStateChanged += (_, isConnected) =>
{
    logger.LogInformation("MQTT 连接状态变更: {State}", isConnected ? "已连接" : "已断开");
};

mqttClient.ReconnectStatusChanged += (_, status) =>
{
    logger.LogWarning("重连状态: {Status}", status);  // Started / Attempting / Success / Failed / ...
};

依赖项

许可证

MIT