Locust压测Flask需将Flask作为独立服务调用,用HttpUser发起HTTP请求;合理配置--users和--spawn-rate以匹配WSGI服务器能力;手动检查JSON响应并调用response.failure标记业务失败;通过X-Request-ID和统一时间格式关联两端日志。

Locust怎么写Flask压测脚本:从启动到定义任务
Locust 不是 Flask 插件,它独立运行,通过 HTTP 请求模拟用户行为。关键在于把 Flask 应用当作普通 Web 服务来调用,而不是在 Locust 内部启动 Flask 实例(否则会干扰并发模型)。
-
locustfile.py中直接使用HttpUser类,不 import Flask 或调用app.run() - 所有接口地址必须指向已启动的 Flask 服务,例如
@#@#@#@#@#@#@#@#@#@0 - 每个请求用
self.client.get()、self.client.post()发起,Locust 自动管理连接池和超时 - 如果 Flask 接口需要认证,可在
on_start()中统一设置self.client.headers
from locust import HttpUser, task, between
<p>class ApiUser(HttpUser):
wait_time = between(1, 3)</p><pre class="brush:php;toolbar:false;">def on_start(self):
self.client.headers.update({"Authorization": "Bearer xyz"})
@task
def get_user_list(self):
self.client.get("/api/users")
并发数、用户数和 spawn rate 怎么配才真实
Locust 的 --users(总用户数)、--spawn-rate(每秒启动用户数)不是简单“开越多越好”,它们和 Flask 的实际处理能力、Gunicorn/Uvicorn 工作进程数强相关。
- 若 Flask 用默认开发服务器(单线程),
--users 10就可能排队阻塞,压测结果反映的是 WSGI 服务器瓶颈,而非业务逻辑 - 生产部署常用 Gunicorn(如
-w 4 -k gevent),此时--users应 ≥ worker 数 × 并发连接上限(gevent 默认约 1000),但需配合--spawn-rate缓慢加压,避免瞬间打满连接 - 常见误配:
--users 1000 --spawn-rate 1000→ 所有用户 1 秒内启动,TCP 连接激增,可能触发本地端口耗尽或目标服务拒绝连接(ConnectionRefusedError)
怎么捕获 Flask 返回的 JSON 错误并计入失败统计
Locust 默认只将 HTTP 状态码 ≥ 400 视为失败,但 Flask 常返回 200 OK + JSON body 包含 {"error": "xxx"}。这类业务错误需手动标记失败,否则压测报告里的“成功率”严重失真。
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac,无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启:辅助功能 + 屏幕录制权限
- 在请求后检查响应体,用
response.json()解析,再判断字段 - 调用
response.failure("reason")主动上报失败,Locust 会统计进 Errors 表 - 注意异常捕获:若响应非 JSON(如 502 Bad Gateway),
response.json()会抛JSONDecodeError,需用 try/except 包裹
@task
def create_order(self):
with self.client.post("/api/orders", json={"item_id": 123}, catch_response=True) as response:
try:
data = response.json()
if data.get("code") != 0:
response.failure(f"API error code: {data.get('code')}")
except Exception as e:
response.failure(f"Invalid JSON: {e}")
Flask 日志和 Locust 报告对不上?时间戳和请求 ID 怎么关联
Locust 压测时看不到 Flask 侧具体哪次请求对应哪个用户行为,尤其出错时难定位。核心办法是让两者共享唯一标识,并统一时间精度。
- Flask 端在请求进入时生成 request id(如用
uuid.uuid4().hex[:8]),写入日志前缀,并通过响应头回传(如X-Request-ID) - Locust 端在请求头中带上自定义 ID(
self.client.get(..., headers={"X-Request-ID": "abc123"})),并在catch_response=True块中读取响应头做日志关联 - Flask 默认日志时间是
time.time()(秒级),Locust 统计精确到毫秒,建议 Flask 日志格式中启用%(asctime)s.%(msecs)03d,避免时间对齐偏差
压测脚本真正难的不是写几个 @task,而是让 Locust 的并发模型和 Flask 的部署形态对齐——比如用了 gevent 却没关掉 Flask 的调试模式,或者忽略了 Werkzeug 开发服务器根本不能扛并发。这些细节不调平,数据就只是数字游戏。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!










