MqttCommunication 1.0.0.1

MqttCommunication：设备与主机通信

MqttCommunication 是基于 .NET Standard 2.0 的 MQTT 双向通信类库，可供 .NET Framework 4.8 设备程序和 .NET 10 主机程序引用。

本文以一台设备 device01 和一台主机 auto-ventilation-host 为例。设备与主机分别连接同一个 Mosquitto 服务，通过服务转发消息。

设备 device01  ←→  Mosquitto :1883  ←→  主机 auto-ventilation-host
                 数据上报、状态、命令和回复

1. 连接准备

• 启动 Mosquitto，确保客户端能够访问其 TCP 1883 端口。
• 同机测试时，两端使用 localhost:1883。
• 分别运行在两台电脑时，两端填写 Mosquitto 所在电脑的地址，例如 192.168.7.49；设备端的 localhost 指设备自己所在的电脑。
• 两端使用相同的 TopicPrefix；若服务启用认证，分别填写有效的用户名和密码。
• 主机发送命令的目标编号必须是 device01，与设备构造时的编号一致。

设备内部 MQTT ClientId 为 device-device01，主机示例为 auto-ventilation-host，不能重复。每端创建一个客户端实例，连接期间一直复用。

2. 引用类库

在设备和主机项目中分别添加项目引用，路径按实际目录调整：

<ItemGroup>
  <ProjectReference Include="..\MqttCommunication\MqttCommunication.csproj" />
</ItemGroup>

项目引用会引入 MQTTnet 和 Newtonsoft.Json 依赖。需要分发 NuGet 包时，在解决方案目录执行：

dotnet pack .\MqttCommunication -c Release -o .\artifacts

3. 两端连接配置

以下配置分别写在设备和主机程序中：

var settings = new MqttConnectionSettings
{
    Host = "localhost",           // 跨电脑时改为 Mosquitto 所在电脑的 IP。
    Port = 1883,
    TopicPrefix = "auto-ventilation",
    OperationTimeout = TimeSpan.FromSeconds(10)
    // Username = "用户名",
    // Password = "密码"
};

构造客户端时读取配置。修改服务器地址或端口后，需要使用新配置重新创建实例。

4. 设备端用法（.NET Framework 4.8）

引入命名空间：

using System;
using System.Threading.Tasks;
using MqttCommunication;

在窗口或业务类中保存字段：

private MqttDeviceClient device;

将下面代码放入初始化连接的 async Task 方法中，settings 使用上一节配置。先注册命令处理器，再连接：

device = new MqttDeviceClient("device01", settings);

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

device.CommandHandler = command =>
{
    switch (command.Type)
    {
        case "startTest":
        {
            // 在这里调用实际启动逻辑。
            // command.ParametersJson 是主机下发参数的 JSON 字符串。
            return Task.FromResult(new CommandReply
            {
                Success = true,
                Message = "启动命令已接收（示例）"
            });
        }
        case "stopTest":
        {
            // 在这里调用实际停止逻辑。
            return Task.FromResult(new CommandReply
            {
                Success = true,
                Message = "停止命令已接收（示例）"
            });
        }
        default:
        {
            return Task.FromResult(new CommandReply
            {
                Success = false,
                Message = "不支持的命令：" + command.Type
            });
        }
    }
};

await device.ConnectAsync();

连接成功后，类库自动发布在线状态。需要上报数据时，在发送按钮或定时任务的异步方法中调用：

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

类库自动填写设备编号、UTC 时间和业务 JSON，并发布到数据主题。回复的 CommandId 和 DeviceId 也由类库填写，调用方只需返回执行结果。

示例不操作真实硬件。长时间测试建议在启动动作后及时返回，后续进度通过数据上报发送，避免停止命令排队等待整个测试结束。

5. 主机端用法（.NET 10）

引入 System 和 MqttCommunication，在窗口或业务类中保存字段：

private MqttHostClient host;

在初始化连接的异步方法中使用相同服务器和主题前缀，先注册事件，再连接：

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

host.StatusReceived += status =>
{
    Console.WriteLine(status.DeviceId + "：" + (status.IsOnline ? "在线" : "离线"));
};

