MqttCommunication 1.0.5

MqttCommunication

设备与主机之间的 MQTT 双向通信类库，适用于 VOC 等设备软件的数据上报、在线状态通知、命令下发及业务回复。

VOC／设备软件（MqttDeviceClient）
                ↕
       MQTT Broker（例如 Mosquitto）
                ↕
主机软件（MqttHostClient）

两端连接同一个 Broker，不是直接连接彼此的 IP。主机软件退出本身不会断开设备与 Broker 的连接；如果 Broker 同时停止，则两端都会受影响。

版本与安装

支持 .NET Framework 4.8、.NET Standard 2.0、.NET 8 和 .NET 10。

dotnet add package MqttCommunication --version 1.0.0.4

本文对应当前仓库源码。项目默认版本为 1.0.0.4，实际发布版本以 NuGet.org 为准。安装旧包不保证包含这里描述的新接口与协议，使用源码项目引用或确认已发布包的版本后再接入。

依赖 MQTTnet 4.3.7.1207 和 Newtonsoft.Json 13.0.3。使用 NuGet 时自动还原依赖；手动分发 DLL 时同时分发依赖及 MqttCommunication.xml。不要混用 NuGet 与另一个版本的本地 DLL 引用。

公共连接接口

MqttDeviceClient 和 MqttHostClient 分别实现 IMqttCommunicationClient，支持统一管理连接状态、错误通知、连接、断开和释放。公共实现复用自 MqttBase，基类不依赖该接口：

using IMqttCommunicationClient client = new MqttDeviceClient("VOC001", settings);
// 也可使用 new MqttHostClient(settings)。
client.Error += error => Console.WriteLine(error.Message);
await client.ConnectAsync();
await client.DisconnectAsync();

设备数据上报、主机命令下发和业务处理器仍通过各自具体类型使用。

连接配置

示例需要以下命名空间；创建处理器类的代码放在类文件中，包含 await 的调用放在异步方法中。客户端应保存为窗口或服务的字段，整个使用期间复用同一实例。

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

设备和主机分别创建配置，使用一致的服务器、端口和主题前缀：

var settings = new MqttConnectionSettings
{
    Host = "192.168.1.100", // 替换为 Broker 地址
    Port = 1883,
    TopicPrefix = "auto-ventilation",
    OperationTimeout = TimeSpan.FromSeconds(10),
    AutoReconnect = true,
    ReconnectInterval = TimeSpan.FromSeconds(5),
    OfflineStatusExpiryInterval = 600
};

配置在客户端构造时读取，修改原 settings 不会动态更新已有实例。当前没有连接总时长限制，也没有新增业务心跳主题或 KeepAlivePeriod 配置属性；协议保活交由 MQTTnet 处理。

设备端接入

实现命令处理接口

public class DeviceCommandHandler : IDeviceCommandHandler
{
    public Task<CommandReply> HandleAsync(DeviceCommand command)
    {
        switch (command.CommandType)
        {
            case "GETCOMPONENT":
                // 示例数据；实际项目应读取设备当前数据。
                return Task.FromResult(new CommandReply
                {
                    Success = true,
                    Message = "读取成功",
                    DataJson = JsonConvert.SerializeObject(new { Components = new[] { "CO", "CO2" } })
                });
            default:
                return Task.FromResult(new CommandReply
                {
                    Success = false,
                    Message = "尚未实现该命令"
                });
        }
    }
}

根据业务补充启动、停止、校准等命令。类库不会替你执行硬件操作；处理器返回的 Success、Message、DataJson 作为业务结果，CommandId、DeviceId 和 CommunicationId 由类库统一填写。未设置处理器时，命令返回失败回复。

注册并连接

字段：

private MqttDeviceClient device;

初始化和连接：

device = new MqttDeviceClient("device01", settings);
device.CommandHandler = new DeviceCommandHandler();
device.Error += ex => Console.Error.WriteLine(ex.ToString());

try
{
    await device.ConnectAsync();
}
catch (OperationCanceledException)
{
    Console.WriteLine("连接已取消");
}
catch (Exception)
{
    // 连接失败已通过 Error 通知；保留实例，让后台继续重试。
    Console.WriteLine("首次连接失败，等待自动重连");
}

AutoReconnect = false 时失败后不会后台重试，可在之后再次调用 ConnectAsync()。构造和参数配置错误需由调用方单独处理，不会因为自动重连而被修复。

