Sparkplug B 驱动

PlantPulse Edge 的 Sparkplug B 驱动是 MQTT 3.1.1 客户端 + Sparkplug B 负载 Protobuf 解码器 的原生 Java 实现。它不依赖外部 paho/tahu jar 包，因此在单个 jar 部署时无需额外配置即可运行。

源码：plantpulse.driver.protocol.sparkplug.*

类	职责
`SparkplugBDriver`	`ProtocolDriver` 实现 — connect/read/write/close, alias·metric 缓存
`SparkplugMqttClient`	MQTT 3.1.1 自包含客户端 (CONNECT / SUBSCRIBE / PUBLISH / PINGREQ)
`SparkplugBPayload`	Sparkplug B 负载 protobuf 线性编码器/解码器 (varint / fixed32 / fixed64 / length-delim)

操作流程

broker ──PUBLISH──> SparkplugMqttClient ──onMessage(topic, payload)──> SparkplugBDriver
                                                                       │
                                                                       ├─ topic 파싱 (spBv1.0/G/<msgType>/edge[/dev])
                                                                       ├─ Payload 디코드 → metric list
                                                                       ├─ NBIRTH/DBIRTH 시 alias→name 등록
                                                                       └─ metricCache[edge[/dev]/name] = formatValue()

PlantPulse 폴러 ──read(addr)──> metricCache[addr]   (in-memory hit, 외부 RTT 없음)
                  write(addr, value) ──encode──> NCMD/DCMD publish → broker

选项

键	默认值	说明
`group-id`	`default`	Sparkplug 组 ID。订阅时作为通配符的一部分使用。驱动会过滤掉组不匹配的消息。
`edge-node-id`	`+`	边缘节点过滤器。如果为 `+` 则包含组内所有节点。
`keep-alive`	`60`	MQTT 保持连接秒数。为 0 时禁用 ping。
`connect-timeout`	`5000`	TCP 连接超时毫秒数。

订阅主题由 spBv1.0/<group-id>/+/<edge-node-id>/# 模式构成（+ 是 message_type，# 是设备部分的通配符）。

Sparkplug 数据类型映射

SparkplugBPayload.Metric.formatValue() 会根据在线数据类型将其转换为合适的字符串。

Sparkplug 数据类型	存储字段	格式化结果
Int8 / Int16 / Int32	`intValue` (varint)	有符号整数 (`Integer.toString`)
UInt8 / UInt16 / UInt32	`intValue`	无符号 (`& 0xFFFFFFFFL`)
Int64 / DateTime	`longValue`	有符号长整数
UInt64	`longValue`	`Long.toUnsignedString`
Float	`floatValue` (fixed32)	`Float.toString`
Double	`doubleValue` (fixed64)	`Double.toString`
Boolean	`booleanValue` (varint)	`"true"` / `"false"`
String / Text / UUID	`stringValue`	按 UTF-8 原样保留
Bytes / File	`bytesValue`	十六进制字符串

is_null=true 指标会被评估为空字符串。

NBIRTH / DBIRTH 的别名学习

Sparkplug 效率核心：BIRTH 发送指标名称与别名对 → 之后 NDATA/DDATA 仅发送别名。

1) edge1 의 NBIRTH:
   metric { name="Temperature" alias=7 datatype=Float float_value=23.5 }
   → driver: aliasByPrefix.put("edge1", {7 → "Temperature"})
            metricCache.put("edge1/Temperature", "23.5")

2) edge1 의 NDATA (이름 생략):
   metric { alias=7 datatype=Float float_value=24.1 }
   → driver: name 없음 → alias=7 → "Temperature"
            metricCache.put("edge1/Temperature", "24.1")

如果 NDATA 在缺少 NBIRTH 的情况下到达，别名查找会失败并被忽略（不会更新缓存）。等待代理重启后发布者重新发布 NBIRTH。

NDEATH / DDEATH

当边缘节点或设备终止时，代理会通过 LWT（遗嘱消息）发布 NDEATH/DDEATH。驱动仅清空别名映射，但保留缓存 — 最后的已知值会作为轮询结果保留（PlantPulse 的监视器会照常显示 lastValue 界面）。

NCMD / DCMD 发送

write(ProtocolAddress) 会根据地址段的数量自动分支。

地址	发布主题	备注
`edge1/SetPoint`	`spBv1.0/<group>/NCMD/edge1`	节点级命令
`edge1/dev1/SetPoint`	`spBv1.0/<group>/DCMD/edge1/dev1`	设备级命令

阶段 1 会将所有命令以 Sparkplug 数据类型 String 发布 — 发送端（Tahu / Ignition）会根据指标定义的数据类型进行转换并处理。

限制与后续工作

TLS / 认证 — 目前仅支持明文 + 匿名。用户名/密码 / mTLS 选项后续添加。
QoS — 所有发布/订阅均为 QoS 0。Sparkplug 推荐仅 NBIRTH 使用 QoS 1，NDATA/DDATA 使用 QoS 0。
DataSet / 模板指标 — 线上数据以原始字节保留（datasetRaw/templateRaw），但解码/格式化尚未实现。
属性 / 元数据 — 仅保留原始字节，解码未实现。
bdSeq — NBIRTH 的 bdSeq 匹配 / NDEATH 验证未实现。

测试位置

test/java/plantpulse/driver/protocol/sparkplug/SparkplugBDriverTest.java — 27 个测试。

初始状态 / 选项解析
未连接时的 read/write 安全性
BIRTH 别名注册，NDATA 仅别名解析，组不匹配阻止，DEATH 别名清理
各数据类型的 formatValue 准确性 (Int32 / Float / Double / Boolean / String / UInt32 / UInt64 / is_null)
负载编码→解码往返
MQTT 剩余长度 / UTF-8 线上工具
safeId clientId 清理

操作流程​

选项​

Sparkplug 数据类型映射​

NBIRTH / DBIRTH 的别名学习​

NDEATH / DDEATH​

NCMD / DCMD 发送​

限制与后续工作​

测试位置​