实战笔记 2026-10-12 16 分钟阅读 #Locust #性能测试 #Python 复制链接

Locust 压测脚本入门

Locust 的价值在于"压测脚本就是普通 Python 代码"：能写分支、能读数据库、能调 SDK。代价是并发模型容易写错——它是 gevent 协程，不是线程，一次阻塞调用就能把并发打回单线程。下面按一个能跑的真实脚本展开，再讲分布式和踩过的坑。

一、环境与版本

python -m venv .venv && source .venv/bin/activate    # Windows: .venv\Scripts\activate
pip install "locust>=2.20" requests
locust -V

Locust 2.x 与 1.x 的差异要留意：--no-web 改成 --headless，--csv 输出文件名不变，但移除了一些旧参数。网上大量 1.x 教程会让人照着敲报错。

二、先定压力模型，再写脚本

写脚本前必须能回答三个问题，否则压出来的数字没有意义：

1. 业务比例：浏览:加购:下单 = 6:3:1，不是平均分配。
2. 思考时间：真实用户两次点击间隔 1~3 秒，压测机不模拟这个，RPS 会虚高。
3. 目标指标：是压 TPS 上限，还是验证 500 并发下 P95 < 300ms。两者场景设计完全不同。

三、一个可用的基础脚本

import random
from locust import HttpUser, task, between, events

class OrderUser(HttpUser):
    # 思考时间 1~3 秒，均匀分布
    wait_time = between(1, 3)

    def on_start(self):
        # 登录：失败要显式标记，否则会算成一次成功的业务请求
        with self.client.post(
            "/api/login",
            json={"username": self.username(), "password": "Passw0rd!"},
            name="登录",
            catch_response=True,
        ) as resp:
            if resp.status_code != 200:
                resp.failure(f"login http {resp.status_code}")
                self.environment.runner.quit()     # 登录都失败，直接停止，别产生脏数据
                return
            body = resp.json()
            if body.get("code") != 0:
                resp.failure(f"login biz {body.get('code')}")
                self.environment.runner.quit()
                return
            self.token = body["data"]["token"]
            self.client.headers.update({"Authorization": f"Bearer {self.token}"})

    def username(self):
        # 从账号池里取，避免几千人共用一个账号导致缓存命中虚高
        return f"qa_user_{random.randint(1, 200):03d}"

    @task(6)
    def browse(self):
        page = random.randint(1, 20)
        self.client.get(f"/api/products?page={page}", name="商品列表")

    @task(3)
    def add_cart(self):
        with self.client.post(
            "/api/cart", json={"skuId": random.randint(1000, 1100), "count": 1},
            name="加购", catch_response=True,
        ) as resp:
            if resp.status_code == 409:
                resp.success()                      # 库存不足是业务正常分支，不算失败
            elif resp.status_code != 200:
                resp.failure(f"add_cart {resp.status_code}")

    @task(1)
    def create_order(self):
        with self.client.post(
            "/api/orders", json={"skuId": 1001, "count": 1},
            name="下单", catch_response=True,
        ) as resp:
            data = resp.json() if resp.status_code == 200 else {}
            if data.get("code") != 0:
                resp.failure(f"order failed: {resp.text[:120]}")

三个容易写错的点：

• name= 必须加。否则 /api/products?page=1 到 page=20 会被统计成 20 个独立接口，P95 全被拆散。
• catch_response=True 之后必须调用 resp.success() 或 resp.failure()，否则请求会被记为成功但没有任何断言。
• on_start 里不要做重活（比如拉全量商品、建复杂数据）。启动阶段所有用户同时执行，会把 RT 曲线打出一个假尖刺。

四、定速：constant_throughput 与 constant_pacing

要"恒定 RPS"而不是"恒定用户数"，用内置等待策略：

from locust import HttpUser, task, constant_throughput, constant_pacing

class FixedRpsUser(HttpUser):
    # 每个用户每秒最多发 1 个请求；配合 -u 50 就是约 50 RPS
    wait_time = constant_throughput(1)

