MQTT 通讯

Darra PLC 提供 MQTT 客户端功能, 支持 QoS 0/1, 自动重连, 用于物联网数据采集和云端集成。

功能概述

特性	说明
协议版本	MQTT 3.1.1 / 5.0
QoS 等级	QoS 0 (至多一次), QoS 1 (至少一次)
自动重连	断线后自动重连
遗嘱消息	支持 Last Will 配置
保留消息	支持 Retained 消息
认证方式	用户名/密码, TLS 证书

应用场景

连接配置

{
  "protocol": "MQTT",
  "broker": "tcp://192.168.1.200:1883",
  "clientId": "DarraPLC_001",
  "username": "plc_user",
  "password": "********",
  "keepAlive": 60,
  "cleanSession": true,
  "tls": {
    "enabled": false,
    "caFile": "",
    "certFile": "",
    "keyFile": ""
  },
  "lastWill": {
    "topic": "darra/plc/001/status",
    "payload": "{\"online\": false}",
    "qos": 1,
    "retain": true
  }
}

主题与变量映射

将 PLC 变量映射到 MQTT 主题进行发布或订阅:

发布 (PLC → 云端)

PLC 变量	MQTT 主题	QoS	周期
MD100 (温度)	darra/plc/001/temp	0	1s
MD104 (压力)	darra/plc/001/pressure	0	1s
MX0.0 (运行状态)	darra/plc/001/running	1	变化时
MD200 (产量)	darra/plc/001/count	1	变化时

订阅 (云端 → PLC)

MQTT 主题	PLC 变量	QoS	说明
darra/plc/001/cmd/speed	MW500	1	速度设定
darra/plc/001/cmd/start	MX500.0	1	启动命令
darra/plc/001/recipe	MW600~MW699	1	配方下载

数据格式

发布数据默认使用 JSON 格式:

{
  "timestamp": "2026-04-07T10:30:00.000Z",
  "deviceId": "DarraPLC_001",
  "values": {
    "temperature": 85.3,
    "pressure": 2.41,
    "running": true,
    "count": 12580
  }
}

自定义格式

支持通过配置切换数据格式: JSON (默认), 原始字节, CSV。对于带宽受限的场景建议使用原始字节格式。

自动重连

连接断开 → 等待 1s → 重连
失败 → 等待 2s → 重连
失败 → 等待 4s → 重连
...
失败 → 等待 60s → 重连 (最大间隔)
连接成功 → 重新订阅所有主题 → 恢复数据间隔重置为 1s

离线缓存

MQTT 断线期间, 待发布的数据会缓存在内存队列中 (默认最大 10000 条)。重连后按顺序发送。超过队列容量时丢弃最旧的数据。

代码示例

// 创建 MQTT 客户端
var adapter = PLCDriverFactory.Create(ProtocolId.MQTT);
await adapter.ConnectAsync(new ConnectionConfig
{
    Broker = "tcp://192.168.1.200:1883",
    ClientId = "DarraPLC_001",
    Username = "plc_user",
    Password = "secret"
});

// 发布数据
string payload = JsonConvert.SerializeObject(new { temperature = 85.3 });
await adapter.PublishAsync("darra/plc/001/temp", payload, QoS.AtMostOnce);

// 订阅主题
adapter.Subscribe("darra/plc/001/cmd/#", QoS.AtLeastOnce, (topic, message) =>
{
    // 处理接收到的指令
    Console.WriteLine($"收到: {topic} = {message}");
});

MQTT 5.0 特性

相比 3.1.1, MQTT 5.0 引入了:

特性	用途
User Properties	消息携带自定义键值对元数据
Reason Codes	更细的错误码
Session Expiry Interval	会话到期时间
Topic Aliases	长主题用 ID 代替, 节省带宽
Message Expiry Interval	消息到期时间
Shared Subscriptions	负载均衡订阅 ($share/group/topic)
Flow Control	发送方限制未确认消息数量
Request/Response	基于 Correlation Data + Response Topic
Payload Format Indicator	标识是二进制/UTF-8
Content Type	MIME 类型

Darra 默认使用 3.1.1, 配置里可切换到 5.0。

QoS 等级详解

QoS	含义	重传	适用
0	至多一次 (Fire and Forget)	无	高频传感器, 丢失可接受
1	至少一次 (需 PUBACK)	客户端重传	大多数生产数据
2	恰好一次 (4 步握手)	有	关键事件, 财务型

QoS 2 比 QoS 1 慢约 3 倍 (4 次往返), 慎用。

保留消息 (Retained)

发布时设置 Retain = true 的消息, Broker 会保留一份, 新订阅者立即收到:

await adapter.PublishAsync("darra/plc/001/status", "online", QoS.AtLeastOnce, retain: true);

适合状态类主题 (在线/离线, 当前配方, 当前模式)。新接入的监控面板不用等下次推送。

遗嘱 (Last Will)

连接时声明 LWT, 客户端异常断开 (非主动 DISCONNECT) 时 Broker 代为发布:

