feat(processor): 同步心跳数据到 room_status 表
- 在 HeartbeatProcessor 中新增异步同步逻辑，在历史表写入成功后尝试更新 room_status 表 - 实现 DatabaseManager.upsertRoomStatus 方法，支持批量更新和自动分区创建 - 添加批次内去重逻辑，避免 PostgreSQL ON CONFLICT 冲突 - 新增相关文档：同步方案、测试报告和提案说明
This commit is contained in:
124
docs/plan-room-status-sync.md
Normal file
124
docs/plan-room-status-sync.md
Normal file
@@ -0,0 +1,124 @@
|
||||
# Room Status 状态表同步方案
|
||||
|
||||
## 1. 背景
|
||||
需要将 Kafka 接收到的心跳数据同步写入 `room_status.room_status_moment` 表。该表为设备实时状态表，由多个服务共同维护，当前服务仅负责更新心跳相关的业务字段，要注意：不能阻碍到其他服务对该表的读写操作，也要注意不能因为其他服务对该表的读写操作而影响到当前服务的正常运行（重要）。
|
||||
|
||||
## 2. 表结构与唯一性分析
|
||||
**目标表**：`room_status.room_status_moment`
|
||||
**逻辑唯一键**：`hotel_id` + `room_id` + `device_id`
|
||||
|
||||
**当前挑战**：
|
||||
- 现有 DDL 中，主键定义为 `PRIMARY KEY (hotel_id, room_id, device_id, guid)`。
|
||||
- 目前尚未看到针对 `(hotel_id, room_id, device_id)` 的唯一索引（Unique Index）。
|
||||
- **建议**：为了支持高效的 `INSERT ... ON CONFLICT` 操作并确保数据唯一性，强烈建议在数据库中添加唯一索引：
|
||||
```sql
|
||||
CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_room_status_unique_device
|
||||
ON room_status.room_status_moment (hotel_id, room_id, device_id);
|
||||
```
|
||||
- **应对策略**：在代码实现中，我们将假设上述唯一约束存在（或通过先查后写的方式兜底，但先查后写性能较差且非原子操作）。考虑到性能要求，**推荐使用 Upsert (ON CONFLICT) 语法**。
|
||||
|
||||
## 3. 字段映射方案
|
||||
|
||||
仅更新以下心跳包中包含的字段，其他字段（如 `sys_lock_status`, `online_status` 等）保持原值。
|
||||
|
||||
| 心跳数据源字段 (Source) | 状态表字段 (Target) | 数据类型 | 说明 |
|
||||
| :--- | :--- | :--- | :--- |
|
||||
| `ts_ms` | `ts_ms` | INT8 | 更新时间 |
|
||||
| `ip` | `ip` | TEXT | 设备IP |
|
||||
| `pms_state` | `pms_status` | INT2 | PMS状态 |
|
||||
| `power_state` | `power_state` | INT2 | 取电状态 |
|
||||
| `cardless_state` | `cardless_state` | INT2 | 无人状态 |
|
||||
| `service_mask` | `service_mask` | INT8 | 服务掩码 |
|
||||
| `insert_card` | `insert_card` | INT2 | 插卡状态 |
|
||||
| `bright_g` | `bright_g` | INT2 | 全局亮度 |
|
||||
| `version` | `agreement_ver` | TEXT | 协议版本 |
|
||||
| `carbon_state` | `carbon_state` | INT2 | 碳达人状态 |
|
||||
| **空调数组** | | | |
|
||||
| `air_address` | `air_address` | TEXT[] | |
|
||||
| `state` | `air_state` | INT2[] | |
|
||||
| `model` | `air_model` | INT2[] | |
|
||||
| `speed` | `air_speed` | INT2[] | |
|
||||
| `set_temp` | `air_set_temp` | INT2[] | |
|
||||
| `now_temp` | `air_now_temp` | INT2[] | |
|
||||
| `solenoid_valve` | `air_solenoid_valve` | INT2[] | |
|
||||
| **能耗数组** | | | |
|
||||
| `elec_address` | `elec_address` | TEXT[] | |
|
||||
| `voltage` | `elec_voltage` | DOUBLE[] | |
|
||||
| `ampere` | `elec_ampere` | DOUBLE[] | |
|
||||
| `power` | `elec_power` | DOUBLE[] | |
|
||||
| `phase` | `elec_phase` | DOUBLE[] | |
|
||||
| `energy` | `elec_energy` | DOUBLE[] | |
|
||||
| `sum_energy` | `elec_sum_energy` | DOUBLE[] | |
|
||||
|
||||
## 4. 写入策略与性能优化
|
||||
|
||||
### 4.1 核心逻辑
|
||||
采用 **Batch Upsert** 模式，结合 PostgreSQL 的 `ON CONFLICT` 语法。
|
||||
|
||||
### 4.2 "仅变化时更新" 的实现
|
||||
利用 PostgreSQL 的 `IS DISTINCT FROM` 语法在数据库层过滤无效更新，减少 WAL 日志和 I/O 开销。
|
||||
|
||||
**SQL 模板示例**：
|
||||
```sql
|
||||
INSERT INTO room_status.room_status_moment (
|
||||
hotel_id, room_id, device_id, guid, ts_ms, ip, pms_status, ...
|
||||
) VALUES (
|
||||
$1, $2, $3, gen_random_uuid(), $4, $5, $6, ...
|
||||
)
|
||||
ON CONFLICT (hotel_id, room_id, device_id)
|
||||
DO UPDATE SET
|
||||
ts_ms = EXCLUDED.ts_ms,
|
||||
ip = EXCLUDED.ip,
|
||||
pms_status = EXCLUDED.pms_status,
|
||||
bright_g = EXCLUDED.bright_g,
|
||||
agreement_ver = EXCLUDED.agreement_ver,
|
||||
...
|
||||
WHERE
|
||||
room_status.room_status_moment.ts_ms < EXCLUDED.ts_ms -- 仅允许更新更新的时间戳（可选，防止乱序）
|
||||
AND (
|
||||
room_status.room_status_moment.pms_status IS DISTINCT FROM EXCLUDED.pms_status
|
||||
OR room_status.room_status_moment.power_state IS DISTINCT FROM EXCLUDED.power_state
|
||||
OR room_status.room_status_moment.bright_g IS DISTINCT FROM EXCLUDED.bright_g
|
||||
OR room_status.room_status_moment.agreement_ver IS DISTINCT FROM EXCLUDED.agreement_ver
|
||||
OR ...
|
||||
);
|
||||
```
|
||||
*注：已确认将在数据库中建立唯一索引 `idx_room_status_unique_device`，因此可以直接使用 `ON CONFLICT`。*
|
||||
|
||||
### 4.3 代码修改计划
|
||||
|
||||
1. **DatabaseManager (`src/db/databaseManager.js`)**:
|
||||
* 新增 `upsertRoomStatus(events)` 方法。
|
||||
* 构建针对 `room_status.room_status_moment` 的批量 Upsert 语句。
|
||||
* 处理数组字段的映射和类型转换。
|
||||
|
||||
2. **HeartbeatProcessor (`src/processor/heartbeatProcessor.js`)**:
|
||||
* 在 `processBatch` 中，当 `insertHeartbeatEvents` (历史表) 成功后，调用 `upsertRoomStatus`。
|
||||
* **异步执行**：状态表的更新不应阻塞主流程（或者根据一致性要求决定是否 `await`）。建议 `await` 但捕获错误，避免影响 Offset 提交（除非要求强一致性）。
|
||||
* 鉴于用户要求“入库成功以后才提交kafka消费回执”，建议将两个写操作串行执行：
|
||||
1. 写历史表 (Must Success)
|
||||
2. 写状态表 (Should Success, Log Error if fail)
|
||||
3. 提交 Offset
|
||||
|
||||
## 5. 待确认事项
|
||||
1. **唯一索引**：已确认创建 `idx_room_status_unique_device`。
|
||||
2. **guid 处理**：已确认使用 `gen_random_uuid()`。
|
||||
|
||||
## 7. 实施记录 (Implemented)
|
||||
|
||||
### 7.1 功能实现
|
||||
- [x] **唯一索引创建**：已在 `docs/room_status_moment.sql` 中添加 `idx_room_status_unique_device`。
|
||||
- [x] **Batch Upsert**：在 `DatabaseManager.upsertRoomStatus` 中实现了基于 `ON CONFLICT` 的批量更新。
|
||||
- [x] **批次内去重**：为了解决 PostgreSQL "ON CONFLICT command cannot affect row a second time" 限制，在应用层实现了批次内去重逻辑（保留 `ts_ms` 最新的记录）。
|
||||
- [x] **自动分区**：实现了 `isRoomStatusMissingPartitionError` 和 `ensureRoomStatusPartitions`，当检测到分区缺失错误时，自动创建对应 `hotel_id` 的分区并重试写入。
|
||||
|
||||
### 7.2 验证结果
|
||||
- **启动测试**：系统正常启动，无报错。
|
||||
- **自动分区**：模拟 `hotel_id=3000` 的数据写入，验证了系统能自动创建 `room_status_moment_h3000` 分区并写入成功。
|
||||
- **去重逻辑**：模拟同一批次包含同一设备的多次更新，验证了去重逻辑生效，无 PostgreSQL 警告。
|
||||
- **数据一致性**：验证了数据库中数据的 `ts_ms` 为最新值。
|
||||
|
||||
## 8. 归档说明
|
||||
- 相关的测试报告已归档至 `docs/archive/` 目录。
|
||||
- 核心代码位于 `src/db/databaseManager.js`。
|
||||
- 数据库脚本位于 `docs/room_status_moment.sql`。
|
||||
Reference in New Issue
Block a user