Files
Web_BLS_Heartbeat_Server/openspec/specs/kafka/spec.md
XuJiacheng d10bedb7e7 docs: 归档心跳数组扩展任务并更新规范文档
将已完成的任务清单、提案及规范从变更目录移至归档目录,并更新主规范文档以包含新增的数组字段与处理要求。这标志着该功能变更已从开发阶段进入完成归档阶段。
2026-02-03 08:41:40 +08:00

4.4 KiB
Raw Blame History

Kafka消息处理规范

Purpose

本规范定义本服务如何连接 Kafka 集群、订阅主题并消费消息(以 buffer 形式透传 payload以及错误处理/重连与消费确认语义。

Requirements

Requirement: Kafka连接管理

系统 MUST 能够建立和维护与 Kafka 集群的连接。

Scenario: 成功连接Kafka集群

  • WHEN 系统启动时
  • THEN 应该成功连接到配置的Kafka集群
  • AND 应该监控连接状态

Scenario: Kafka连接断开重连

  • WHEN Kafka连接断开时
  • THEN 系统应该自动尝试重连
  • AND 重连失败时应该记录错误日志

Requirement: 心跳消息消费

系统 MUST 能够消费 Kafka 队列中的心跳消息。

Scenario: 消费心跳消息

  • WHEN Kafka队列中有心跳消息时
  • THEN 系统应该消费该消息
  • AND 将消息传递给处理器进行解包

Scenario: 二进制 payload 透传

  • WHEN Kafka 消息 value 可能为二进制压缩数据(非纯文本)
  • THEN Consumer 应使用 buffer 方式接收 value/key
  • AND 将原始 buffer 交由 Processor 执行解码/解压与反序列化

Scenario: 消息消费确认

  • WHEN 消息处理完成后
  • THEN 系统应该向Kafka确认消息已消费

Requirement: 消息过滤与路由

系统 MUST 能够根据消息类型过滤和路由心跳消息。

Scenario: 过滤无效消息

  • WHEN 接收到无效格式的消息时
  • THEN 系统应该丢弃该消息
  • AND 记录错误日志

Scenario: 路由有效消息

  • WHEN 接收到有效格式的心跳消息时
  • THEN 系统应该将消息路由到正确的处理器

Requirement: 心跳消息载荷格式(生产者约束)

Kafka 心跳消息 MUST 包含数据库落库所需的必填字段,并采用 UTF-8 编码。

Scenario: JSON 心跳消息

  • WHEN 生产者向主题推送心跳消息
  • THEN 消息 value 应为 JSONUTF-8
  • AND 至少包含 ts_ms、hotel_id、room_id、device_id、ip、power_state、guest_type、cardless_state、service_mask、pms_state、carbon_state、device_count、comm_seq
  • AND 可选包含 extrajson object
  • AND 可选包含 electricityarray of object
  • AND 可选包含 air_conditionerarray of object

Scenario: electricity 数组字段结构

  • WHEN 生产者在心跳消息中携带 electricity 字段时
  • THEN electricity MUST 为数组
  • AND 数组每个元素 MUST 为对象,包含:
    • addressstring设备地址
    • voltagedouble电压
    • amperedouble电流
    • powerdouble功率
    • phasestring相位
    • energydouble能耗
    • sum_energydouble总能耗

Scenario: air_conditioner 数组字段结构

  • WHEN 生产者在心跳消息中携带 air_conditioner 字段时
  • THEN air_conditioner MUST 为数组
  • AND 数组每个元素 MUST 为对象,包含:
    • addressstring设备地址
    • stateint2开关状态
    • modelint2运行模式
    • speedint2风速设置
    • set_tempint2设定温度
    • now_tempint2当前温度
    • solenoid_valveint2电磁阀门状态

Requirement: 分区键友好的 Kafka Key

系统 MUST 支持使用 hotel_id:device_id 作为 Kafka message key 以获得更好的分区与有序性。

Scenario: 缺失 key 仍可处理

  • WHEN 消息未携带 key
  • THEN 系统仍应能够消费与处理该消息

Scenario: 使用 device_id 作为 key

  • WHEN 生产者发送消息
  • THEN 建议使用 hotel_id:device_id 作为 Kafka message key
  • AND 以提升同设备有序性与消费侧批量聚合效率

ADDED Requirements

Requirement: 心跳消息扩展数组字段

Kafka 心跳消息 SHALL 支持携带电力与空调子设备的数组字段以便消费端拆列落库与查询优化<E4BC98><E58C96>?

Scenario: 携带 electricity 数组

  • WHEN 生产者需要上报电力设备明细时
  • THEN 消息可选包<E98089><E58C85>?electricityarray of object<63><74>?- AND 每个元素包含 address、voltage、ampere、power、phase、energy、sum_energy

Scenario: 携带 air_conditioner 数组

  • WHEN 生产者需要上报空调设备明细时
  • THEN 消息可选包<E98089><E58C85>?air_conditionerarray of object<63><74>?- AND 每个元素包含 address、state、model、speed、set_temp、now_temp、solenoid_valve