await adapter.ConnectAsync(new ConnectionConfig {
    ...
    LastWill = new LwtConfig {
        Topic = "darra/plc/001/status",
        Payload = "{\"online\": false}",
        QoS = QoS.AtLeastOnce,
        Retain = true
    }
});

搭配 Retained 消息, 其他监控系统能实时知道 PLC 的上下线。

主题命名规范

推荐分层设计:

<ns>/<line>/<station>/<direction>/<category>/<key>

示例:

darra/line01/station01/telemetry/temperature/z1
darra/line01/station01/telemetry/temperature/z2
darra/line01/station01/command/start
darra/line01/station01/event/alarm/overheat
darra/line01/station01/status/online

通配符:

• + 匹配单层 (darra/+/+/telemetry/#)
• # 匹配多层尾 (darra/line01/#)
• $share/group/... 共享订阅 (MQTT 5)

消息格式

IDE 内置多种序列化格式:

格式	大小	可读性	场景
JSON	中	高	通用, 开发友好
MsgPack	小	低	带宽敏感
CBOR	小	低	IoT 标准
Protobuf	最小	低	强类型契约
Raw 16 字节	固定	极低	超窄带 (LoRa)
Sparkplug B	中	中	工业 MQTT 标准

在连接配置里按主题或全局选择。

Sparkplug B

Sparkplug B 是 Eclipse 基金会定义的工业 MQTT 标准, 在 MQTT 之上规定了:

• 主题结构: spBv1.0/<GroupID>/<MsgType>/<EdgeNodeID>[/DeviceID]
• 消息类型: NBIRTH, NDEATH, NDATA, DBIRTH, DDEATH, DDATA, NCMD, DCMD
• Payload 使用 Protobuf, 含序列号、时间戳、变量列表
• 完整状态模型 (节点上下线、设备上下线)

Darra 的 MQTT 适配器内置 Sparkplug B 支持, 开启后自动按规范收发。上位系统 (Ignition, HiveMQ) 可以零配置接入。

TLS 加密

生产环境建议启用 TLS:

{
  "broker": "ssl://mqtt.example.com:8883",
  "tls": {
    "enabled": true,
    "caFile": "certs/ca.crt",
    "certFile": "certs/client.crt",
    "keyFile": "certs/client.key",
    "verifyServerName": true,
    "minVersion": "TLS1.2"
  }
}

Broker 侧 (Mosquitto / EMQX / HiveMQ) 对应配置服务器证书和客户端 CA。

Broker 选择

Broker	优点	适合
Mosquitto	轻量 (< 5MB), 开源	单机、小规模
EMQX	百万连接, 集群	大规模 IoT
HiveMQ	企业级, 商业支持	工业关键
VerneMQ	Erlang, 分布式	高可用
AWS IoT Core	云托管	云原生
Azure IoT Hub	云托管 + 设备孪生	微软生态
Aliyun IoT	国内云	中国区部署

客户端 ID

clientId 必须唯一, 推荐格式: <产品>_<站点>_<序列号>, 如 DarraPLC_Line01_001。重复 ID 会导致 Broker 踢掉旧连接。

会话持久化

Clean Session 的两种设置:

设置	行为
cleanSession=true	每次连接都是新会话, 断开后订阅和离线消息丢弃
cleanSession=false	断线后保留订阅, Broker 缓存离线消息, 重连后补发

工业场景推荐 false, 避免短暂断网后丢数据。注意 Broker 侧要保持足够的持久化空间。

离线队列

Darra 适配器本地也维护一个发送队列:

• 连接中: 直接发送
• 断线: 入队 (FIFO, 最多 10000 条)
• 重连: 按队列顺序补发
• 队列满: 丢最旧 (FIFO 替换)

队列持久化到磁盘 (可选), 防止 Runtime 重启丢失:

{
  "queue": {
    "maxSize": 10000,
    "persistToDisk": true,
    "persistPath": "C:/Darra/mqtt_queue"
  }
}

诊断

IDE 内置 MQTT 诊断面板:

• 所有订阅主题实时消息流
• 收 / 发字节/秒 统计
• 重连次数 + 断线原因
• 队列深度
• 服务器 Ping 延迟

与数据库协同

常见架构:

PLC → MQTT Publish → Broker → 消息桥 → 数据库
                              ↓
                              SCADA / Dashboard

Broker 侧配置:

• Mosquitto: 插件 + 桥接 plugin
• EMQX: 规则引擎, 直接写入 MySQL / Kafka / ClickHouse
• 云 IoT: Rule → Lambda / Function → 存储

排错

症状	处理
连接被拒绝 Code=5	用户名密码错
连接正常但无消息	主题拼写检查 (区分大小写)
收到消息重复	QoS 1 时 Broker 重试, 客户端需幂等
主题通配符订阅不到	检查 # 必须在末尾, + 是单层
TLS 握手失败	CA 证书不匹配 / 时间不同步
Sparkplug B 节点显示离线	检查 NDEATH 的 bdSeq 是否正确递增