Hyz.MqttClient 1.0.3

Hyz.MqttClient

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

安装

dotnet add package Hyz.MqttClient

快速开始

1. 注册服务

using Hyz.MqttClient.Extensions;

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

2. 定义消息处理器

消息处理器类必须声明为 partial（源码生成器会在编译期合并 IMqttSubscriber 实现），并实现 IMqttMessageHandler 或派生自 MqttMessageHandlerBase<TMessage>，否则 AddHyzMqttHandlers 扫描时不会发现它。

using Hyz.MqttClient.Core.Attributes;
using Hyz.MqttClient.Interfaces;

public partial class MqttHandlers : IMqttMessageHandler
{
    // 处理字符串消息
    [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;
    }

    // IMqttMessageHandler 接口方法（必须实现）
    public Task HandleMessageAsync(string topic, string payload, CancellationToken cancellationToken = default)
    {
        return Task.CompletedTask;
    }
}

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

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

3. 连接并订阅

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

// 发布消息(组合方法结束后 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] 标记的方法会被源码生成器收集并由 StartHyzMqttSubscriptionsAsync 触发订阅； 也可以通过 IMqttClientHelper.SubscribeAsync(topic, handler) 手动注册接口实现的处理器（不走生成器路径，写入 MessageHandlers 字典）。

订阅启动入口指南

AddHyzMqttHandlers / StartHyzMqttSubscriptionsAsync / StartAllHyzMqttSubscriptionsAsync / ConnectAndStartSubscriptionsAsync / ConnectAndStartAllSubscriptionsAsync 五者职责不同，按场景选择：

Connect* 组合方法：内部三步顺序固定 Use(name) → ConnectFromConfigAsync → StartHyzMqttSubscriptionsAsync，消除「Start* 要求调用方先自行 Use(name)」的隐含不变量，任一步失败立即返回 false 不再静默吞错。

Start* 方法：仅启动订阅，不负责连接，调用前必须先 mqttClient.Use(name).ConnectFromConfigAsync()；否则底层 SubscribeAsync 在 !IsConnected 时静默返回 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);

// 注册
builder.Services.AddHyzMqttConfig(options =>
{
    options.Server = "localhost";
    options.Port = 1883;
});
builder.Services.AddHyzMqtt();
builder.Services.AddHyzMqttHandlers();

var app = builder.Build();

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

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

app.Run();

// 处理器
public partial class MyHandler : IMqttMessageHandler
{
    [MqttSubscribe("topic/test")]
    public async Task Handle(string payload)
    {
        Console.WriteLine(payload);
        await Task.CompletedTask;
    }

    public Task HandleMessageAsync(string topic, string payload, CancellationToken ct = default)
        => 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.Services.AddHyzMqttConfig("server1", options =>
{
    options.Server = "mqtt1.example.com";
});
builder.Services.AddHyzMqttConfig("server2", options =>
{
    options.Server = "mqtt2.example.com";
});
builder.Services.AddHyzMqtt();

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

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

    public Task HandleMessageAsync(string topic, string payload, CancellationToken ct = default)
        => Task.CompletedTask;
}

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

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

WebSocket 连接

builder.Services.AddHyzMqttConfig(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);

MqttConfig 配置

IMqttClientHelper 主要方法

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

事件： MessageReceived、ConnectionStateChanged、ReconnectStatusChanged

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

依赖项

许可证

MIT