跳到主要内容

Apache Kafka 驱动

PlantPulse Edge 的 Apache Kafka 驱动是对 kafka-clients 4.1.2 库进行封装的 Java 实现。 它通过 consumer 订阅外部流处理平台（Apache Kafka / Confluent Cloud / AWS MSK / Azure Event Hubs Kafka API）的主题， 将记录保存在内存缓存中，并在每个标签轮询周期返回缓存值。 写入使用 producer.send（异步）。

来源：plantpulse.driver.protocol.kafka.*

类职责
KafkaDriverProtocolDriver 实现 — 连接/读取/写入/关闭、主题缓存、惰性订阅、后台轮询线程

库：lib/kafka-clients-4.1.2.jar

与 MQTT / WebSocket 的差异​

PlantPulse 提供 三种基于消息的驱动 —— 它们都共享相同的 JsonExtract 4 模式 JSON 解码器。

驱动传输协议负载使用场景
MQTT_CLIENTMQTT v3.1.1任意（通常为 JSON/纯文本）IIoT broker（HiveMQ / Mosquitto / EMQX），QoS 0/1/2
WEBSOCKETws/wss任意（通常为 JSON）流式服务器 / 专有推送 API
KAFKA（本页面）Kafka 传输协议记录键/值（通常为 JSON 字符串）高吞吐量流处理、数据重放（offset）、consumer group
SPARKPLUG_BMQTT + ProtobufSparkplug B 负载Cirrus Link / Tahu / Ignition 标准（基于 MQTT 的 IIoT 规范）

Kafka 的核心是 持久化日志 + offset 重放 —— 当新 consumer 以 auto-offset=earliest 注册时， 它可以接收历史数据（在 broker 的保留期限内）。

操作流程​

broker ──ConsumerRecord──> KafkaConsumer.poll() ──> pollerLoop (background thread)
                                                       │
                                                       ├─ rawByTopic.put(topic, value)
                                                       └─ JsonExtract.updateCache(topicCache, topic, value)

PlantPulse 폴러 ──read(addr)──> JsonExtract.parse(addr.address) ──> JsonExtract.extract(...)   (in-memory hit)
                  write(addr, value) ──KafkaProducer.send(ProducerRecord(topic, value))──> broker

首次调用 read 时，如果缓存未命中，会触发惰性订阅（调用 consumer.subscribe(...)），然后返回空字符串—— 从下一个周期开始，实际值才会返回。在 tag_map 中注册的主题会在连接后立即通过 preSubscribeFromTagMap() 批量订阅，因此第一个周期就可能收到值。

4 模式 JSON 解码规范​

JsonExtract 工具统一了 MQTT / WebSocket / Kafka 三个驱动的消息解码。

模式地址格式操作
SCALAR<topic> 或 <topic>.value将整个消息作为字符串返回。如果是 JSON 对象，则回退到原始数据。
KEY<topic>:<json-key>获取顶层 JSON 对象中某个键的值
PATH<topic>:$.<json-path>JSON Pointer 动态求值（例如：$.data.tags.T1 → /data/tags/T1）
RAW<topic>:_raw_最后的原始消息（用于调试）

缓存结构 —— 使用单个 Map<String, String>，键为 topic + "|" + field：

键含义
topic|_raw_最后的原始消息
topic|value标量（仅当消息不是 JSON 时）
topic|<json-key>顶层 JSON 键对应的值（仅当是对象时）

PATH 模式不经过缓存，而是每次都通过 JSONPointer.queryFrom(...) 对 rawByTopic 中最后的原始消息进行求值。

选项​

键默认值说明
group-idplantpulse-edge-<opc_id>consumer group ID。offset 提交单位。
security-protocolPLAINTEXTPLAINTEXT / SSL / SASL_PLAINTEXT / SASL_SSL
sasl-mechanism（无）PLAIN / SCRAM-SHA-256 / SCRAM-SHA-512
sasl-jaas-config（无）JAAS 配置（例如：...PlainLoginModule required username="u" password="p";）
auto-offsetlatestearliest / latest —— 新 group 的起始位置

enable.auto.commit=true 始终被设置，broker 会自动提交 offset（默认 5 秒周期）。

连接生命周期​

方法操作
connect()应用选项 → 创建 consumer / producer → 启动 poller 线程 → 批量订阅 tag_map
pollerLoop()循环处理 consumer.poll(500ms) → 按记录执行 JsonExtract.updateCache(...)
read()解析 → 缓存未命中则惰性订阅 → 缓存命中则返回值
write()producer.send(new ProducerRecord<>(topic, value))（异步）
close()consumer.wakeup() → 线程 join → 关闭 consumer/producer → 清空缓存

限制及后续工作​

  • 重连 —— kafka-clients 自带重连逻辑，但当整个 broker 宕机而导致长生命周期 poll() 失败时， 驱动侧不会单独标记 setConnected(false) → 建议通过 broker 侧指标监控运行状态。
  • AWS MSK IAM —— 需要添加 aws-msk-iam-auth jar（当前未包含）。未来考虑将 aws-msk-iam-auth-1.x.jar 放入 lib/ ， 并公开 sasl.client.callback.handler.class 作为选项。
  • TLS truststore —— 无法通过驱动选项指定，使用 JVM 系统 truststore。如果 broker 使用自签名私有 CA， 请将 CA 证书导入 JVM truststore，或在网关启动选项中添加 -Djavax.net.ssl.trustStore=…。
  • 分区/键指定 producer —— 写入时始终不携带键，因此分区是轮询分配的。这意味着 同一标签的写入不保证按顺序到达。如果需要按分区顺序，请将主题设置为 1 个分区，或在 consumer 端处理顺序保证。
  • 事务性 producer —— 不支持。当前也未显式启用幂等 producer。

测试位置​

test/java/plantpulse/driver/protocol/kafka/KafkaDriverTest.java —— 18 个测试。

  • 类加载 / 层级（BaseProtocolDriver / ProtocolDriver）
  • 初始状态 / 元数据（isConnected/isWriteSupported/isExternal/getDriverSource）
  • 未连接时的读取/写入/空值安全
  • getDebugBaseUrl —— kafka://，默认端口（9092）
  • 选项解析 —— group-id / security-protocol / sasl-* / auto-offset，options=null 安全
  • 4 模式解码 —— 直接填充缓存验证 KEY / PATH / RAW / SCALAR / .value 后缀
  • close() 未连接安全

test/java/plantpulse/driver/protocol/common/JsonExtractTest.java —— 23 个测试（集成验证 4 模式解析器/缓存/提取）。

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