实战: Modbus TCP → MQTT

Name: Darra Software PLC
Author: Darra

概述

最典型的数字化改造场景: 现场有多台 Modbus RTU 智能电表挂在 RS485 总线上, 通过一台 Modbus RTU/TCP 网关接入以太网。现在要把电表数据实时上传云端 (Node-RED / Thingsboard / 阿里云 IoT)。

本章给出 DarraRT 一键打通 Modbus TCP → PLC → MQTT → 云平台的全链路配置, 含故障演练与 QoS 调优。

适用场景

典型场景	设备数量	上云频率	本文是否适用
能源监控 (电表/水表/气表)	10-100 台	1 min	YES
厂区环境 (温湿度/PM)	5-50 点	10 s	YES
设备状态 (变频器/电机)	1-20 台	1 s	YES
产线计件 (光电/计数器)	10-200 点	事件驱动	YES
毫秒级控制	—	—	NO (走 EtherCAT)

硬件假设

┌───────┐    ┌───────┐    ┌───────┐
│ 电表 1 │    │ 电表 2 │    │ 电表 3 │
└───┬───┘    └───┬───┘    └───┬───┘
    │ RS485 总线 (Modbus RTU)     │
    └──────┬─────┴──────┬─────────┘
           │            │
     ┌─────▼────────────▼──────┐
     │  Modbus RTU/TCP 网关      │
     │  (例: 有人串口服务器)      │
     │  IP: 192.168.1.100:502   │
     └─────────┬────────────────┘
               │ 以太网
               │
     ┌─────────▼────────────────┐
     │ DarraRT PLC (Modbus Master) │
     │  + Service + 网关           │
     └─────────┬────────────────┘
               │ TLS (MQTTS:8883)
     ┌─────────▼────────────────┐
     │  MQTT Broker (EMQX/Mosq)  │
     └─────────┬────────────────┘
               │ 订阅
     ┌─────────▼────────────────┐
     │  Node-RED / Thingsboard   │
     └──────────────────────────┘

3 台电表的 Modbus 从站号分别为 1/2/3, 网关把 RTU 帧封装进 TCP, 对 PLC 看就是 3 个 Unit ID。

前置条件

DarraRT Service ≥ 2.4
Modbus RTU/TCP 网关能 ping 通 (ping 192.168.1.100 稳定)
MQTT Broker 账号密码已申请, 允许用户名 plc_line1 发布 factory/line1/#
防火墙放行 502 / 1883 / 8883
电表寄存器表: 电压 40001 (INT16, 0.1V), 电流 40003 (INT32, 0.001A), 功率 40007 (INT32, W), 电度 40017 (INT32, 0.01kWh)

详细步骤

步骤 1: 配置 PLC 作为 Modbus TCP Master

IDE 打开项目, 通讯配置 → 新增 Modbus TCP 客户端:

# modbus_client.yaml (IDE 生成)
clients:
  - id: meter_gw
    type: modbus_tcp
    host: 192.168.1.100
    port: 502
    timeout_ms: 500
    retries: 3
    poll_interval_ms: 1000
    slaves:
      - unit: 1
        name: meter1
        requests:
          - fc: 3            # Read Holding Registers
            start: 40001
            count: 20
            target: DB_Meter1
      - unit: 2
        name: meter2
        requests:
          - fc: 3
            start: 40001
            count: 20
            target: DB_Meter2
      - unit: 3
        name: meter3
        requests:
          - fc: 3
            start: 40001
            count: 20
            target: DB_Meter3

步骤 2: 定义数据块映射到 MW 区

// DB_Meter 数据块 (每台电表一个)
TYPE T_Meter :
STRUCT
    voltage_raw  : INT;    // 40001, 0.1V
    _reserved1   : INT;    // 40002 (占位)
    current_raw  : DINT;   // 40003-40004, 0.001A
    power_raw    : DINT;   // 40007-40008, W
    _reserved2   : ARRAY[0..7] OF INT;  // 40009-40016
    kwh_raw      : DINT;   // 40017-40018, 0.01kWh

    // 运算后
    voltage_V    : REAL;
    current_A    : REAL;
    power_kW     : REAL;
    energy_kWh   : REAL;
    online       : BOOL;
    last_update  : TIME;
END_STRUCT
END_TYPE

VAR_GLOBAL
    DB_Meter1 : T_Meter;
    DB_Meter2 : T_Meter;
    DB_Meter3 : T_Meter;
END_VAR

OB1 中做尺度变换:

// OB1 (200ms 扫描)
DB_Meter1.voltage_V   := INT_TO_REAL(DB_Meter1.voltage_raw) * 0.1;
DB_Meter1.current_A   := DINT_TO_REAL(DB_Meter1.current_raw) * 0.001;
DB_Meter1.power_kW    := DINT_TO_REAL(DB_Meter1.power_raw) * 0.001;
DB_Meter1.energy_kWh  := DINT_TO_REAL(DB_Meter1.kwh_raw) * 0.01;

// 在线检测: 如果 last_poll_time 在 3s 内则认为在线
DB_Meter1.online := ModbusClient_IsHealthy('meter_gw', 1);

步骤 3: 配置 MQTT Publisher

# gateway.yaml (同 [协议桥接总览](./protocol-gateway.md))
sinks:
  - id: iot_mqtt
    type: mqtt
    url: ssl://iot.example.com:8883
    client_id: plc_line1
    username: plc_line1
    password: ${MQTT_PASS}    # 从环境变量读
    keepalive: 30
    clean_session: false      # 持久会话
    qos_default: 1
    ca_cert: ./certs/ca.pem
    client_cert: ./certs/client.pem
    client_key: ./certs/client.key

    lwt:
      topic: factory/line1/status
      payload: '{"state":"offline","ts":0}'
      qos: 1
      retain: true

    buffer:
      mode: memory_then_disk
      memory_size: 50000
      disk_path: ./spool/mqtt
      disk_max_mb: 200