设备 DeviceId 保留调用方传入的业务编号，可以重复。类库每个实例自动生成一次 CommunicationId = DeviceId + "_" + GUID，用于所有主题和 MQTT ClientId（device- 加通信编号）。状态、上报数据、回复及离线遗嘱均包含两个编号。两个属性在构造后只读；同一实例断线重连编号不变，重新创建实例或重启软件会生成新通信编号。主机应按 CommunicationId 区分和控制设备，按 DeviceId 显示业务编号。

默认自动获取本机 IPv4，优先选择具有 IPv4 网关的活动网卡。多网卡机器可明确指定上报地址：

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

以上是替代构造方式，不要覆盖仍在运行的客户端。IP 用于状态上报，不是绑定 MQTT 连接的本地网卡；在线、离线和遗嘱使用相同地址。

上报数据

if (device.IsConnected)
{
    try
    {
        await device.PublishAsync(new { Pressure = 101.3, Flow = 2.5 });
    }
    catch (Exception ex)
    {
        // 检查状态后仍可能发生掉线，发布错误在此捕获。
        Console.Error.WriteLine(ex.ToString());
    }
}

类库自动填充设备编号、时间和业务 JSON。当前实际实现使用 DateTime.Now（设备本地时间），不是 UTC。发布完成不表示主机业务已经处理；掉线期间不自动缓存或补发业务数据。

主机端接入

实现消息处理接口

public class MqttHostHandler : IMqttHostHandler
{
    public Task HandleTelemetryAsync(DeviceTelemetry telemetry)
    {
        Console.WriteLine($"数据：{telemetry.DeviceId} / {telemetry.DateTime} / {telemetry.PayloadJson}");
        return Task.CompletedTask;
    }

    public Task HandleStatusAsync(DeviceStatus status)
    {
        Console.WriteLine($"状态：{status.DeviceId} / {status.IpAddress} / 在线={status.IsOnline}");
        return Task.CompletedTask;
    }

    public Task HandleReplyAsync(CommandReply reply)
    {
        Console.WriteLine($"回复：{reply.CommandId} / 成功={reply.Success} / {reply.Message}");
        return Task.CompletedTask;
    }
}

字段：

private MqttHostClient host;

初始化和连接：

host = new MqttHostClient(settings, "auto-ventilation-host");
host.CommandTimeout = TimeSpan.FromSeconds(10);
host.Handler = new MqttHostHandler();
host.Error += ex => Console.Error.WriteLine(ex.ToString());

try
{
    await host.ConnectAsync();
}
catch (OperationCanceledException)
{
    Console.WriteLine("连接已取消");
}
catch (Exception)
{
    Console.WriteLine("首次连接失败，等待自动重连");
}

处理器必须在连接前设置，避免错过刚订阅时收到的保留状态。多个主机实例使用不同的 ClientId。主机按 IP 管理设备时，可用 status.IpAddress 匹配设备记录，再保存对应的 status.DeviceId 和 status.CommunicationId；IP 和业务编号都可能重复，路由及设备记录应以 CommunicationId 区分。设备列表及配置持久化由业务软件负责。

未设置 Handler 不影响底层命令回复匹配，但不会收到这三个业务回调。

下发命令

支持无参数、带业务参数和完整命令三种形式：

// 无参数重载会使用 {} 作为参数 JSON。
// communicationId 取自设备状态的 status.CommunicationId，不能传可能重复的 DeviceId。
var components = await host.SendCommandAsync(communicationId, "GETCOMPONENT");
var start = await host.SendCommandAsync(communicationId, "STARTTEST",
    new { DurationSeconds = 60 });

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

实际调用需要捕获异常并检查业务结果：

try
{
    var reply = await host.SendCommandAsync(communicationId, "GETCOMPONENT");
    if (!reply.Success)
    {
        Console.WriteLine("设备处理失败：" + reply.Message);
        return;
    }
    Console.WriteLine(reply.DataJson);
}
catch (TimeoutException ex)
{
    Console.WriteLine("等待回复超时，设备是否已经执行未知：" + ex.Message);
}
catch (OperationCanceledException)
{
    Console.WriteLine("命令等待已取消或连接已断开");
}
catch (Exception ex)
{
    Console.Error.WriteLine(ex.ToString());
}

• CommandTimeout 默认 10 秒，是单次命令等待期限，不是连接有效期。
• 命令编号和有效期未填写时由主机生成。业务参数和返回 JSON 的字段结构由两端约定。
• HandleReplyAsync 也会收到匹配、迟到或重复的回复。匹配的命令等待先被完成，再调用回调；不要在方法返回值和回调中重复执行业务。
• 重连不会自动重发命令。确需重试同一业务时，复用原 CommandId，不得给不同业务使用同一编号。
• 设备实例内缓存执行任务与结果；断线重连保留缓存，释放实例或程序重启后不保证去重。已过期且完成的记录会被清理，不能把内存缓存当作永久幂等存储。
• host.CancelPendingCommands() 只取消主机等待，不会撤销设备已收到的命令，也不会关闭自动重连。

