508 lines
11 KiB
Markdown
508 lines
11 KiB
Markdown
# 本地 MQTT 并发与稳定性测试方案
|
||
|
||
更新日期：2026-07-14
|
||
|
||
本文档用于说明 Banban / TalkingQ 后端在不调用真实外部服务的前提下，如何做本地全流程并发和稳定性测试。测试重点是 MQTT 设备接入、消息解析、服务端回包、GPS 查询响应、告警处理、数据库落库、后端健康状态和资源变化。
|
||
|
||
## 1. 测试目标
|
||
|
||
### 1.1 并发度
|
||
|
||
并发度用于回答系统同时承载多少设备连接和消息流量。
|
||
|
||
本项目里主要看：
|
||
|
||
- 同时在线的 MQTT 设备客户端数量。
|
||
- 设备状态上报 `msg_id=000` 的吞吐。
|
||
- 告警事件 `msg_id=010` 的吞吐和落库。
|
||
- 服务端 GPS 查询 `msg_id=001` 的下发和设备响应。
|
||
- 服务端事件响应 `event_resp` 是否完整返回。
|
||
|
||
### 1.2 稳定性
|
||
|
||
稳定性用于回答系统在持续运行一段时间后是否保持正确和健康。
|
||
|
||
本项目里主要看：
|
||
|
||
- 设备连接是否持续保持。
|
||
- 上行消息和服务端回包是否丢失。
|
||
- 数据库写入是否持续成功。
|
||
- 后端健康检查是否持续返回 `200`。
|
||
- 错误日志是否新增。
|
||
- CPU、内存、数据库连接数、MQTT 连接数是否持续爬升或堆积。
|
||
|
||
## 2. 测试边界
|
||
|
||
### 2.1 本次覆盖
|
||
|
||
- 本地 FastAPI 后端。
|
||
- 本地 MQTT broker。
|
||
- 本地 MySQL。
|
||
- 真实 MQTT 客户端连接，每个模拟设备一个 MQTT client。
|
||
- MQTT 主流程：
|
||
- `device/{device_id}/event` 上行。
|
||
- `device/{device_id}/event_resp` 下行响应。
|
||
- `device/{device_id}/command` 服务端命令。
|
||
- `device/{device_id}/response` 设备响应。
|
||
- 状态上报、告警、GPS 查询、异步持久化、告警落库。
|
||
|
||
### 2.2 本次不覆盖
|
||
|
||
- 线上 MQTT broker。
|
||
- 线上服务器网络和 TLS。
|
||
- 真实高德逆地理编码调用。
|
||
- 真实短信发送。
|
||
- 真实设备固件网络抖动。
|
||
- 断线重连风暴。
|
||
- 慢数据库、数据库断连、外部服务长超时。
|
||
|
||
这些场景需要单独做线上灰度压测或故障注入测试。
|
||
|
||
## 3. 外部服务隔离策略
|
||
|
||
本地全流程测试要验证主流程，但不能依赖真实外部服务。
|
||
|
||
推荐策略：
|
||
|
||
- GPS 响应使用无效坐标 `lat=0,lng=0`。
|
||
- 后端应忽略无效 MQTT 定位，不调用高德解析。
|
||
- 告警测试账号不配置可用家长手机号。
|
||
- 后端短信链路应走“无手机号跳过”，不调用短信服务。
|
||
- 不使用真实 COS 上传、微信模板消息或其他公网依赖作为通过条件。
|
||
|
||
通过标准不是“外部服务成功”，而是“外部服务不可用或被跳过时，MQTT 主流程、回包和落库不受影响”。
|
||
|
||
## 4. 测试环境
|
||
|
||
### 4.1 本地服务
|
||
|
||
在 `talkingq-url` 目录运行：
|
||
|
||
```bash
|
||
.venv/bin/amqtt
|
||
```
|
||
|
||
启动后本地 MQTT broker 监听：
|
||
|
||
```text
|
||
127.0.0.1:1883
|
||
```
|
||
|
||
启动后端：
|
||
|
||
```bash
|
||
.venv/bin/uvicorn main:app --host 127.0.0.1 --port 8080
|
||
```
|
||
|
||
健康检查：
|
||
|
||
```bash
|
||
curl -sS -o /dev/null -w 'openapi_http=%{http_code} time=%{time_total}\n' \
|
||
http://127.0.0.1:8080/openapi.json
|
||
```
|
||
|
||
预期：
|
||
|
||
```text
|
||
openapi_http=200
|
||
```
|
||
|
||
### 4.2 配置要求
|
||
|
||
`talkingq-url/.env` 应指向本地服务：
|
||
|
||
```text
|
||
DB_HOST=127.0.0.1
|
||
DB_PORT=3306
|
||
TALKINGQ_MQTT_BROKER=127.0.0.1
|
||
TALKINGQ_MQTT_PORT=1883
|
||
```
|
||
|
||
### 4.3 测试设备数据
|
||
|
||
建议使用专门的压测设备 ID，避免污染真实设备：
|
||
|
||
```text
|
||
TalkingQ_LOAD000001
|
||
TalkingQ_LOAD000002
|
||
...
|
||
TalkingQ_LOAD000500
|
||
```
|
||
|
||
测试前确认：
|
||
|
||
- `device_auth` 有对应设备。
|
||
- `device_bindings` 有对应绑定。
|
||
- 告警落库使用的 `owner_user_id`、`child_id` 不超过当前数据库字段范围。
|
||
|
||
检查示例：
|
||
|
||
```sql
|
||
SELECT COUNT(*) FROM device_auth WHERE device_id LIKE 'TalkingQ_LOAD%';
|
||
SELECT COUNT(*) FROM device_bindings WHERE device_id LIKE 'TalkingQ_LOAD%';
|
||
```
|
||
|
||
## 5. 压测脚本要求
|
||
|
||
压测脚本应满足以下行为：
|
||
|
||
- 每个设备使用一个独立 MQTT client。
|
||
- 每个 client 订阅：
|
||
- `device/{device_id}/event_resp`
|
||
- `device/{device_id}/command`
|
||
- 每个 client 周期性发布 `msg_id=000` 设备信息。
|
||
- 按固定比例发布 `msg_id=010` 告警。
|
||
- 收到服务端 `msg_id=001` GPS 查询后，回复 `device/{device_id}/response`。
|
||
- 默认 GPS 模式使用 `invalid`，返回 `0,0`，避免真实高德调用。
|
||
- 统计发送数、回包数、GPS 命令数、GPS 响应数、发布延迟。
|
||
|
||
状态上报 payload 示例：
|
||
|
||
```json
|
||
{
|
||
"msg_id": "000",
|
||
"data": {
|
||
"id": "TalkingQ_LOAD000001",
|
||
"power": 79,
|
||
"signal": 4,
|
||
"version": "load-client-1.0.0",
|
||
"voice": 60,
|
||
"imei": "990000000000001",
|
||
"device_type": 2,
|
||
"latitude": 0,
|
||
"longitude": 0,
|
||
"coord_type": "gcj02"
|
||
}
|
||
}
|
||
```
|
||
|
||
告警 payload 示例：
|
||
|
||
```json
|
||
{
|
||
"msg_id": "010"
|
||
}
|
||
```
|
||
|
||
GPS 响应 payload 示例：
|
||
|
||
```json
|
||
{
|
||
"msg_id": "001",
|
||
"status": "success",
|
||
"data": {
|
||
"latitude": 0,
|
||
"longitude": 0,
|
||
"coord_type": "gcj02",
|
||
"accuracy_m": 10,
|
||
"source": 1,
|
||
"battery_pct": 80
|
||
}
|
||
}
|
||
```
|
||
|
||
## 6. 推荐测试阶段
|
||
|
||
### 6.1 冒烟测试
|
||
|
||
目的：确认脚本、broker、后端、DB 基本可用。
|
||
|
||
建议参数：
|
||
|
||
```text
|
||
20 个 MQTT client
|
||
1 轮状态上报
|
||
每 10 台设备触发 1 次告警
|
||
settle 5 秒
|
||
```
|
||
|
||
通过标准：
|
||
|
||
- client 连接成功数等于请求数。
|
||
- `sent_total == event_resp_total`。
|
||
- `resp_000 == sent_000`。
|
||
- `resp_010 == sent_010`。
|
||
- `command_001 == gps_response_sent`。
|
||
|
||
### 6.2 短时并发测试
|
||
|
||
目的：验证较高设备规模下的瞬时吞吐。
|
||
|
||
建议参数：
|
||
|
||
```text
|
||
500 个 device_id
|
||
3 轮状态上报
|
||
每 10 台设备触发 1 次告警
|
||
settle 20 秒
|
||
```
|
||
|
||
通过标准：
|
||
|
||
- 所有事件收到 `event_resp`。
|
||
- 告警落库数量与发送数量一致。
|
||
- 健康检查仍返回 `200`。
|
||
- 新增错误日志为 `0`。
|
||
|
||
### 6.3 真实连接稳定性测试
|
||
|
||
目的：验证 500 个独立 MQTT client 持续在线时，主流程是否稳定。
|
||
|
||
建议参数：
|
||
|
||
```text
|
||
500 个 MQTT client
|
||
持续 30 分钟
|
||
每 15 秒每台设备上报一次 msg_id=000
|
||
每 60 秒每 20 台设备触发一次 msg_id=010
|
||
GPS mode = invalid
|
||
settle 20 秒
|
||
```
|
||
|
||
通过标准：
|
||
|
||
- 500 个 client 全部连接成功。
|
||
- 连接数在测试期间保持稳定。
|
||
- `sent_total == event_resp_total`。
|
||
- `sent_000 == resp_000`。
|
||
- `sent_010 == resp_010`。
|
||
- `command_001 == gps_response_sent`。
|
||
- 30 分钟内后端健康检查持续返回 `200`。
|
||
- 新增错误日志为 `0`。
|
||
- MySQL 连接数没有持续堆积。
|
||
- 后端和 broker RSS 没有持续单调增长。
|
||
|
||
### 6.4 长时稳定性测试
|
||
|
||
目的：把 30 分钟结论提升为更可靠的上线前参考。
|
||
|
||
建议参数：
|
||
|
||
```text
|
||
500 个 MQTT client
|
||
持续 2 小时
|
||
或过夜 8-12 小时
|
||
保持 15 秒状态上报和 60 秒告警节奏
|
||
```
|
||
|
||
增加观察项：
|
||
|
||
- 日志滚动后是否还能正常写入。
|
||
- 定时任务是否干扰 MQTT 主流程。
|
||
- 数据库表增长后查询是否变慢。
|
||
- 后端进程 RSS 是否长期爬升。
|
||
- broker 是否出现连接残留。
|
||
|
||
## 7. 监控指标
|
||
|
||
建议每 60 秒记录一次：
|
||
|
||
```text
|
||
sample
|
||
elapsed_s
|
||
backend_cpu
|
||
backend_mem
|
||
backend_rss_kb
|
||
broker_cpu
|
||
broker_mem
|
||
broker_rss_kb
|
||
mqtt_established
|
||
mysql_threads
|
||
mysql_processes
|
||
http_code
|
||
http_time_s
|
||
new_error_lines
|
||
```
|
||
|
||
采集方式示例：
|
||
|
||
```bash
|
||
ps -o %cpu=,%mem=,rss= -p <backend_pid>
|
||
ps -o %cpu=,%mem=,rss= -p <broker_pid>
|
||
lsof -nP -iTCP:1883 -sTCP:ESTABLISHED
|
||
curl -sS -o /dev/null -w '%{http_code} %{time_total}' http://127.0.0.1:8080/openapi.json
|
||
```
|
||
|
||
MySQL 连接数：
|
||
|
||
```sql
|
||
SHOW STATUS LIKE 'Threads_connected';
|
||
SHOW PROCESSLIST;
|
||
```
|
||
|
||
错误日志过滤：
|
||
|
||
```bash
|
||
rg -n "(ERROR|failed|DataError|Out of range|告警地址解析失败|当前位置地址解析失败|amap|高德)" \
|
||
talkingq-url/logs/talkingq_YYYY-MM-DD.log
|
||
```
|
||
|
||
## 8. 本次执行结果
|
||
|
||
执行日期：2026-07-14
|
||
|
||
### 8.1 单元测试
|
||
|
||
```text
|
||
86 passed
|
||
```
|
||
|
||
### 8.2 500 客户端 30 分钟稳定性测试
|
||
|
||
测试参数：
|
||
|
||
```text
|
||
devices_requested=500
|
||
duration_seconds=1800
|
||
info_interval_seconds=15
|
||
alarm_every=20
|
||
alarm_interval_seconds=60
|
||
gps_mode=invalid
|
||
```
|
||
|
||
结果：
|
||
|
||
```text
|
||
elapsed_seconds=1820.053
|
||
client_ready_success=500
|
||
client_ready_failed=0
|
||
sent_total=60750
|
||
event_resp_total=60750
|
||
sent_000=60000
|
||
resp_000=60000
|
||
sent_010=750
|
||
resp_010=750
|
||
command_001=750
|
||
gps_response_sent=750
|
||
publish_latency_ms_avg=29.301
|
||
publish_latency_ms_p95=48.727
|
||
```
|
||
|
||
### 8.3 数据库验证
|
||
|
||
```text
|
||
device_settings_load=500
|
||
device_alarm_events_load_total=877
|
||
device_alarm_events_load_last_40m=750
|
||
child_location_current_load=0
|
||
mysql_threads_connected=6
|
||
```
|
||
|
||
说明：
|
||
|
||
- `device_alarm_events_load_last_40m=750` 与本轮告警发送数一致。
|
||
- `child_location_current_load=0` 是预期结果，因为 GPS 使用 `0,0`，后端忽略无效定位。
|
||
|
||
### 8.4 资源变化
|
||
|
||
后端：
|
||
|
||
```text
|
||
压测前 RSS 约 129.9MB
|
||
500 客户端连接后 RSS 约 131-132MB
|
||
中途回落到约 120-121MB
|
||
结束后约 121.6MB
|
||
CPU 基本 0.1%-0.2%
|
||
```
|
||
|
||
MQTT broker：
|
||
|
||
```text
|
||
压测前 RSS 约 43.9MB
|
||
500 客户端连接后 RSS 约 73-75MB
|
||
后半程稳定在约 69-70MB
|
||
结束后约 70.1MB
|
||
CPU 基本 0.0%-0.3%
|
||
```
|
||
|
||
连接数：
|
||
|
||
```text
|
||
压测前 mqtt_established=2
|
||
压测中 mqtt_established=1002
|
||
压测后 mqtt_established=2
|
||
```
|
||
|
||
说明：本机 loopback 下 `lsof` 会看到连接两端，`1002` 约等于 `501` 条 MQTT 连接，包括 500 个设备客户端和 1 条后端服务连接。
|
||
|
||
### 8.5 日志验证
|
||
|
||
测试窗口内新增错误日志：
|
||
|
||
```text
|
||
0
|
||
```
|
||
|
||
未发现：
|
||
|
||
- `ERROR`
|
||
- `failed`
|
||
- `DataError`
|
||
- `Out of range`
|
||
- `告警地址解析失败`
|
||
- `当前位置地址解析失败`
|
||
- `amap`
|
||
- `高德`
|
||
|
||
## 9. 本次结论
|
||
|
||
本地环境下，当前项目已验证可以稳定承载：
|
||
|
||
```text
|
||
500 个独立 MQTT 客户端
|
||
持续 30 分钟
|
||
60750 条事件完整回包
|
||
750 条告警完整落库
|
||
750 次 GPS 查询完整响应
|
||
```
|
||
|
||
测试期间：
|
||
|
||
- 后端健康检查持续 `200`。
|
||
- 新增错误日志为 `0`。
|
||
- MySQL 连接没有堆积。
|
||
- 后端和 broker 内存没有持续爬升。
|
||
- 外部服务未调用时，MQTT 主流程不受影响。
|
||
|
||
因此可以判断：本地单实例下，500 台设备级 MQTT 连接和当前消息节奏的 30 分钟稳定性通过。
|
||
|
||
## 10. 风险和下一步
|
||
|
||
### 10.1 当前结论不能直接等同线上容量
|
||
|
||
原因：
|
||
|
||
- 本地 broker 是 `amqtt`，不是线上 broker。
|
||
- 本地网络没有公网延迟、TLS、丢包、移动网络抖动。
|
||
- 高德和短信没有真实调用。
|
||
- 测试时长是 30 分钟，不是 24 小时。
|
||
- 没有模拟断线重连风暴。
|
||
|
||
### 10.2 建议补充测试
|
||
|
||
上线前建议继续做：
|
||
|
||
- 500 客户端持续 2 小时。
|
||
- 500 客户端过夜 8-12 小时。
|
||
- 断线重连测试：一次性断开 20%、50%、100% 客户端后重连。
|
||
- 慢 DB 测试：模拟 DB 写入变慢，确认 MQTT 回包和主流程是否可控。
|
||
- 高德失败测试：模拟超时、HTTP 500、无效 Key，确认地址解析失败不影响告警落库。
|
||
- 短信失败测试：模拟短信服务超时或失败，确认告警主流程不被拖垮。
|
||
- 线上灰度测试：在真实 broker 和服务器规格下从 50、100、300、500 逐步提升。
|
||
|
||
## 11. 通过标准汇总
|
||
|
||
一次测试可判定通过，需要同时满足：
|
||
|
||
- 客户端连接成功数等于目标数。
|
||
- `sent_total == event_resp_total`。
|
||
- `sent_000 == resp_000`。
|
||
- `sent_010 == resp_010`。
|
||
- `command_001 == gps_response_sent`。
|
||
- 告警新增落库数量等于 `sent_010`。
|
||
- 健康检查全程 `200`。
|
||
- 新增错误日志为 `0`。
|
||
- MySQL 连接数没有持续增长。
|
||
- 后端和 broker RSS 没有持续单调增长。
|
||
- 测试结束后 MQTT 连接数回落到基线。
|