routes:
  - from: plc_main.DB_Meter1.power_kW
    to: iot_mqtt
    topic: factory/line1/meter1/power
    qos: 1
    retain: false
    payload_template: |
      {"ts":{{now_ms}},"v":{{value}},"unit":"kW","quality":{{quality}}}
    throttle_ms: 1000
    deadband: 0.05           # 50W 死区

  - from: plc_main.DB_Meter1.energy_kWh
    to: iot_mqtt
    topic: factory/line1/meter1/energy
    qos: 1
    retain: true             # 累积量 retain, 新订阅立即拿到
    payload_template: |
      {"ts":{{now_ms}},"v":{{value}},"unit":"kWh"}
    throttle_ms: 60000       # 每分钟一次足够

  # 批量相同规则给 meter2 / meter3
  - from: plc_main.DB_Meter2.*
    to: iot_mqtt
    topic_template: "factory/line1/meter2/{{var_name}}"
    qos: 1
    throttle_ms: 1000

  - from: plc_main.DB_Meter3.*
    to: iot_mqtt
    topic_template: "factory/line1/meter3/{{var_name}}"
    qos: 1
    throttle_ms: 1000

  # 上线消息 (对应 LWT)
  - from: internal.gateway_started
    to: iot_mqtt
    topic: factory/line1/status
    qos: 1
    retain: true
    payload_template: '{"state":"online","ts":{{now_ms}}}'

步骤 4: QoS / Retain / LWT 说明

项	取值	含义
QoS 0	At most once	丢失不补, 适合遥测流, 如瞬时电流
QoS 1	At least once	可能重复, Broker 确认前不删, 适合计数
QoS 2	Exactly once	四次握手, 慢, 适合关键命令
Retain	最新一条存 Broker	新订阅者立即收到, 适合状态 / 累积量
LWT	断线遗嘱	掉线时 Broker 替你发, 用来通知离线

原则: 功率流量用 QoS 0, 电度累积用 QoS 1 + Retain, 控制命令用 QoS 2, 在线状态用 LWT + Retain。

步骤 5: 断线重连与消息队列

网关 SDK 内部实现了:

指数退避: 1s → 2s → 4s → ... → 60s 封顶
内存队列: 默认 5 万条, 溢出落盘到 ./spool/mqtt
持久会话: clean_session=false, Broker 保留未确认 QoS 1/2 消息
发布确认: QoS 1 收到 PUBACK 才清队列, 超时重发

Publish 失败 → 入队 → 重连成功 → Flush 队列 → 按序重发

步骤 6: Node-RED 订阅示例

// Node-RED Function 节点: 解析电表功率
const msg_ = {
    topic: msg.topic,
    payload: {
        meter: msg.topic.split('/')[2],
        power_kW: msg.payload.v,
        ts: msg.payload.ts
    }
};
return msg_;

Node-RED 流: MQTT in (factory/line1/+/power) → Function → InfluxDB out → Grafana 可视化。

步骤 7: Thingsboard 订阅

Thingsboard 要求设备发 JSON 到 v1/devices/me/telemetry, DarraRT 网关改 topic:

routes:
  - from: plc_main.DB_Meter1.power_kW
    to: iot_mqtt
    topic: v1/devices/me/telemetry
    payload_template: |
      {"meter1_power_kW":{{value}},"ts":{{now_ms}}}

并在 Broker 用每个电表独立 MQTT 会话或 Thingsboard Gateway 子设备协议。

排错表

现象	原因	处理
Modbus 读不到	网关 IP/端口错; Unit ID 错	`telnet 192.168.1.100 502` 通吗
寄存器值错位	字节序 (大小端)	切换 `byte_order: big/little/word_swap`
MQTT 连不上	证书过期 / 用户名错	日志搜 `MQTT connack code=5`
发布慢	QoS 2 + 网络差	降到 QoS 0/1
丢数	缓冲区溢出	加大 `memory_size` 或开磁盘 spool
Retain 没用	Broker 不支持 / 权限	看 Broker 日志
LWT 没触发	Keepalive 太大	降到 30s 以下

故障演练

演练	预期
拔网线 60s 再插	网关断线 → 消息入队 → 重连 → 按序重发
Broker 重启	持久会话恢复, QoS 1 消息不丢
电表掉电	`online=FALSE`, 停发遥测, 但 `status=offline` retain 发布
时间跳变	消息 ts 异常, 用 NTP 同步

高级技巧

稀疏发布: 功率平稳时 5 min 发一次, 波动大时 1s 发, 用 dynamic_throttle
批量主题: 单条消息多字段 factory/line1/meter1/all 比每点一条省 80% 流量
冷备: 第二台 MQTT Broker 作 standby_url, 主断自动切
端到端验证: 发布时打印 Publish 耗时, Broker 侧 $SYS/broker/load/publish/received/1min
加密流量: 强制 TLS 1.3, 禁 ssl_allow_cleartext: false

概述​

适用场景​

硬件假设​

前置条件​

详细步骤​

步骤 1: 配置 PLC 作为 Modbus TCP Master​

步骤 2: 定义数据块映射到 MW 区​

步骤 3: 配置 MQTT Publisher​

步骤 4: QoS / Retain / LWT 说明​

步骤 5: 断线重连与消息队列​

步骤 6: Node-RED 订阅示例​

步骤 7: Thingsboard 订阅​

排错表​

故障演练​

高级技巧​

相关文档​

概述