Locust 负载测试入门指南（Python）

本文依据官方文档整理，未执行运行验证或性能基准。代码片段展示局部用法，业务函数、数据和环境需按项目补齐；版本与配置以所引文档为准。

直接回答：Locust 是一个用 Python 编写测试脚本的开源负载测试框架，底层基于轻量级的 gevent 协程而非线程，可在单机运行多个虚拟用户，具体容量取决于脚本、协议、硬件和响应时间，并且自带实时 Web 界面观察压测过程。 Python 团队做负载测试时，它通常是第一候选。

为什么是 Locust

• 代码即脚本：用户行为用普通 Python 类描述，可进版本库、可复用；
• 资源开销低：gevent 的协作式调度减少部分线程开销，但阻塞库和CPU密集操作可能限制吞吐；
• 可分布式：master/worker 模式横向扩展施压能力；
• 实时可视化：Web UI 实时展示 RPS、响应时间分布、失败情况。

安装与项目准备

pip install locust

使用所安装 Locust 版本支持的 Python，并在独立虚拟环境固定依赖版本。

第一个测试

新建 locustfile.py：

from locust import HttpUser, task, between

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)   # 每个任务之间停 1～3 秒，模拟思考时间

    @task
    def list_products(self):
        self.client.get("/api/products")

启动：

locust --web-host=127.0.0.1 --host=http://127.0.0.1:8000

浏览器打开 http://localhost:8089，设置并发数与爬升速率，点 Start 即可开始压测并实时看曲线。

看懂结果界面

响应时间陡增或失败率抬头时，把 Locust 结果与同一时间窗的服务端数据对照。应用通过 OpenTelemetry 接入观测云后，可在链路详情比较数据库、外部请求与应用处理的 Span 耗时，决定下一轮测试应验证哪项瓶颈；同时检查施压机是否先耗尽资源。

多接口与条件检查

用权重分配流量比例，用响应校验过滤"假成功"：

class ShopUser(HttpUser):
    wait_time = between(1, 3)

    @task(3)  # 权重 3：浏览更频繁
    def browse(self):
        self.client.get("/api/products")

    @task(1)  # 权重 1：下单较少
    def order(self):
        with self.client.post("/api/orders",
                              json={"sku": "A1", "qty": 1},
                              catch_response=True) as resp:
            try:
                body = resp.json()
            except ValueError:
                resp.failure("响应不是 JSON")
            else:
                if not isinstance(body, dict) or body.get("code") != 0:
                    resp.failure("业务响应不符合预期")

catch_response=True 让你按业务逻辑判定成败——状态码 200 但业务码异常也算失败。

无 Web UI 运行（CI 友好）

locust --headless -u 10 -r 1 -t 30s --host=http://127.0.0.1:8000

-u 总用户数、-r 每秒爬升用户数、-t 时长；示例是小规模起点，不是推荐容量。必须先启动获授权的本地测试应用，测试订单连接沙箱并可清理；设错误率、资源和费用停止条件，Web UI 不要未经访问控制暴露公网。加 --csv=results 可把指标导出成 CSV 供归档对比。

自定义负载形状

继承 LoadTestShape 可以编程控制任意压力曲线——阶梯、尖峰、波浪：

from locust import LoadTestShape

class StepShape(LoadTestShape):
    step_time = 60      # 每阶 60 秒
    step_load = 100     # 每阶加 100 用户
    spawn_rate = 20
    time_limit = 600

    def tick(self):
        run_time = self.get_run_time()
        if run_time >= self.time_limit:
            return None
        current_step = int(run_time // self.step_time) + 1
        return (current_step * self.step_load, self.spawn_rate)

常见问题（FAQ）

Q：Locust 和 K6 怎么选？
A：团队主力语言决定：Python 选 Locust，JS/TS 选 K6。两者都是代码优先，能力相当；对比时应使用同一脚本语义和资源预算，本文不作性能排名。

Q：单机最多能模拟多少用户？
A：取决于脚本复杂度和机器配置，简单 HTTP 场景几千到上万都有可能。出现施压机 CPU 打满的情况就该上分布式模式了。

Q：wait_time 能去掉吗？
A：按负载模型决定。会话模拟需要思考时间；测服务吞吐可有意采用不同节奏，但必须说明模型，不能把用户数直接当RPS。

官方参考

• 官方文档 1
• 官方文档 2
• 官方文档 4

资料核对日期：2026-09-29。

延伸阅读

• 负载测试入门完全指南
• Minitest 入门指南（Ruby）