Skip to content

Commit 4e6f79f

Browse files
author
hezhuozhuo
committed
feat(flow): Implement Attribute, Event, and Telemetry data flow processors
- Added AttributeFlow for processing attribute messages with support for device configuration and gateway handling. - Introduced EventFlow for managing event messages, including gateway and sub-device event processing. - Created TelemetryFlow to handle telemetry messages, including data forwarding and storage integration. - Implemented Bus for message distribution between adapters and flows, with support for multiple message types. - Added FlowManager to manage the lifecycle of all flows, including starting and stopping them. - Defined error handling for bus operations and flow processing. - Enhanced logging for better traceability and debugging.
1 parent c75cefe commit 4e6f79f

21 files changed

Lines changed: 2806 additions & 11 deletions

‎configs/conf-dev.yml‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ mqtt_server: vernemq
5353

5454
mqtt:
5555
access_address: 47.92.253.145:1883
56-
broker: 127.0.0.1:1883 # 默认localhost:1883
56+
broker: 47.92.253.145:1883 # 默认localhost:1883
5757
# broker: 47.92.253.145:1883
5858
user: root # 默认root
5959
pass: root # 默认root
@@ -161,4 +161,9 @@ storage:
161161
channel_buffer_size: 10000 # 输入channel缓冲区大小
162162
telemetry_batch_size: 500 # 遥测数据批量大小
163163
telemetry_flush_interval: 1000 # flush间隔（毫秒，0表示关闭定时flush）
164-
enable_metrics: true # 是否启用Prometheus监控
164+
enable_metrics: true # 是否启用Prometheus监控
165+
166+
# Flow 数据流处理层配置
167+
flow:
168+
enable: true # 是否启用Flow层（默认true）
169+
bus_buffer_size: 1000 # Bus消息总线缓冲区大小

‎configs/conf.yml‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -161,4 +161,9 @@ storage:
161161
channel_buffer_size: 10000 # 输入channel缓冲区大小
162162
telemetry_batch_size: 500 # 遥测数据批量大小
163163
telemetry_flush_interval: 1000 # flush间隔（毫秒，0表示关闭定时flush）
164-
enable_metrics: true # 是否启用Prometheus监控
164+
enable_metrics: true # 是否启用Prometheus监控
165+
166+
# Flow 数据流处理层配置
167+
flow:
168+
enable: true # 是否启用Flow层（默认true）
169+
bus_buffer_size: 1000 # Bus消息总线缓冲区大小

‎docs/demand-community/2025.10.15-Flow层网关测试指南.md‎

