跳到主要内容

MQTT 客户端驱动

PlantPulse Edge 的 MQTT 客户端驱动是对 Eclipse Paho mqttv3 1.2.5 库进行封装的 Java 实现。 它订阅外部 IIoT broker（如 HiveMQ / Mosquitto / EMQX / AWS IoT Core / Azure IoT Hub 等）的 topic 并将消息保存在内存缓存中， 按照标签的轮询周期返回缓存值。写操作则立即发布。

源代码：plantpulse.driver.protocol.mqtt_client.*

类职责
MQTTClientDriverProtocolDriver 实现 — connect/read/write/close、topic 缓存、惰性订阅

库依赖：lib/org.eclipse.paho.client.mqttv3-1.2.5.jar

与 Sparkplug B 的区别​

PlantPulse 提供两种基于 MQTT 的驱动。

驱动负载使用场景
MQTT_CLIENT（本页）任意 String/JSON/binary通用 IIoT broker、自定义 topic、ASCII / JSON 负载
SPARKPLUG_BSparkplug B 负载（Protobuf）Cirrus Link / Tahu / Ignition 标准，NBIRTH/DBIRTH 别名

Sparkplug B 自带线缆编解码器和别名学习逻辑，因此仅需通用 MQTT 的环境下 此驱动更轻量，仅使用标准 Paho API 即可满足需求。

操作流程​

broker ──PUBLISH──> Paho MqttClient ──MqttCallback.messageArrived(topic, payload)──> Driver
                                                                                       │
                                                                                       └─ lastValueByTopic.put(topic, new String(payload))

PlantPulse 폴러 ──read(addr)──> lastValueByTopic[addr]   (in-memory hit)
                  write(addr, value) ──Paho.publish(topic, value, qos, retain=false)──> broker

首次调用 read 时若缓存未命中，则进行惰性订阅（使用 QoS 选项值）并返回空字符串 — 从下一周期起 实际值才会到达。tag_map 中注册的 topic 将在连接后立即通过 preSubscribeFromTagMap() 批量订阅，因此第一个周期即可收到值（取决于消息发布频率）。

选项​

键默认值说明
username（无）broker 认证用户名
password（无）broker 认证密码
tlsfalsetrue → ssl://（默认 8883），false → tcp://（默认 1883）
qos0订阅 / 发布 QoS（0 / 1 / 2）。超出范围的值自动限制。
keep-alive60保活周期（秒）— MqttConnectOptions.setKeepAliveInterval
clean-sessiontrueMqttConnectOptions.setCleanSession
client-id（自动）显式客户端 ID。未指定时自动生成 PP-<opc_id>-<random6>（≤23 字符）

MqttConnectOptions.setAutomaticReconnect(true) 始终启用，因此暂时性 broker 中断后会自动重连 （应用 Paho 默认退避策略）。

Address 映射​

address操作
factory/line1/temp单个 topic — 此 topic 的最后一条消息
device/+/status通配符（单层）— 匹配 topic 的最后一条消息（最近到达）
factory/#通配符（多层）— 最后一段，所有子 topic
"" / null返回空

通配符注意：Paho 的回调不会按实际到达的 topic 分别缓存条目， 而是以到达的 topic 本身作为键。即使用 device/+/status 订阅， lastValueByTopic 中会分别存储 device/sensor01/status、device/sensor02/status， 而用户以 通配符 topic 本身读取时，缓存中无条目（通配符本身不是 发布的目标）。注册通配符模式时，应按 device/+/status 方式注册，然后 通过 read 的结果单独追踪最后到达 topic 的值 — 常规使用中建议按 topic 逐一 显式注册。

写操作（发布）​

client.publish(topic, value.getBytes(), qos, /*retain=*/ false);
  • value 的字节将原样作为负载发送（默认字符集为 UTF-8）。
  • 固定为 retain=false，无法更改 — 由于 broker 不保留最后一条消息， 后续才开始订阅的客户端在下次写入发生前无法收到值。 对于需要保留消息的消费者，请在 broker 侧配置桥接或单独发布器。
  • QoS 按选项值原样使用。若为 QoS 1/2，Paho 会在内部重传直至确认。

连接生命周期​

方法操作
connect()应用选项 → 组装 broker URL → MqttClient.connect() → tag_map 批量订阅
read()缓存未命中时惰性订阅 → 缓存命中则返回值
write()发布（retain=false）
close()disconnect() + close()（尽力而为，忽略异常）→ 清空缓存
connectionLost(cause)回调中触发 setConnected(false) — Paho 自动尝试重连

限制及后续工作​

  • mTLS / 客户端证书 — 目前仅支持用户名/密码。对于 AWS IoT 等 X.509 mTLS 环境，需要单独 配置 truststore / keystore（Paho MqttConnectOptions.setSocketFactory 尚未暴露）。
  • retained / will 消息 — 当前固定为 retain=false，未设置 LWT。
  • 通配符缓存分离 — 以通配符模式本身读取时结果为空（按到达 topic 分别存条目）。
  • JSON path 提取 — 即使负载为 JSON，驱动也仅返回原始字符串。JSONPath 提取需在 format / 计算阶段处理（或使用单独解析节点）。
  • 负载编码 — 当前使用平台默认字符集。二进制负载可能损坏 UTF-8 → 未来考虑暴露 payload-encoding 选项（utf8 / iso-8859-1 / base64）。

测试位置​

test/java/plantpulse/driver/protocol/mqtt_client/MQTTClientDriverTest.java — 31 个测试。

  • 初始状态 / 元信息（isConnected/isWriteSupported/isExternal/getDriverSource）
  • 未连接时 read/write/null/empty 安全性
  • getDebugBaseUrl — tcp/ssl、显式端口、默认端口（1883/8883）
  • 选项解析 — username/password/tls/keep-alive/clean-session/qos/client-id、无效值 → 默认值
  • 静态辅助函数 — clampQos、defaultClientId（长度/随机/null）、nullIfEmpty、parseInt、parseBool
  • 常量校验（DEFAULT_PORT、DEFAULT_PORT_TLS、DEFAULT_KEEP_ALIVE_SEC、DEFAULT_QOS）

依赖真实 broker 的集成测试单独进行 — 本单元测试无需外部 broker 即可运行。