host.TelemetryReceived += data =>
{
    Console.WriteLine(data.DeviceId + " "
        + data.DateTime.ToLocalTime().ToString("HH:mm:ss") + " "
        + data.PayloadJson);
};

host.ReplyReceived += reply =>
{
    Console.WriteLine("收到回复：" + reply.CommandId + " " + reply.Message);
};

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

await host.ConnectAsync();

设备连接后，在主机发送按钮的异步方法中下发命令并等待业务回复：

try
{
    var reply = await host.SendCommandAsync(
        "device01", "startTest", new { DurationSeconds = 60 });

    Console.WriteLine(reply.Success ? "设备处理成功" : "设备处理失败");
    Console.WriteLine(reply.Message);
}
catch (TimeoutException ex)
{
    // 超时表示未及时收到业务回复，不能据此认定设备没有执行。
    Console.WriteLine(ex.Message);
}
catch (OperationCanceledException)
{
    Console.WriteLine("已取消等待或连接已断开。");
}
catch (Exception ex)
{
    Console.WriteLine("发送失败：" + ex.Message);
}

停止命令使用同一个设备编号：

var reply = await host.SendCommandAsync("device01", "stopTest", new { });

SendCommandAsync 返回的结果与 ReplyReceived 可能对应同一条回复。事件还会收到迟到或重复的回复，不要在两个入口重复执行同一项业务。

6. 断开与释放

在各自程序的异步方法中断开：

// 设备程序。
await device.DisconnectAsync();

// 主机程序。
await host.DisconnectAsync();

普通断开后，同一个实例可以再次调用 ConnectAsync()。永久不用时，先等待断开完成，再调用该实例的 Dispose()。

连接过程中也可以由另一个按钮调用 DisconnectAsync() 取消连接。主机仅取消命令等待时调用：

host.CancelPendingCommands();

取消等待和断开通信都不会撤销设备已经开始执行的动作。连接、发布和发送命令等直接调用的异常通过返回的 Task 抛出，需要在调用处捕获；后台错误通过 Error 事件通知。

7. WPF 界面调用

连接和发送按钮使用 await 调用上述方法。通信事件运行在后台线程，更新控件时切回 UI 线程，例如：

host.StatusReceived += status =>
{
    Dispatcher.BeginInvoke(new Action(() =>
    {
        // StatusText 为示例控件名，替换为自己的控件。
        StatusText.Text = status.IsOnline ? "在线" : "离线";
    }));
};

事件应及时返回。不要在通信回调中连接、断开、释放当前客户端或同步等待命令回复；连接生命周期由窗口按钮和关闭流程管理。无需向类库传入 CancellationTokenSource 或 CancellationToken。

现有界面联调只需启动 MqttClient01.Net48 和 MqttSubscriber.Net10：两端连接同一个服务，设备点击发送数据，主机填写目标 device01 后发送启停命令。

8. 主题与通信行为

主题区分大小写，由类库统一生成。本文只使用一台设备，当前主机内部仍采用设备通配订阅；若同一服务、同一前缀下存在其他设备，也会收到它们的消息。需要隔离这对设备与主机时，为两端设置相同且独立的 TopicPrefix。

• 使用 MQTT 3.1.1、QoS 1 和 UTF-8 JSON。MQTT 确认表示协议层交付，不代表设备业务执行成功，应以 CommandReply 为准。
• 当前没有自动重连，掉线后由调用方重新连接。
• 状态为保留消息，主机后连接也能获得 Broker 保存的最近状态；异常离线状态由遗嘱发布，需要等待 Broker 检测到断开。
• 普通数据和命令不保留，使用 CleanSession，主机离线期间的数据不会自动补发。
• 命令按 CommandId 在当前设备实例内去重。重试同一业务需使用 SendCommandAsync(deviceId, DeviceCommand) 并复用 CommandId；去重记录不跨程序重启。
• 单个客户端按队列顺序处理消息，耗时事件会延迟后续消息。设备业务自身的硬件超时应由业务代码处理。