# 本地 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 ps -o %cpu=,%mem=,rss= -p 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 连接数回落到基线。