class FixedIntervalUser(HttpUser):
    # 每个用户每 2 秒发一次（固定节奏，不受响应时间影响）
    wait_time = constant_pacing(2)

区别：

策略	语义	适用
between(a, b)	随机等待	模拟真实用户
constant_throughput(n)	每用户每秒 n 次，会自动扣掉响应时间	恒定 RPS
constant_pacing(t)	每用户固定 t 秒一轮	恒定节奏、定时任务

constant_throughput 的实际 RPS = 用户数 × n，但响应变慢时会自动缩短等待，所以它是"最少 n 次/秒"，不是精确值。要精确控制 RPS，用 LoadTestShape 反推用户数。

五、自定义指标：事件钩子

Locust 默认只统计请求级指标。要做"登录成功率""下单转化率"，用事件钩子自己记：

from locust import events

order_success = 0
order_total = 0

@events.request.add_listener
def on_request(request_type, name, response_time, response_length,
               response, context, exception, **kwargs):
    global order_success, order_total
    if name == "下单":
        order_total += 1
        if exception is None and response is not None and response.status_code == 200:
            order_success += 1

@events.quitting.add_listener
def on_quit(environment, **kwargs):
    if order_total:
        rate = order_success / order_total * 100
        print(f"[自定义指标] 下单成功率 {rate:.2f}% ({order_success}/{order_total})")
        # 成功率低于阈值时让压测失败，方便 CI 卡口
        if rate < 99:
            environment.process_exit_code = 1

注意：events.request 的回调是全局串行执行的，里面不要做 IO（写文件、发 HTTP）。要落盘就用内存计数，退出时统一写。多 worker 时计数器是各进程独立的，需要汇总就用 master 的 --csv 结果。

六、分布式与多进程

# 单机多核：用 --processes 打满 CPU（-1 表示等于核数）
locust -f locustfile.py --headless -u 2000 -r 100 -t 10m \
  --processes -1 --host https://api.example.com --csv=run1

# 多机分布式：master 只做汇总，不产生压力
locust -f locustfile.py --master --master-bind-port 5557 \
  --expect-workers 3 --host https://api.example.com

# 每台 worker
locust -f locustfile.py --worker --master-host 192.168.1.10 --master-port 5557

要点：

• --expect-workers 3 会让 master 等到 3 个 worker 连上才开始，否则前 10 秒的 RPS 曲线是残缺的。
• 脚本、数据文件、Python 依赖版本必须在所有 worker 上一致。版本不一致会出现"某些 worker 报 KeyError"这种难查的问题。
• 分布式模式下 master 不产生压力，但 CSV 只在 master 上生成，worker 不需要挂盘。
• 容器里跑 worker 记得 --master-host 用服务名而不是 127.0.0.1。

七、负载形状：阶梯加压与自动停止

from locust import LoadTestShape

class StepShape(LoadTestShape):
    stages = [
        {"duration": 300,  "users": 100, "spawn_rate": 10},
        {"duration": 600,  "users": 300, "spawn_rate": 20},
        {"duration": 1200, "users": 500, "spawn_rate": 20},
        {"duration": 1800, "users": 800, "spawn_rate": 40},
    ]

    def tick(self):
        run_time = self.get_run_time()
        for stage in self.stages:
            if run_time < stage["duration"]:
                return stage["users"], stage["spawn_rate"]
        return None       # 返回 None 自动结束压测

配合阈值退出：

@events.quitting.add_listener
def abort_on_error(environment, **kwargs):
    stats = environment.stats.total
    if stats.fail_ratio > 0.01:                  # 失败率 > 1% 直接判失败
        print(f"失败率 {stats.fail_ratio:.2%} 超阈值")
        environment.process_exit_code = 1

这才是"压测能进 CI"的关键：有明确的通过/失败判定，而不是让人看图表。

八、真实踩过的坑

8.1 端口耗尽：Cannot assign requested address

现象：并发上到 800 左右，开始大量 ConnectionError，服务端却毫无压力。

