banban/docs/local-mqtt-stability-concurrency-test-plan.md

本地 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 目录运行：

.venv/bin/amqtt

启动后本地 MQTT broker 监听：

127.0.0.1:1883

启动后端：

.venv/bin/uvicorn main:app --host 127.0.0.1 --port 8080

健康检查：

curl -sS -o /dev/null -w 'openapi_http=%{http_code} time=%{time_total}\n' \
  http://127.0.0.1:8080/openapi.json

预期：

openapi_http=200

4.2 配置要求

talkingq-url/.env 应指向本地服务：

DB_HOST=127.0.0.1
DB_PORT=3306
TALKINGQ_MQTT_BROKER=127.0.0.1
TALKINGQ_MQTT_PORT=1883

4.3 测试设备数据

建议使用专门的压测设备 ID，避免污染真实设备：

TalkingQ_LOAD000001
TalkingQ_LOAD000002
...
TalkingQ_LOAD000500

测试前确认：

• device_auth 有对应设备。
• device_bindings 有对应绑定。
• 告警落库使用的 owner_user_id、child_id 不超过当前数据库字段范围。

检查示例：

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 示例：

{
  "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 示例：

{
  "msg_id": "010"
}

GPS 响应 payload 示例：

{
  "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 基本可用。

建议参数：

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 短时并发测试

目的：验证较高设备规模下的瞬时吞吐。

建议参数：

500 个 device_id
3 轮状态上报
每 10 台设备触发 1 次告警
settle 20 秒

通过标准：

• 所有事件收到 event_resp。
• 告警落库数量与发送数量一致。
• 健康检查仍返回 200。
• 新增错误日志为 0。

6.3 真实连接稳定性测试

目的：验证 500 个独立 MQTT client 持续在线时，主流程是否稳定。

建议参数：

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 分钟结论提升为更可靠的上线前参考。

建议参数：

500 个 MQTT client
持续 2 小时
或过夜 8-12 小时
保持 15 秒状态上报和 60 秒告警节奏

增加观察项：

• 日志滚动后是否还能正常写入。
• 定时任务是否干扰 MQTT 主流程。
• 数据库表增长后查询是否变慢。
• 后端进程 RSS 是否长期爬升。
• broker 是否出现连接残留。

7. 监控指标

建议每 60 秒记录一次：

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

采集方式示例：

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 连接数：

SHOW STATUS LIKE 'Threads_connected';
SHOW PROCESSLIST;

错误日志过滤：

rg -n "(ERROR|failed|DataError|Out of range|告警地址解析失败|当前位置地址解析失败|amap|高德)" \
  talkingq-url/logs/talkingq_YYYY-MM-DD.log

8. 本次执行结果

执行日期：2026-07-14

8.1 单元测试

86 passed

8.2 500 客户端 30 分钟稳定性测试

测试参数：

devices_requested=500
duration_seconds=1800
info_interval_seconds=15
alarm_every=20
alarm_interval_seconds=60
gps_mode=invalid

结果：

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 数据库验证

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 资源变化

后端：

压测前 RSS 约 129.9MB
500 客户端连接后 RSS 约 131-132MB
中途回落到约 120-121MB
结束后约 121.6MB
CPU 基本 0.1%-0.2%

MQTT broker：

压测前 RSS 约 43.9MB
500 客户端连接后 RSS 约 73-75MB
后半程稳定在约 69-70MB
结束后约 70.1MB
CPU 基本 0.0%-0.3%

连接数：

压测前 mqtt_established=2
压测中 mqtt_established=1002
压测后 mqtt_established=2

说明：本机 loopback 下 lsof 会看到连接两端，1002 约等于 501 条 MQTT 连接，包括 500 个设备客户端和 1 条后端服务连接。

8.5 日志验证

测试窗口内新增错误日志：

0

未发现：

• ERROR
• failed
• DataError
• Out of range
• 告警地址解析失败
• 当前位置地址解析失败
• amap
• 高德

9. 本次结论

本地环境下，当前项目已验证可以稳定承载：

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 连接数回落到基线。