XuJiacheng
86a1e79153
- 添加项目基础结构，包括 .gitignore、package.json、Docker 配置和环境变量示例 - 实现核心模块：Kafka 消费者、PostgreSQL 数据库管理器、Redis 客户端与错误队列 - 添加工具类：日志记录器、指标收集器、UUID 生成器 - 实现数据处理器，支持 0x36 上报和 0x0F 命令的解析与存储 - 添加数据库初始化脚本和分区管理，支持按时间范围分区 - 引入 Zod 数据验证和 Vitest 单元测试框架 - 提供完整的项目文档，包括数据库设计、Kafka 格式规范和 Redis 集成协议
93 lines
4.1 KiB
Markdown
93 lines
4.1 KiB
Markdown
# BLS RCU Action Server 开发文档
|
||
|
||
## 1. 项目概述
|
||
本项目旨在构建一个后端服务，负责从 Kafka 接收 RCU（客房控制单元）的通讯日志，解析数据结构，并将其持久化存储到 PostgreSQL 数据库中。
|
||
|
||
核心目标是将不同类型的通讯协议数据（0x36 上报、0x0F 下发、ACK）统一存储，并针对不定长数据结构采用 JSON 格式进行灵活保存。
|
||
|
||
## 2. 系统架构
|
||
**数据流向**: `MCU/Server` (产生数据) -> `Kafka` (消息队列) -> `Action Server` (消费 & 解析) -> `PostgreSQL` (存储)
|
||
|
||
- **Kafka Topic**: `blwlog4Nodejs-rcu-action-topic` (分区数: 6)
|
||
- **数据库**: PostgreSQL
|
||
|
||
## 3. 数据库设计
|
||
|
||
数据库名：`bls_rcu_action`
|
||
模式名: `rcu_action`
|
||
|
||
### 3.1 表结构设计
|
||
表名: `rcu_action_events`
|
||
|
||
| 字段名 | 类型 | 说明 | 备注 |
|
||
| :--- | :--- | :--- | :--- |
|
||
| **guid** | VARCHAR(32) | 主键 (Key2) | 32位无符号UUID |
|
||
| **ts_ms** | BIGINT | 日志产生时间 (Key1) | **必填** (L49) |
|
||
| **write_ts_ms** | BIGINT | 入库时间 | **必填** (L50) |
|
||
| **hotel_id** | INTEGER | 酒店ID | **必填** (L51) Index |
|
||
| **room_id** | VARCHAR(32) | 房间ID | **必填** (L52) Index |
|
||
| **device_id** | VARCHAR(32) | 设备ID | **必填** (新增) Index |
|
||
| **direction** | VARCHAR(10) | 数据方向 | **必填** (L53) "上报"/"下发" Index |
|
||
| **cmd_word** | VARCHAR(10) | 命令字 | **必填** (L54) 如 "0x36", "0x0F" Index |
|
||
| **frame_id** | INTEGER | 通讯帧号 | **必填** (L55) 用于串联命令与状态 |
|
||
| **udp_raw** | TEXT | UDP消息原文 | **必填** (L56) Hex字符串 |
|
||
| **action_type** | VARCHAR(20) | 记录行为类型 | **必填** (L57) Index |
|
||
| **sys_lock_status** | SMALLINT | 系统锁状态 | (L59) 可空 |
|
||
| **report_count** | SMALLINT | 本次上报数量 | (L60) 可空 |
|
||
| **dev_type** | SMALLINT | 设备类型 | (L61) 可空 (统一字段) |
|
||
| **dev_addr** | SMALLINT | 设备地址 | (L62) 可空 (统一字段) |
|
||
| **dev_loop** | INTEGER | 设备回路 | (L63) 可空 (统一字段) |
|
||
| **dev_data** | INTEGER | 设备状态 | (L64) 可空 (0x36状态) |
|
||
| **fault_count** | SMALLINT | 本次故障数量 | (L65) 可空 |
|
||
| **error_type** | SMALLINT | 故障类型 | (L69) 可空 (0x36故障) |
|
||
| **error_data** | SMALLINT | 故障内容 | (L70) 可空 (0x36故障) |
|
||
| **type_l** | SMALLINT | 执行方式 | 可空 (0x0F下发) |
|
||
| **type_h** | SMALLINT | 执行内容 | 可空 (0x0F下发) |
|
||
| **details** | JSONB | 业务详情数据 | 存储不定长设备列表、故障信息等 |
|
||
| **extra** | JSONB | 扩展信息 | 存储通讯原文等扩展数据 |
|
||
|
||
**主键定义**: `(ts_ms, guid)`
|
||
**索引定义**: 备注带index的字段为需要索引的字段，用于提高查询效率。
|
||
|
||
### 3.2 字典定义
|
||
**Action Type (记录行为类型)**:
|
||
- `"0FACK"`: ACK (应答)
|
||
- `"0F下发"`: 下发控制 (0x0F 下发)
|
||
- `"36上报"`: 设备回路状态 (0x36 上报)
|
||
|
||
**Direction (方向)**:
|
||
- `"上报"`: Upload
|
||
- `"下发"`: Download
|
||
|
||
## 4. 数据解析与存储映射
|
||
|
||
### 4.1 0x36 上报数据 (设备状态/故障)
|
||
* **命令字**: "0x36"
|
||
* **拆分逻辑**: 根据 `project.md` 说明，一条 UDP 可能包含多个设备状态，需拆分为多条记录入库，每条记录填充 `dev_type` 等字段。同时将完整的不定长列表存入 `details` 以便追溯。
|
||
* **Action Type**: 4 (设备回路状态)
|
||
|
||
**Mapping**:
|
||
- `sys_lock_status` -> P0
|
||
- `report_count` -> P7
|
||
- `dev_type`, `dev_addr`... -> 从 P8~P13 循环解析，每组生成一条 DB 记录
|
||
- `details`: `{ "all_devices": [...], "all_faults": [...] }`
|
||
- `extra`: `{ "raw_hex": "..." }`
|
||
|
||
### 4.2 0x0F 下发数据 (控制指令)
|
||
* **命令字**: "0x0F"
|
||
* **Action Type**: 2 (下发控制)
|
||
* **存储逻辑**: 主要是控制指令，通常作为单条记录存储。若包含多个设备控制，可选择存第一条到字段，或仅存入 JSON。根据 "0x0F不定长存为JSON" 的需求，主要依赖 `details` 字段。
|
||
|
||
**Mapping**:
|
||
- `details`: `{ "control_params": [ ... ] }`
|
||
- `extra`: `{ "raw_hex": "..." }`
|
||
|
||
### 4.3 0x0F 上报数据 (ACK)
|
||
* **命令字**: "0x0F"
|
||
* **Action Type**: "0FACK" (ACK)
|
||
|
||
**Mapping**:
|
||
- `details`: `{ "ack_code": "0x00" }`
|
||
- `extra`: `{ "raw_hex": "..." }`
|
||
|