plantpulse-messaging（消息层）

作用

作为 PlantPulse 的神经系统的消息代理层。它负责采集来自工业现场的传感器数据，并在模块之间传递事件。共有两个代理——Kafka 和 HiveMQ (MQTT)。

项目	值
模块名	`plantpulse-messaging`
安装路径	`/opt/kopens/plantpulse-platform/plantpulse-messaging/`
数据位置	`/data1/pp-data/kafka/kraft_combined_logs`
`pd` 服务	`messaging` — 仅 MASTER 节点
账户	`PP_MQ_USER` / `PP_MQ_PASSWORD` — Kafka 和 HiveMQ 共享同一个值

STOMP (ActiveMQ) 已退役

随着浏览器实时推送改为 SSE，STOMP 代理已从架构中移除。61000 / 61004 端口和 PP_STOMP_* 不再提供服务。如果旧防火墙规则中仍有残留，请清理。

架构

Kafka

以 KRaft 模式（无 ZooKeeper）运行的单个代理同时兼任控制器。

项目	值
端口	9092 (SASL_PLAINTEXT) · 9093 (controller, 内部) · 9094 (SASL_SSL)
认证	SASL/PLAIN — 静态 JAAS (`config/jaas.conf`)。账户 `PP_MQ_USER`
广播地址	`PP_KAFKA_ADVERTISED_HOST` — 启动时根据主机地址生成
主题前缀	`PP_TOPIC_PREFIX` (默认 `pp`)

代理只读取一个文件

plantpulse-messaging/kafka/config/kafka.properties 是代理配置，由主机的 /etc/kopens/conf/kafka.properties.template 渲染生成。同一目录下的 broker.properties · server.properties · controller.properties 是 Apache Kafka 发行版自带的示例文件，没有人会读取。如果在容器内看到 advertised.listeners=PLAINTEXT://localhost:9092 这样的值，不要惊讶。

模板中的关键行（占位符在渲染时填充）：

process.roles=broker,controller
listeners=SASL_PLAINTEXT://0.0.0.0:${PP_KAFKA_PORT},CONTROLLER://0.0.0.0:${PP_KAFKA_CONTROLLER_PORT:9093},SASL_SSL://0.0.0.0:${PP_KAFKA_TLS_PORT}
advertised.listeners=SASL_PLAINTEXT://${PP_KAFKA_ADVERTISED_HOST}:${PP_KAFKA_PORT},…
sasl.enabled.mechanisms=PLAIN
allow.everyone.if.no.acl.found=false

log.dirs=${PP_DATA_DIR}/kafka/kraft_combined_logs
log.retention.ms=3600000            # 1시간 — 버퍼이지 보관소가 아닙니다
log.segment.bytes=10485760          # 10MB
num.partitions=4
default.replication.factor=1        # 단일 브로커
message.max.bytes=157286400
auto.create.topics.enable=true

Kafka 是 1 小时缓冲区

代理的默认保留时间为 1 小时。数据的长期存储由 Aster TSDB · 时序引擎 · Iceberg 负责，Kafka 则充当它们之间的缓冲区。主题级别的保留时间以及«谁决定了这个值»由 pd retention 显示。

主要主题

会添加前缀 pp (PP_TOPIC_PREFIX)。以下是平台代码使用的代表性主题。

主题	用途
`event` · `event-row`	服务器 ↔ 引擎事件总线
`pp-tag-point`	标签值流 — 需要为每个标签开启«Kafka 发布»才会填充
`pp-tag-alarm` · `pp-asset-alarm` · `pp-asset-event`	报警 · 资产事件
`pp-asset-data` · `pp-asset-aggregation` · `pp-asset-health-status`	资产数据和聚合
`pp-batch`	批处理作业
`pp-edge-status` · `pp-opc-status` · `pp-diagnostic`	边缘 · OPC · 诊断状态
`pp-domain-changed-event`	元数据变更
`pp-production-oee` · `pp-production-ram` · `pp-production-ems`	生产指标

运维命令

