跳到主要内容

Sparkplug B

PlantPulse Edge의 Sparkplug B 发布者(outbound) + 命令接收者(inbound) 手册。以标准 SPB 1.0 主题发布 NBIRTH / DBIRTH / DDATA / NDEATH，并处理主机(SCADA)发送的 NCMD / DCMD，支持标签写入及 rebirth。与 SCADA / HiveMQ / Ignition 等兼容。

项目值
状态阶段 1 (BIRTH/DATA/DEATH) + 阶段 3 (NCMD/DCMD 入站) 已激活。阶段 2 (alias) 可通过开关启用
控制器包plantpulse.app.edge.component.queue.transfer.sparkplug
默认值sparkplug.enable=true (运行激活)
命令接收sparkplug.cmd.enable=true 时订阅并处理 NCMD/DCMD
代理Edge 的 HiveMQ (复用 mqtt.server.*)
认证MQTT username/password (app.properties 的 mqtt.server.user / mqtt.server.password)。2026.05+ install.sh 为每台机器自动生成随机密码。HiveMQ broker 的 auth.properties 由 container-entrypoint 在每次启动时与 app.properties 的 mqtt.* 自动同步 — 操作员只需修改 app.properties
2026.05+ 容器模式的机密

MQTT 账户通过 MQTT_USER / MQTT_PASSWORD_FILE 或 MQTT_PASSWORD 注入。 外部 SCADA (Ignition / HiveMQ Edge 等) 使用 Sparkplug 连接到此机器时，操作员需从 自行管理的 secret 文件中获取密码后使用。

主题架构​

spBv1.0/{group_id}/(N|D){BIRTH|DEATH|DATA|CMD}/{edge_node_id}[/{device_id}]
项目映射
group_idEdgeContext.site_id (例如：SITE_00001)。可通过 sparkplug.group.id 覆盖
edge_node_idEdgeContext.id (例如：EDGE_00303)。可通过 sparkplug.edge.node.id 覆盖
device_idopc_id (例如：OPC_UA_Kepware) — sparkplug.device.strategy=per-opc
metric nametag_id

消息类型​

消息主题负载核心发布时间点
NBIRTHspBv1.0/{g}/NBIRTH/{n}bdSeq metric、seq=0连接后立即
DBIRTHspBv1.0/{g}/DBIRTH/{n}/{d}该 OPC 的所有标签 metric (初始为 null)，seq 增加连接后按 OPC 分别发送
DDATAspBv1.0/{g}/DDATA/{n}/{d}单个 metric (tag_id、value、datatype、ts)，seq 增加队列中收到 1 个 Point 时
NDEATHspBv1.0/{g}/NDEATH/{n}bdSeq metric，无 seq (规范)MQTT will (异常终止) / 断开连接时显式发布
NCMDspBv1.0/{g}/NCMD/{n}主机 → 边缘节点命令 (例如：Node Control/Rebirth=true)主机的用户操作 / 超时 / 状态同步
DCMDspBv1.0/{g}/DCMD/{n}/{d}主机 → 边缘设备 metric 写入 (目标 metric 名称或 alias + 新值)主机将 OPC 的特定标签值更改为设定点时

生命周期 / 序列​

  1. bdSeq — 0 ~ 255 循环，磁盘持久化 (./data/sparkplug/bdseq 默认)。每次会话开始时 +1。
  2. seq — 0 ~ 255 循环。在 NBIRTH 中以 0 开始，DBIRTH/DDATA 均在发布时递增 1。
  3. NDEATH = MQTT Will — 在 CONNECT 时注册为 will，异常终止时代理自动发布。正常终止时 disconnect() 显式发布。

数据类型映射​

APP_TAG.data_type (自由字符串) → MetricDataType (忽略大小写，trim)。

edgeSPB
Float / SingleFloat
DoubleDouble
Byte / Int8Int8
Short / Int16Int16
Int / Integer / Int32Int32
Long / Int64Int64
UInt8/16/32/64UInt8/16/32/64
Boolean / BoolBoolean
String / TextString
DateTime / DateDateTime
其他 / nullString (回退，log.warn)

Point.value 以 String 存储，因此在放入 SPB metric value 之前需按上述类型进行强制转换。 若转换失败，则将 metric 值以 null 发布。

配置 (src/app.properties)​

# --------------------------------------------------------------------------------------
# SPARKPLUG B
# 브로커 접속은 mqtt.server.* 를 그대로 재사용한다.
# --------------------------------------------------------------------------------------
sparkplug.enable=false
sparkplug.group.id=                  # 비우면 site_id
sparkplug.edge.node.id=              # 비우면 EdgeContext.id
sparkplug.device.strategy=per-opc    # per-opc | single (현재 per-opc 만 구현)
sparkplug.qos=0                      # DDATA 빈번 → 0 권장
sparkplug.bdseq.path=                # 비우면 ./data/sparkplug/bdseq
sparkplug.alias.enable=true          # Phase 2 토글 (현 무영향)
sparkplug.cmd.enable=false           # Phase 3 토글 (현 무영향)

组件​

文件角色
SparkPlugQueueTransfer.javaQueueTransfer 实现。仅暴露 init/trans/close
SparkPlugSession.java生命周期 (CONNECT+will / NBIRTH / DBIRTH / DDATA / disconnect-NDEATH / seq / bdSeq)
SparkPlugConfig.java属性 → 配置对象 (代理复用 mqtt.server.*)
SparkPlugTopicBuilder.java主题字符串构建器
SparkPlugDataTypeMapper.javadata_type → MetricDataType + value 强制转换
SparkPlugBdSeqStore.javabdSeq 磁盘持久化 (0 ~ 255 循环)
PointQueueProcessorFactory.java在 transfer_list 中添加 SparkPlugQueueTransfer