自定义命令名称

DeviceCommand.CommandType 和 SendCommandAsync 的命令类型参数使用非空字符串，由主机与设备业务共同约定，区分大小写。新增命令无需修改类库，可在业务公共项目中使用字符串常量统一名称。设备自行返回不支持命令的失败回复。

await host.SendCommandAsync(communicationId, "ReadTemperature");

这是协议变更：原来 JSON 中的枚举数字改为命令名称字符串，两端及手工发送工具需要同步更新。旧 CommandType 枚举仅保留定义并标记弃用，不再用于命令 API；数字命令不再映射到原来的枚举成员。

Error 通知与日志

Error 仍为 Action<Exception> 事件，当前没有 IMqttErrorHandler 接口。它是可选通知，不注册不会阻止连接，也不会自动生成文件日志。

client.Error += ex =>
{
    // 替换为软件现有的文件日志工具；ToString() 包含内部异常与堆栈。
    Console.Error.WriteLine(ex.ToString());
};

其中 client 表示已有的 device 或 host 实例。

首次失败有两条通知渠道，避免在 Error 和 catch 中重复记录同一错误。Error 处理器自身抛出的异常会被类库捕获，但仍应使用可靠、及时返回的日志实现。仅更新界面或写控制台不等于已保存日志文件。

自动重连与旧会话清理

默认流程：

连接失败／异常掉线
  → Error 通知
  → 取消旧会话并等待接收任务结束
  → 清理旧连接与队列
  → 等待 ReconnectInterval
  → 重新连接、订阅
  → 设备重新发布在线状态
  → 失败则继续重试

同一实例只有一个后台连接流程。首次连接失败抛出异常后，后台仍可继续重试；不要在 catch 中立即释放客户端。IsConnected 表示连接及订阅等初始化已经完成，窗口一直打开不代表 MQTT 连接正常。

连接进行中重复调用 ConnectAsync() 不会创建第二个重连任务；重连期间的重复调用可能等待恢复。不要在 UI 线程同步阻塞等待。正在主动断开时调用连接会抛出异常，应先等待 DisconnectAsync() 完成。

保留客户端实例、处理器注册、连接配置和设备命令缓存；不删除业务参数、测试结果、文件，不强制终止已开始的设备动作。

重试间隔从清理完成后开始计算，不保证每 5 秒一定发起一次连接。 网络操作超时和清理时间另外计算；若业务回调一直不返回，可能阻塞旧会话清理及后续重连。自动重连不提供离线数据缓存、命令重放或业务心跳超时判断。

在线状态与主题

以下使用默认前缀，{CommunicationId} 为设备上报的完整通信编号：

主题区分大小写；主机使用 + 通配订阅设备状态、上报数据和回复。

• 使用 MQTT 5，Broker 需要支持 MQTT 5；订阅端仍可使用 MQTT 3.1.1。
• 设备成功连接时发布保留的在线状态，不设置过期时间；正常下线和异常离线遗嘱默认 600 秒过期，可通过 OfflineStatusExpiryInterval 配置。
• 同一通信编号再次上线会覆盖旧离线状态；软件重启生成新通信编号，旧主题在离线状态／遗嘱发布后开始计算有效期。突然断电时还需要等待 Broker 检测掉线，不能从进程退出时刻直接计算十分钟。
• Mosquitto 建议配置 retain_expiry_interval 1，每分钟扫描并清理已过期的保留消息。设备重新上线会覆盖旧离线状态，旧消息过期不会删除新的在线状态。
• 历史未设置有效期的保留消息不会自动补上有效期，需单独清理；消息过期也不会主动通知订阅者删除已有记录。
• 异常掉线由 Broker 检测后发布离线遗嘱，检测并非瞬时完成。
• 主机重新连接并订阅后，可以收到 Broker 当前保留的状态；该状态不是主动探测结果。
• Broker 重启后的保留消息恢复取决于其持久化设置；设备重新连接后会再次发布在线状态。
• 当前使用 CleanSession，不保证离线期间业务消息补发。

主机显示离线时先检查两端各自的 IsConnected。主机自身离线时，设备状态应显示未知；两端均在线但界面异常时，检查 HandleStatusAsync 是否收到消息、处理器是否在连接前注册、设备 IP／编号是否匹配，以及服务器和前缀是否一致。

主动断开、释放和 UI 线程

// 分别放入设备端和主机端的异步关闭流程。
await device.DisconnectAsync();
device.Dispose();

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

