MQTT 3.1.1

Zeus.Protocols.Mqtt 面向消息型设备、边缘网关和工业物联网 Broker。它把 MQTT 会话放在 Zeus 通道之上，支持发布/订阅、QoS、保留消息、遗嘱、自动保活和断线重连；主题消息也可以直接映射成 Zeus 点表。

安装

dotnet add package Zeus.Communications
dotnet add package Zeus.Protocols.Mqtt

无硬件联调

虚拟 Broker 可以在本地验证连接、订阅、发布和点表写回：

using Zeus;

var memory = new MqttBrokerMemory();
memory.SetText("factory/temperature", "25.3");
memory.SetText("factory/running", "true");

await using var app = ZeusHost.Create(builder =>
{
    builder.AddAcquisition(TimeSpan.FromMilliseconds(500));
    builder.AddVirtualChannel("mqtt-link", new MqttBrokerResponder(memory));
    builder.AddMqtt(
        "gateway",
        "mqtt-link",
        new MqttOptions { ClientId = "zeus-mqtt-demo" },
        points: map => map
            .Double("temperature", "factory/temperature")
            .Boolean("running", "factory/running")
            .Double("setpoint", "factory/setpoint")
            .Writable("setpoint"));
});

await app.StartAsync();
await Task.Delay(600);
Console.WriteLine(app.Points.Get<double>("temperature"));

await app.Points.WriteAsync("setpoint", 18.6);

MqttBrokerMemory 保存虚拟 Broker 的保留消息。发布空载荷并设置 retain: true 会删除对应的保留消息。

连接真实 Broker

MQTT 通常使用 TCP 1883 端口。只替换通道，设备和点表代码可以保持不变：

await using var app = ZeusHost.Create(builder =>
{
    builder.AddTcpClient("mqtt-link", "192.168.1.20", 1883);
    builder.AddMqtt(
        "gateway",
        "mqtt-link",
        new MqttOptions
        {
            ClientId = "zeus-gateway-01",
            Username = "zeus",
            Password = "secret",
            KeepAliveSeconds = 60,
            AutomaticKeepAlive = true,
            AutomaticReconnect = true
        },
        points: map => map
            .Double("temperature", "factory/temperature")
            .Writable("temperature"));
});

await app.StartAsync();

QoS、保留消息和遗嘱

客户端发布默认为 QoS 0，也可以明确使用 QoS 1 或 QoS 2：

var gateway = app.Devices.Get<MqttDevice>("gateway");

await gateway.PublishTextAsync(
    "factory/status",
    "online",
    MqttQualityOfService.AtLeastOnce,
    retain: true);

await gateway.Client.SubscribeAsync(
    "factory/#",
    MqttQualityOfService.ExactlyOnce);

var message = await gateway.Client.WaitForMessageAsync("factory/status");

QoS 1 使用 PUBACK，QoS 2 使用 PUBREC / PUBREL / PUBCOMP。订阅过滤器支持 + 和 #，发布主题和点表主题不能包含通配符。

遗嘱需要同时配置主题和载荷：

var options = new MqttOptions
{
    ClientId = "zeus-gateway-01",
    WillTopic = "factory/status",
    WillPayload = "offline"u8.ToArray(),
    WillQualityOfService = MqttQualityOfService.AtLeastOnce,
    WillRetain = true
};

点表映射

MqttPointMap 支持常用消息载荷类型：

API	载荷	说明
`Text`	UTF-8 文本	原样转成 `string`
`Boolean`	`true` / `false` 或 `1` / `0`	转成 `bool`
`Int32` / `Int64`	十进制文本	转成整数
`Double`	不变文化格式的数字文本	转成 `double`
`Bytes`	原始字节	转成 `byte[]`

数值点可以设置报警限；可写点发布时会使用该点的 QoS 和 retain 配置：

points: map => map
    .Double("temperature", "factory/temperature",
        new PointAlarmLimits(0, 80))
    .WithQualityOfService("temperature", MqttQualityOfService.AtLeastOnce)
    .Retained("temperature")
    .Writable("temperature")

JSON 配置

{
  "acquisition": { "intervalMilliseconds": 500, "pollImmediately": true },
  "channels": [
    { "name": "mqtt-link", "type": "virtual", "responder": "mqtt" }
  ],
  "devices": [
    {
      "name": "gateway",
      "channel": "mqtt-link",
      "type": "mqtt",
      "mqttClientId": "zeus-json-gateway",
      "mqttKeepAliveSeconds": 60,
      "mqttAutomaticKeepAlive": true,
      "mqttAutomaticReconnect": true,
      "points": [
        {
          "name": "temperature",
          "topic": "factory/temperature",
          "dataType": "double",
          "mqttQos": "1",
          "mqttRetain": true,
          "lowAlarmLimit": 0,
          "highAlarmLimit": 80
        },
        {
          "name": "running",
          "topic": "factory/running",
          "dataType": "boolean"
        },
        {
          "name": "setpoint",
          "topic": "factory/setpoint",
          "dataType": "double",
          "writable": true
        }
      ]
    }
  ]
}

设备类型可以写 mqtt、mqtt311 或 mqtt-3-1-1。虚拟通道的 responder 写 mqtt；真实 Broker 则把通道改为 tcp 并填写 host 和 port。

连接问题排查

现象	首先检查
CONNECT 超时	TCP 地址/端口、防火墙和 Broker 是否监听
CONNACK 被拒绝	ClientId、用户名密码、Broker 是否允许 MQTT 3.1.1
收不到消息	主题过滤器、Broker ACL、订阅 QoS 和 `$` 系统主题规则
点表没有值	载荷是否符合点的 `dataType`，以及首次采集是否已经完成
断线后未恢复	`AutomaticReconnect`、通道是否重新进入 `Open`、重连退避参数

现场排查时可打开通道报文追踪，确认 CONNECT、SUBSCRIBE、PUBLISH 和确认报文是否符合对端实现。

下一步：JSON 配置、通信报文追踪或虚拟通道。

安装​

无硬件联调​

连接真实 Broker​

QoS、保留消息和遗嘱​

点表映射​

JSON 配置​

连接问题排查​

安装