Lines changed: 221 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,221 @@
1+
# Flow层网关消息测试指南
2+
3+
## 测试前检查清单
4+
5+
### 1. 配置检查
6+
确认 `configs/conf.yml` 中 Flow 已启用:
7+
```yaml
8+
flow:
9+
enable: true # 必须为 true
10+
bus_buffer_size: 1000 # Bus消息总线缓冲区大小
11+
```
12+
13+
### 2. 日志级别设置
14+
为了看到详细的调试日志,设置日志级别为 DEBUG:
15+
```yaml
16+
system:
17+
log_level: debug # 或者 info
18+
```
19+
20+
### 3. 重启服务
21+
修改配置后需要重启服务才能生效。
22+
23+
## 测试步骤
24+
25+
### 测试1: 网关遥测上报
26+
27+
#### 1.1 准备测试数据
28+
使用 MQTT 客户端发送消息到主题: `gateway/telemetry/{gateway_device_id}`
29+
30+
Payload (需要先包装成标准格式):
31+
```json
32+
{
33+
"device_id": "16df85ab-a749-5a35-04d0-8bc8819ba825",
34+
"values": "{\"gateway_data\":{\"switch\":1},\"sub_gateway_data\":{\"47ac607c\":{\"gateway_data\":{\"temperature\":28.5,\"version\":\"v0.1\",\"switch\":true},\"sub_device_data\":{\"1d651702\":{\"temperature\":28.3}}}}}"
35+
}
36+
```
37+
38+
#### 1.2 观察日志输出
39+
查找以下关键日志:
40+
41+
**1. Adapter层日志:**
42+
```
43+
level=debug msg="Gateway telemetry routing to Flow layer" topic="gateway/telemetry/xxx"
44+
level=debug msg="Telemetry message published to bus via Flow layer" device_id=xxx is_gateway=true msg_type=gateway_telemetry
45+
```
46+
47+
**2. Flow层日志:**
48+
```
49+
level=debug msg="Processing gateway message" device_id=xxx
50+
level=debug msg="Processing sub device" device_id=xxx parent_id=xxx
51+
level=debug msg="Processing sub gateway" device_id=xxx parent_id=xxx depth=1
52+
```
53+
54+
**3. Storage层日志:**
55+
```
56+
level=info msg="Telemetry data stored" device_id=xxx tenant_id=xxx points=N
57+
```
58+
59+
#### 1.3 验证数据库
60+
检查 `telemetry_datas` 表:
61+
```sql
62+
-- 网关自身数据
63+
SELECT * FROM telemetry_datas
64+
WHERE device_id = '16df85ab-a749-5a35-04d0-8bc8819ba825'
65+
ORDER BY ts DESC LIMIT 1;
66+
67+
-- 子网关数据
68+
SELECT * FROM telemetry_datas
69+
WHERE device_id IN (
70+
SELECT id FROM devices WHERE sub_device_addr = '47ac607c'
71+
)
72+
ORDER BY ts DESC LIMIT 1;
73+
74+
-- 子设备数据
75+
SELECT * FROM telemetry_datas
76+
WHERE device_id IN (
77+
SELECT id FROM devices WHERE sub_device_addr = '1d651702'
78+
)
79+
ORDER BY ts DESC LIMIT 1;
80+
```
81+
82+
### 测试2: 网关属性上报
83+
84+
#### 2.1 发送消息
85+
主题: `gateway/attributes/{gateway_device_id}`
86+
87+
Payload:
88+
```json
89+
{
90+
"device_id": "16df85ab-a749-5a35-04d0-8bc8819ba825",
91+
"values": "{\"gateway_data\":{\"online_status\":\"online\",\"signal_strength\":85},\"sub_gateway_data\":{\"47ac607c\":{\"gateway_data\":{\"location\":\"Building A-2F\",\"firmware_version\":\"v1.2.3\",\"online_status\":\"online\"},\"sub_device_data\":{\"1d651702\":{\"battery_level\":78,\"location\":\"Room 201\"}}}}}"
92+
}
93+
```
94+
95+
#### 2.2 验证数据库
96+
检查 `attribute_datas` 表:
97+
```sql
98+
SELECT device_id, key, value, ts
99+
FROM attribute_datas
100+
WHERE device_id IN (
101+
'16df85ab-a749-5a35-04d0-8bc8819ba825',
102+
(SELECT id FROM devices WHERE sub_device_addr = '47ac607c'),
103+
(SELECT id FROM devices WHERE sub_device_addr = '1d651702')
104+
)
105+
ORDER BY ts DESC;
106+
```
107+
108+
### 测试3: 网关事件上报
109+
110+
#### 3.1 发送消息
111+
主题: `gateway/event/{gateway_device_id}`
112+
113+
Payload:
114+
```json
115+
{
116+
"device_id": "16df85ab-a749-5a35-04d0-8bc8819ba825",
117+
"values": "{\"gateway_data\":{\"method\":\"SystemAlert\",\"params\":{\"alert_type\":\"system_start\",\"timestamp\":1726581600,\"message\":\"Gateway system started successfully\"}},\"sub_gateway_data\":{\"47ac607c\":{\"gateway_data\":{\"method\":\"TemperatureAlert\",\"params\":{\"temperature\":35.2,\"threshold\":30.0,\"timestamp\":1726581650,\"severity\":\"warning\"}},\"sub_device_data\":{\"1d651702\":{\"method\":\"BatteryAlert\",\"params\":{\"battery_level\":18,\"threshold\":20,\"timestamp\":1726581700,\"status\":\"low_battery\"}}}}}}"
118+
}
119+
```
120+
121+
#### 3.2 验证数据库
122+
检查 `event_datas` 表:
123+
```sql
124+
SELECT device_id, identify, data, ts
125+
FROM event_datas
126+
WHERE device_id IN (
127+
'16df85ab-a749-5a35-04d0-8bc8819ba825',
128+
(SELECT id FROM devices WHERE sub_device_addr = '47ac607c'),
129+
(SELECT id FROM devices WHERE sub_device_addr = '1d651702')
130+
)
131+
ORDER BY ts DESC;
132+
```
133+
134+
## 常见问题排查
135+
136+
### 问题1: 日志显示 "using legacy processing"
137+
**原因:** `mqttAdapter` 未注册或为 nil
138+
139+
**排查步骤:**
140+
1. 检查 `flow.enable` 配置是否为 `true`
141+
2. 查找日志中是否有 `"MQTT Adapter registered for Flow layer"`
142+
3. 检查 `WithFlowService()` 是否在 `WithMQTTService()` 之前调用
143+
4. 重启服务
144+
145+
### 问题2: 日志显示走了Flow层但数据未拆分
146+
**原因:** Topic判断逻辑错误或消息类型未正确识别
147+
148+
**排查步骤:**
149+
1. 检查日志中的 `msg_type` 字段,应该是 `gateway_telemetry/gateway_attribute/gateway_event`
150+
2. 检查 Topic 格式是否正确: `gateway/{type}/{device_id}`
151+
3. 确认 `detectMessageType()` 逻辑正确
152+
153+
### 问题3: 网关自身数据有,但子设备数据缺失
154+
**原因:**
155+
- 子设备地址(sub_device_addr)不匹配
156+
- 子设备未在数据库中注册
157+
- 脚本处理后 Payload 格式不正确
158+
159+
**排查步骤:**
160+
1. 检查数据库中子设备的 `sub_device_addr` 字段
161+
2. 查找日志中 "Sub device not found" 警告
162+
3. 验证脚本处理后的 Payload 是否符合 `GatewayPublish` 结构
163+
164+
### 问题4: 多层网关只处理了第一层
165+
**原因:** 递归深度限制或子网关数据结构错误
166+
167+
**排查步骤:**
168+
1. 检查是否有 "Maximum gateway depth (5) exceeded" 日志
169+
2. 验证 `sub_gateway_data` 的嵌套结构是否正确
170+
3. 查看 `processSubGateways()` 的调用栈
171+
172+
## 性能验证
173+
174+
### 并发测试
175+
使用MQTT压测工具发送大量网关消息,验证:
176+
- Bus缓冲区是否溢出
177+
- Storage写入是否及时
178+
- 内存使用是否稳定
179+
- CPU使用率是否合理
180+
181+
### 建议指标
182+
- Bus消息处理延迟: < 10ms
183+
- Storage写入延迟: < 50ms (批量) / < 10ms (直写)
184+
- 网关拆分性能: 单条消息拆分10个子设备 < 5ms
185+
- 内存使用: 增长稳定,无泄漏
186+
187+
## 成功标志
188+
189+
✅ 日志中显示 "routing to Flow layer"
190+
✅ 日志中显示 `msg_type=gateway_*`
191+
✅ 网关自身数据正确入库
192+
✅ 子设备数据正确拆分并入库
193+
✅ 多层网关(最多5层)正确递归处理
194+
✅ 心跳更新正常
195+
✅ 场景联动触发正常
196+
197+
## 测试命令示例
198+
199+
### 使用 mosquitto_pub 测试
200+
```bash
201+
# 网关遥测
202+
mosquitto_pub -h localhost -t "gateway/telemetry/16df85ab-a749-5a35-04d0-8bc8819ba825" \
203+
-m '{"device_id":"16df85ab-a749-5a35-04d0-8bc8819ba825","values":"{\"gateway_data\":{\"switch\":1}}"}'
204+
205+
# 网关属性
206+
mosquitto_pub -h localhost -t "gateway/attributes/16df85ab-a749-5a35-04d0-8bc8819ba825" \
207+
-m '{"device_id":"16df85ab-a749-5a35-04d0-8bc8819ba825","values":"{\"gateway_data\":{\"online_status\":\"online\"}}"}'
208+
209+
# 网关事件
210+
mosquitto_pub -h localhost -t "gateway/event/16df85ab-a749-5a35-04d0-8bc8819ba825" \
211+
-m '{"device_id":"16df85ab-a749-5a35-04d0-8bc8819ba825","values":"{\"gateway_data\":{\"method\":\"SystemAlert\",\"params\":{\"alert_type\":\"test\"}}}"}'
212+
```
213+
214+
### 查看实时日志
215+
```bash
216+
# Linux/Mac
217+
tail -f logs/thingspanel.log | grep -E "Flow|gateway|Gateway"
218+
219+
# Windows PowerShell
220+
Get-Content logs\thingspanel.log -Wait | Select-String -Pattern "Flow|gateway|Gateway"
221+
```

0 commit comments

Comments
 (0)