DisconnectAsync() 停止重试并等待清理完成，之后仍可重新 ConnectAsync()。Dispose() 永久释放实例，不可再次连接。

消息处理器和 Error 回调运行在后台流程中。单个客户端按顺序处理接收消息，应及时返回：

• WPF 使用 Dispatcher.BeginInvoke 更新界面；Avalonia 使用 Dispatcher.UIThread.Post。
• 投递后检查窗口是否关闭及消息来源是否仍为当前客户端，避免旧连接更新新界面。
• 不在回调中调用当前客户端的连接、断开或释放方法。
• 不在回调中等待当前客户端发送的新命令回复，异步 await 同样可能阻塞后续回复处理。
• 长时间硬件任务应由业务层管理，命令返回“已接受”后可通过上报数据报告进度。

Demo 与验证

• Demo/MqttClient01.Net48～MqttClient04.Net48：四个 .NET Framework 4.8 WPF 设备示例。
• Demo/MqttSubscriber.Net10：.NET 10 WPF 主机示例。
• tests/MqttCommunication.ReconnectTests：自动重连测试使用内嵌 Broker；身份隔离与消息过期测试启动独立本地 Mosquitto，不连接业务 Broker。

dotnet build MqttCommunication.sln
# 测试默认查找 D:\Mosquitto\mosquitto.exe，其他路径请设置：
$env:MOSQUITTO_EXE = "D:\Mosquitto\mosquitto.exe"
dotnet run --project tests/MqttCommunication.ReconnectTests

集成测试覆盖连接失败上报与重试、掉线恢复、恢复后发布、重复连接等待、主动停止、再次连接、禁用重连、Error 处理器异常、同业务编号设备命令隔离、字符串自定义命令、状态和遗嘱过期。实际硬件操作、生产网络和界面行为仍需在部署环境联调。

两种过期时间与消息历史

指令和回复使用 QoS 1、Retain=false，目前不设置 MQTT 消息有效期。网页重连不能从 Mosquitto 补取已经转发的历史指令或回复；需要长期审计时，业务服务应在收到时保存到数据库。保留消息每个主题仅保存最新一条，旧版本无有效期的消息不会自动补上过期时间。

Mosquitto 与网页控制台

网页文件为 sites/mqtt-live-console/outputs/mqtt-web.html，可作为单文件部署到 IIS 静态站点，更新文件后按 Ctrl+F5 刷新。使用 MQTT 5 WebSocket，浏览器不能直接连接 TCP 1883；HTTP 页面填写 ws://服务器IP:9001，HTTPS 页面需要 wss://。

网页支持多 Broker 配置、独立连接／重连／订阅、连接切换和删除、消息搜索导出、发布指令、可折叠左栏及连接指南配置复制。设备卡片显示业务 DeviceId，内部按 CommunicationId 区分和发送命令。只显示上报在线的设备；收到离线状态立即移除，并弹出可关闭的五秒提示。不展示离线设备历史，实时消息每个连接最多 500 条、重连清空；未选中的连接仍在后台收消息。浏览器只保存连接名称和地址，不保存连接密码。

内网测试配置示例（需要支持相应选项的 Mosquitto；本机已验证版本为 2.1.2）：

# TCP 监听，设备和主机类库使用
listener 1883 0.0.0.0
listener_allow_anonymous true

# 日志路径需要存在，且服务账户有写入权限
log_dest file D:/Mosquitto/mosquitto.log
log_type error
log_type warning
log_type notice
log_type information
connection_messages true
log_timestamp true
log_timestamp_format %Y-%m-%d %H:%M:%S

# WebSocket 监听，网页使用
listener 9001
protocol websockets
listener_allow_anonymous true

# 每分钟扫描已过期的保留消息，不是所有消息一分钟过期
retain_expiry_interval 1

修改实际加载的配置后重启 Mosquitto，并放行对应端口。匿名配置仅用于可信内网测试，两个监听分别生效；正式部署按环境配置认证和 ACL。上述日志级别不包含全部协议调试日志，排查时可临时使用 log_type all。未配置持久化时，不保证 Broker 重启后恢复保留状态；需要时单独配置 persistence true 和可写的持久化目录。

手动打包

在仓库根目录使用 PowerShell：

dotnet pack .\MqttCommunication\MqttCommunication.csproj -c Release -p:Version=1.0.5 -p:PackageVersion=1.0.5 -o "E:\nuget"

版本号末尾不要加 \；PowerShell 多行续行使用反引号。包版本可通过构建参数覆盖，不需要修改源文件或自动提交版本号。手动上传只使用本机安全注入的 API Key，不将 Key 写入仓库。