docker exec plantpulse-datalake pd status messaging    # UP (advertised <address>:9092 answered ApiVersions)
docker exec plantpulse-datalake pd flow                # 컨슈머 그룹별 lag
docker exec plantpulse-datalake pd node topic          # 토픽 "event" describe (비밀번호 자동 주입)
docker exec plantpulse-datalake pd retention           # 토픽별 보존과 세그먼트 크기
docker exec plantpulse-datalake pd restart messaging

若要直接使用 Kafka 发行版的工具，需要将 SASL 凭据作为客户端配置文件传入。pd flow · pd node topic 可以替代这项操作，请优先使用它们。

9092 端口开放不代表就绪

代理会在启用请求处理之前约 10 秒就开放 9092 端口。因此，pd 的健康检查不是检查端口，而是检查«向广播地址发送 ApiVersions 请求并收到响应»。启动后立即执行的 DOWN (port 9092 is bound but … did not answer) 请在 30 秒后重试。

调优要点

全部在 /etc/kopens/conf/kafka.properties.template 中修改，并通过 restart-datalake.sh 生效。

项目	默认	备注
`log.retention.ms`	1 小时	如果消费端可能长时间停滞，则增加此值。磁盘占用会相应增加
`num.partitions`	4	新主题的分区数。需与消费者线程数匹配
`num.io.threads` · `num.network.threads`	8 · 3	若使用 NVMe，则 io 可提升至 16
`message.max.bytes`	150MB	为大型消息（文件队列）而设置得较大

HiveMQ (MQTT)

轻量级 IoT 设备 · 传感器和 Edge 代理接入的标准 MQTT 代理（Community Edition）。

项目	值
端口	1883 (明文) · 1884 (TLS) — 由代理容器发布到主机并转发至数据湖
认证	username / password — 安全扩展读取 `conf/auth.properties`
账户	`PP_MQ_USER` / `PP_MQ_PASSWORD` (与 Kafka 相同)
配置	`mqtt/conf/config.xml` ← `/etc/kopens/conf/hivemq.xml.template`

TLS 监听器使用平台公共 keystore (/var/security/plantpulse/master/master.keystore.jks)，不要求客户端证书 (NONE)。

# 외부에서 발행 · 구독 (디버깅)
mosquitto_pub -h <server-ip> -p 1883 -u mq -P '<PP_MQ_PASSWORD>' -t test/topic -m "hello"
mosquitto_sub -h <server-ip> -p 1883 -u mq -P '<PP_MQ_PASSWORD>' -t 'test/#'

日志

docker exec plantpulse-datalake pd logs --lines 100 messaging

代理	文件
Kafka	`plantpulse-messaging/kafka/logs/`
HiveMQ	`plantpulse-messaging/mqtt/logs/hivemq.log` (`event.log` 是按消息的审计日志，因此 `pd logs` 故意省略)

常见问题

症状	原因	处理
其他机器的客户端在 bootstrap 后断开连接	广播地址是容器内部地址	节点文件 `PP_MASTER_IP` → 修改配置的方法
`pd status messaging` 是 `DOWN (advertises 127.0.0.1…)`	节点文件中没有地址	同上。与渲染 exit 8 的原因相同
消费者 lag 增加	消费者处理慢 · 停滞	`pd flow` 中的 `MEMB` · `LAG`, 消费者容器日志
磁盘已满	增加了保留时间但消费者停止	`pd retention` 的 kafka 表, `log.retention.ms`
MQTT TLS 握手失败	证书过期 · SAN 缺失	安全设置 — `plantpulse-certs` 会重新签发
轮换后只有 Kafka 认证失败	`PP_MQ_PASSWORD` 是两个代理共用的	使用 `passwd.sh PP_MQ_PASSWORD` 同时更新两者 → 密码

安全 / 外部暴露

端口	推荐
9092 (Kafka SASL 明文)	仅私有网络
9094 (Kafka SASL_SSL)	外部暴露时使用
1883 (MQTT 明文)	仅私有网络
1884 (MQTT TLS)	外部暴露时使用

作用​

架构​

Kafka​

代理只读取一个文件​

主要主题​

运维命令​

调优要点​

HiveMQ (MQTT)​

日志​

常见问题​

安全 / 外部暴露​

相关文档​

作用

架构