MqttCommunication 1.0.0.3

MqttCommunication 使用说明

面向设备与主机的 MQTT 双向通信类库。目标框架为 .NET Standard 2.0，可供 .NET Framework 4.8 和 .NET 10 程序引用。当前项目版本为 1.0.0.3。

设备程序（MqttDeviceClient） ←→ Mosquitto :1883 ←→ 主机程序（MqttHostClient）

设备负责上报数据、状态和处理指令；主机负责接收数据、发送指令并等待业务回复。两端连接同一个 Broker，不是直接连接彼此的 IP。

安装

在主机和设备项目中分别安装：

dotnet add package MqttCommunication --version 1.0.0.3

NuGet 自动引入 MQTTnet 4.3.7.1207 和 Newtonsoft.Json 13.0.3。不要再同时添加另一个版本的本地 DLL 引用。手动分发 DLL 时，需要同时分发依赖，保留同目录下的 MqttCommunication.xml 才能显示调用提示。

连接配置

下面示例代码分别放入两端程序中。引入命名空间：

using System;
using System.Threading.Tasks;
using MqttCommunication;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;

两端使用相同服务器、端口和主题前缀：

var settings = new MqttConnectionSettings
{
    Host = "localhost",
    Port = 1883,
    TopicPrefix = "auto-ventilation",
    OperationTimeout = TimeSpan.FromSeconds(10)
};

同机测试使用 localhost；跨电脑时，Host 填写 Mosquitto 所在电脑的 IP，例如 192.168.X.XX。服务需要认证时设置 Username、Password。

每个程序保存并复用一个客户端实例。以下 await 代码放在异步方法内；字段放在窗口或业务类中。

设备连接与 IP

private MqttDeviceClient device;

// 自动获取运行设备程序的电脑 IPv4。
device = new MqttDeviceClient("device01", settings);

// 如果需要指定上报 IP，使用下面的构造方式替代上一行：
// device = new MqttDeviceClient("device01", settings, "192.168.X.XX");

ipAddress 未传、为空或空白时自动获取。自动选择优先使用有 IPv4 网关的活动网卡，多网卡机器可显式指定。地址用于状态上报，不是绑定 MQTT 连接的本地网卡。在线、离线与遗嘱使用同一地址。

设备 MQTT ClientId 为 device-device01；主机默认 ClientId 为 auto-ventilation-host。同一 Broker 上 ClientId 必须唯一。设备编号在构造后应保持不变。

先注册处理器，再连接：

device.Error += ex =>
{
    Console.WriteLine("设备通信异常：" + ex.Message);
};

device.CommandHandler = command =>
{
    switch (command.CommandType)
    {
        case CommandType.GETCOMPONENT:
        {
            // 示例结构，实际应读取设备组分。
            return Task.FromResult(new CommandReply
            {
                Success = true,
                Message = "读取成功",
                DataJson = JsonConvert.SerializeObject(new { Components = new[] { "CO", "CO2" } })
            });
        }
        case CommandType.GETRANGE:
        {
            // 示例结构，实际应读取设备量程。
            return Task.FromResult(new CommandReply
            {
                Success = true,
                Message = "读取成功",
                DataJson = JsonConvert.SerializeObject(new { Min = 0, Max = 100, Unit = "ppm" })
            });
        }
        case CommandType.STARTTEST:
        {
            // 在这里解析 command.ParametersJson 并调用真实启动逻辑。
            return Task.FromResult(new CommandReply { Success = true, Message = "启动命令已接收（示例）" });
        }
        case CommandType.STOPTEST:
        {
            // 在这里调用真实停止逻辑。
            return Task.FromResult(new CommandReply { Success = true, Message = "停止命令已接收（示例）" });
        }
        default:
        {
            return Task.FromResult(new CommandReply { Success = false, Message = "不支持的命令" });
        }
    }
};

await device.ConnectAsync();

以上返回值仅演示格式，不代表真实硬件数据。类库自动填写回复的 CommandId、DeviceId，并保留 DataJson。长时间操作建议启动后及时返回，通过数据上报报告进度，避免停止指令排队。

设备上报数据

连接完成后，在发送按钮或定时任务中调用：

await device.PublishTelemetryAsync(new { Pressure = 101.3, Flow = 2.5 });

自动填写 DeviceId、DateTime 和 PayloadJson。当前源码使用 DateTime.Now，即设备本地时间。方法完成表示 MQTT 发布完成，不代表主机业务已经处理。

主机连接与接收

private MqttHostClient host;

host = new MqttHostClient(settings, "auto-ventilation-host");
host.CommandTimeout = TimeSpan.FromSeconds(10);

host.StatusReceived += status =>
{
    Console.WriteLine($"设备={status.DeviceId} IP={status.IpAddress} 在线={status.IsOnline}");
};

host.TelemetryReceived += data =>
{
    Console.WriteLine($"{data.DeviceId} {data.DateTime} {data.PayloadJson}");
};

host.ReplyReceived += reply =>
{
    Console.WriteLine($"回复={reply.CommandId} 成功={reply.Success} 内容={reply.DataJson}");
};