根因：客户端 TIME_WAIT 占满本地端口。默认 ip_local_port_range 只有约 2.8 万个端口，短连接 + 高并发很快耗尽。

# 查看
ss -s | grep timewait
cat /proc/sys/net/ipv4/ip_local_port_range

# 调整（压测机侧，临时生效）
sysctl -w net.ipv4.ip_local_port_range="1024 65535"
sysctl -w net.ipv4.tcp_tw_reuse=1
sysctl -w net.ipv4.tcp_fin_timeout=15
sysctl -w net.core.somaxconn=65535
ulimit -n 65535

更好的办法是复用连接：Locust 的 HttpUser.client 本身就是 requests.Session，会复用连接。但如果你在 task 里自己 requests.get(...)，每个请求都是新连接，端口必炸。

8.2 gevent 阻塞：并发上不去，CPU 只跑满一个核

现象：-u 1000 但 RPS 只有 50，CPU 单核 100%。

根因：在协程里调用了同步阻塞库（pymysql、boto3、time.sleep 之外的 requests 之外的 C 扩展）。

# 错误：同步 MySQL 驱动会阻塞整个 event loop
import pymysql
conn = pymysql.connect(...)

# 正确：monkey patch 必须在所有 import 之前
import gevent.monkey
gevent.monkey.patch_all()
import pymysql       # patch 之后再导入

排查方法：压测时 py-spy dump --pid <locust_pid>，看协程卡在哪个栈。

8.3 统计项被打散

self.client.get(f"/api/item/{id}") 不加 name，1000 个 ID 就是 1000 行统计，P95 全是单样本。所有带路径参数的请求都必须加 name。

8.4 on_start 失败被当成正常

默认情况下 on_start 里的请求失败不会让用户退出，也不会让压测失败。大量用户"登录失败但继续跑"，结果看起来 RPS 很高、错误率很低，实际全是未登录请求打 401。必须在 on_start 里显式 failure()。

8.5 CSV 看不懂

--csv=run1 生成三个文件：

文件	内容
run1_stats.csv	各接口的请求数、失败数、中位数、P95、P99、RPS
run1_failures.csv	失败原因聚合
run1_stats_history.csv	按时间点的用户数、RPS、响应时间（画曲线用）

stats_history.csv 才是判断"是否稳定"的依据：只看最终汇总会掩盖中途的尖刺。

8.6 压测机先到瓶颈

判断压测机是否成为瓶颈的三个信号：

• 压测机 CPU 用户态 > 80%，且服务端 CPU 不高
• RPS 不再随用户数增长，RT 线性上升
• 压测机 ss -s 中 timewait 持续增长

处理：加 worker、复用连接、关闭详细日志（--loglevel WARNING）、别在压测机上跑监控采集。

九、Locust vs JMeter

维度	Locust	JMeter
脚本形态	Python 代码	GUI + XML
复杂逻辑	直接写代码、可调库	需 JSR223/插件
并发模型	gevent 协程（注意阻塞）	线程（吃内存）
分布式	原生 master/worker	需配 RMI，坑多
报告	Web UI + CSV	聚合报告 + HTML
精确 RPS	constant_throughput / Shape	Constant Throughput Timer
适合	开发型团队、链路复杂	传统团队、需要 GUI

选型建议：接口链路有签名、加密、动态依赖时用 Locust；只需要压登录/查询这类简单接口时 JMeter 更快。

十、验收清单

• 每个 task 都有 name，统计项数量与业务接口数一致
• on_start 失败会让压测停止或标记失败
• 业务失败（code != 0）通过 catch_response 显式标记
• 思考时间符合真实用户，不是 0
• 账号按用户分散，不是全员一个账号
• 压测机端口、文件句柄、CPU 已确认不是瓶颈
• 有明确的通过/失败阈值，并能在 CI 中返回非 0
• 保存了 stats_history.csv，能画出随时间变化的曲线

相关阅读

• JMeter 分布式压测实践
• 性能瓶颈定位：CPU、内存、IO与GC
• 性能测试指标体系与容量评估