依赖 jar​

添加到 WebContent/WEB-INF/lib/:

jar角色
tahu-core-1.0.14.jarEclipse Tahu Sparkplug B 负载模型 + 编码器
protobuf-java-3.25.5.jartahu 的 protobuf 依赖
commons-compress-1.27.1.jartahu 的 commons-compress 依赖
库同步

WEB-INF/lib 的 jar 在构建时从 Nexus 物化，因此不提交。新依赖需在 build.gradle 中添加坐标。

验证 (开发服务器 / 2026-05-06)​

使用 paho-mqtt 订阅者捕获 spBv1.0/#:

[SUMMARY] total=511
  NDEATH: 1   ← 재시작 시 will 자동 발행
  NBIRTH: 1   ← seq=0
  DBIRTH: 10  ← OPC 10개 모두, seq=1..10
  DDATA: 499  ← 13초간 ≈ 38 msgs/s

DBIRTH metric 数 = APP_TAG 中的对应 OPC 标签数完全一致 (OPC_UA_Kepware=11，OPC_00303=56， OPC_LS_XBM_0001=28 等)。

DDATA 主题示例:

spBv1.0/SITE_00001/DDATA/EDGE_00303/OPC_UA_Kepware  bytes=44  seq=125  ts=...

负载解码验证工具:

MQTT_USER="${MQTT_USER:-edge}" \
MQTT_PASSWORD="$(tr -d '\r\n' < /run/secrets/mqtt-password)" \
python3 /tmp/spb_sub.py 100.106.92.21 1883 "$MQTT_USER" "$MQTT_PASSWORD" 60

(此脚本为临时文件，不包含在构建产物中。)

阶段 3 — 命令(NCMD/DCMD)接收行为​

当 sparkplug.cmd.enable=true 时，SparkPlugSession 自动订阅以下主题。

spBv1.0/{group}/NCMD/{edge_node}
spBv1.0/{group}/DCMD/{edge_node}/+        # 모든 device 와일드카드

收到的消息通过静态方法 parseCmdTopic(String) → CmdTopic 验证后进行分支处理。

命令处理
NCMD Node Control/Rebirth=truerebirthAsync() — 重新发布 NBIRTH + 重新发布所有 OPC 的 DBIRTH + (alias 激活时) 更新 aliasReverseMap (refreshAliasReverseMap())
DCMD metric 写入通过 metric name 或 alias 反向查找 tag_id → 调用对应 OPC 的 driver write(tagId, value) → 将结果反映到下一个 DDATA 周期

主题验证 (parseCmdTopic)​

输入结果
spBv1.0/Plant1/NCMD/EDGE_00303OK (NCMD，deviceId=null)
spBv1.0/Plant1/DCMD/EDGE_00303/OPC_LSOK (DCMD，deviceId=OPC_LS)
null / 空字符串 / 少于 4 段拒绝 (null)
spBv1.0/Plant1/NDATA/EDGE_00303拒绝 (出站类型)
其他 namespace (spBv2.0/...)拒绝
spBv1.0/Plant1/DCMD/EDGE_00303 (4 段，缺少 deviceId)拒绝
6+ 段忽略额外段 (仅使用 4-5)

→ 单元测试 SparkPlugCommandTopicTest 保证上述 10 种情况。

rebirth 安全网​

修复了 rebirthAsync() 在重新发布 NBIRTH/DBIRTH 后遗漏更新 aliasReverseMap，导致后续 DCMD 使用过时 alias 进行查找的缺陷 — 添加了 refreshAliasReverseMap() 调用 (subscribeCommands() 也使用同一辅助函数，DRY)。

路线图​

阶段内容状态
1 — BIRTH/DATA/DEATHNBIRTH / DBIRTH / DDATA / NDEATH(will)✅ 完成
2 — alias 优化在 NBIRTH/DBIRTH 时创建 tag_id → alias(Long) 表，DDATA 仅发布 alias (负载减少 30 ~ 40%)。开关：sparkplug.alias.enable可选 (开关)
3 — NCMD/DCMD 入站处理主机发送的 metric 写入命令 + 节点控制/Rebirth✅ 完成 (sparkplug.cmd.enable=true)
4 — STATE / 运行可见性订阅 STATE/{primary_host_id} → 主节点 OFFLINE 时暂停 DDATA。在 /api/v1/edge 响应中添加 sparkplug 部分 (connected / bdSeq / seq / devices)TBD

已知限制​

  • DDATA value=null 时未设置 isNull 标志：当前仅调用 Metric.setValue(null)。根据规范，显式声明 isNull=true 更安全 → 后续补充候选。
  • 单个 metric 每条 DDATA：队列每笔传输 1 条，因此不进行批量发布优化。若频率更高， 可发展为批量 DDATA。
  • bdSeq 路径默认为工作目录相对路径：生产服务器建议显式绝对路径 (/opt/kopens/.../data/sparkplug/bdseq)。
  • DCMD 写入失败响应未实现：当前即使 driver write() 执行失败，也不向主机发送 NDATA/DDATA 结果通知。 后续考虑添加 result_status metric。