host.Error += ex =>
{
    Console.WriteLine("主机通信异常：" + ex.Message);
};

await host.ConnectAsync();

事件在连接前注册，防止错过订阅后立即到达的保留状态。主机按 IP 管理设备时，用 status.IpAddress 匹配维护 IP，再将对应 DeviceNumber 更新为 status.DeviceId；此匹配和配置保存属于主机业务，类库不维护设备列表。

主机查询和控制

try
{
    var reply = await host.SendCommandAsync("device01", CommandType.GETCOMPONENT, new { });
    if (!reply.Success)
    {
        Console.WriteLine("查询失败：" + reply.Message);
        return;
    }
    if (string.IsNullOrWhiteSpace(reply.DataJson))
    {
        Console.WriteLine("设备未返回数据");
        return;
    }

    // JToken 同时支持对象和数组。字段结构由设备和主机约定。
    var data = JToken.Parse(reply.DataJson);
    Console.WriteLine(data.ToString());
}
catch (TimeoutException ex)
{
    Console.WriteLine("等待回复超时，设备是否已执行未知：" + ex.Message);
}
catch (OperationCanceledException)
{
    Console.WriteLine("等待已取消或会话已断开");
}
catch (Exception ex)
{
    Console.WriteLine("查询或解析失败：" + ex.Message);
}

其他调用示例：

var range = await host.SendCommandAsync("device01", CommandType.GETRANGE, new { });
var start = await host.SendCommandAsync("device01", CommandType.STARTTEST, new { DurationSeconds = 60 });
var stop = await host.SendCommandAsync("device01", CommandType.STOPTEST, new { });

当前泛型重载需要 parameters 参数；无参数业务用 new { } 传空对象。若协议需要 JSON null，显式使用 SendCommandAsync<object>("device01", CommandType.GETCOMPONENT, null)。当前没有仅接收设备编号和枚举的两参数重载。

SendCommandAsync 的返回值与 ReplyReceived 事件可能是同一条回复，避免重复执行业务。事件还可能收到迟到或重复回复。

命令编号与重试

当前默认 JSON 序列化发送枚举数字，已有成员不能重排，新成员追加到末尾。两端须使用一致的协议；成员名称为 GETRANGE，不是 GETRANG。

重试同一业务时使用完整命令重载并复用 CommandId：

var command = new DeviceCommand
{
    CommandId = Guid.NewGuid().ToString("N"),
    CommandType = CommandType.GETCOMPONENT,
    ParametersJson = "{}",
    ExpiresAtUtc = DateTimeOffset.UtcNow.AddSeconds(30)
};
var reply = await host.SendCommandAsync("device01", command);

重试时复用同一个 command，不重新生成编号。设备实例内缓存执行结果，重启后缓存丢失；同一个编号不得用于不同业务。已过期的查询若需要重新采样，应作为新请求生成新编号。

状态保留消息

主题区分大小写。主机内部通过 + 通配订阅设备状态、数据和回复；本文只使用一个设备。需要隔离不同系统时，两端设置相同且独立的 TopicPrefix。

• 连接成功调用 PublishStatusAsync(true, token)，发布并保留在线状态。
• 正常断开调用 PublishStatusAsync(false, token)，覆盖为离线状态。
• 异常掉线由 Broker 检测后发布保留的离线遗嘱。
• PublishAsync 的 retain 参数通过 WithRetainFlag(retain) 设置；遗嘱通过 WithWillRetain() 设置。
• MQTT 3.1.1 保留消息没有自动过期时间，每个主题只保存最近一条。新的保留消息会覆盖，空载荷保留消息会删除；服务重启能否恢复取决于 Broker 持久化设置。
• 主机重新订阅应收到 Broker 当前保存的状态，无需设备再次连接。普通上报和命令不保留，CleanSession 模式不保证离线期间的消息补发。

若主机重启不显示在线，先确认 StatusReceived 是否触发，再检查状态 IP 是否为空、是否匹配维护 IP，以及两端是否连接相同服务和前缀。不要把界面未更新直接当成 MQTT 未收到。

断开、释放和界面线程

// 分别在设备、主机的异步关闭流程中调用。
await device.DisconnectAsync();
device.Dispose();

await host.DisconnectAsync();
host.Dispose();

普通 DisconnectAsync 后可以再次 ConnectAsync；Dispose 后不能复用。当前无自动重连，掉线由调用方重新连接。连接过程中可以由另一个按钮调用 DisconnectAsync 取消。

host.CancelPendingCommands() 只取消当前命令等待，不撤销已经开始的设备动作。直接调用方法的异常在 await 处捕获，后台异常通过 Error 事件通知。

回调运行在后台线程，单个客户端按顺序处理接收消息。WPF 使用 Dispatcher.BeginInvoke，Avalonia 使用 Dispatcher.UIThread.Post 更新控件。回调应尽快返回，不能在回调内连接、断开、释放当前客户端或同步等待回复。内部取消令牌由类库管理，调用者无需传入。