Locust实战：从安装到分布式压测

说实话，压测工具我接触过不少。JMeter、wrk、ab 都用过。但 Locust 给我的感觉最特别——它用 Python 写脚本，天然适合我们这些搞 AI 系统的工程师。你想想看，我们平时就在写 Python，现在压测脚本也用 Python，这多顺滑。

我个人习惯把 Locust 放在性能工具链的第一环。为什么？因为它上手快，出活也快。今天我就带你走一遍完整的流程：安装、写脚本、Web UI 跑起来、再到分布式压测。

1. 安装 Locust

安装其实没什么坑。直接用 pip 就行。但我建议你创建一个虚拟环境，别污染全局的 Python 包。

# 创建虚拟环境
python -m venv locust_env
source locust_env/bin/activate  # Linux/Mac
# 或者 locust_env\Scripts\activate  # Windows

# 安装 Locust
pip install locust

验证一下安装是否成功：

locust --version

如果看到版本号，那就成了。嗯，这里要注意一点——Locust 依赖的 gevent 库在某些老旧系统上可能编译失败。我曾经在一台 CentOS 7 的机器上折腾了半小时，最后发现是 gcc 版本太老。解决办法很简单：升级 gcc 或者直接用 Python 3.8+ 的预编译包。

小技巧：如果你在公司内网环境，记得先配好 pip 代理。不然下载 gevent 这种带 C 扩展的包会非常痛苦。

2. 编写压测脚本

Locust 的脚本结构很清晰。核心就两个东西：HttpUser 和 TaskSet。说白了，HttpUser 代表一个虚拟用户，TaskSet 定义这个用户要干什么。

我给你看一个我实际用过的脚本。当时在压一个 AI 推理服务的 API：

from locust import HttpUser, task, between

class AIInferenceUser(HttpUser):
    # 每个用户执行任务之间的等待时间（秒）
    wait_time = between(1, 3)

    @task(3)  # 权重为3，表示这个任务被选中的概率更高
    def predict(self):
        payload = {
            "text": "今天天气真不错",
            "model": "bert-base-chinese"
        }
        with self.client.post(
            "/v1/predict",
            json=payload,
            catch_response=True
        ) as response:
            if response.status_code != 200:
                response.failure(f"请求失败: {response.status_code}")
            elif response.elapsed.total_seconds() > 2.0:
                response.failure("响应超时 > 2s")

    @task(1)
    def health_check(self):
        self.client.get("/health")

这里有几个关键点我想强调一下：

  • @task 的权重：我习惯把核心接口的权重设高一些。比如预测接口权重 3，健康检查权重 1。这样模拟的流量更真实。
  • catch_response=True：这个参数很重要。不加的话，Locust 只认 HTTP 状态码。加了之后，你可以自定义什么算成功、什么算失败。
  • response.elapsed：这是 Locust 自带的响应时间统计。我经常用它来做超时判断。
注意：不要在 @task 里写太重的逻辑。比如不要做复杂的 JSON 序列化，不要调外部数据库。这些会拖慢压测本身的性能，导致结果不准确。

3. Web UI 模式运行

脚本写好了，怎么跑？我最推荐先用 Web UI 模式。它让你能实时看到压测情况，方便调试。

locust -f ai_inference_test.py --web-host 0.0.0.0 --web-port 8089

启动后，打开浏览器访问 http://localhost:8089。你会看到一个简洁的界面：

  • Number of users：模拟的用户总数
  • Spawn rate：每秒启动的用户数
  • Host：目标服务器的地址

填好参数，点击 "Start swarming"，压测就开始了。Web UI 会实时显示：

指标 说明 我关注的阈值
RPS 每秒请求数 低于预期值 20% 就要排查
响应时间 (P50/P95/P99) 百分位响应时间 P99 超过 3s 算异常
失败率 请求失败的比例 超过 1% 立刻停止

我个人习惯先跑一个小规模的测试，比如 10 个用户、每秒启动 2 个。观察几分钟，确认脚本没问题，再上大规模。

核心经验：Web UI 模式适合调试和初步验证。但正式压测时，我建议用无头模式（--headless），减少 UI 带来的性能开销。

4. 分布式压测

单机压测有个瓶颈——你的机器资源有限。当需要模拟上千甚至上万个用户时，单机就扛不住了。这时候就需要分布式压测。

Locust 的分布式架构很简单：一个 Master 节点 + 多个 Worker 节点。Master 负责协调，Worker 负责实际发请求。

架构图：下面这张 SVG 图展示了 Locust 分布式压测的核心逻辑。
Master 节点 协调 & 数据聚合 Worker 1 Worker 2 Worker N 目标服务器 发送请求 发送请求 发送请求 请求流量 请求流量 请求流量

启动分布式压测的步骤：

  1. 启动 Master 节点：
locust -f ai_inference_test.py --master --web-host 0.0.0.0 --web-port 8089
  1. 启动 Worker 节点：（可以在多台机器上分别启动）
locust -f ai_inference_test.py --worker --master-host=192.168.1.100

这里有几个坑，我踩过，你得注意：

  • 所有节点的脚本必须完全一致。Master 和 Worker 的 -f 参数指向同一个脚本文件。我曾经犯过傻，在 Worker 上改了脚本没同步，结果压测数据全乱了。
  • Worker 和 Master 之间的网络要通。默认端口是 5557（通信）和 5558（数据聚合）。记得在防火墙里放开。
  • Worker 的数量不是越多越好。我测试过，当 Worker 超过 20 个时，Master 的 CPU 会成为瓶颈。建议根据你的 Master 机器配置来定，一般 10-15 个 Worker 比较合理。
我的实战建议：分布式压测时，先在 2-3 台 Worker 上跑一轮，观察 Master 的 CPU 和内存使用。如果 Master 负载不高，再逐步增加 Worker。别一上来就搞 20 个 Worker，容易翻车。

5. 压测结果分析

压测跑完了，数据怎么看？Locust 的 Web UI 会生成几个关键图表：

  • 总请求数和 RPS 曲线：看系统能扛住多少并发
  • 响应时间分布：看 P95、P99 是否在可接受范围内
  • 失败率曲线：如果失败率突然飙升，说明系统扛不住了

我个人习惯把数据导出成 CSV，然后用 Python 做进一步分析。Locust 支持 --csv 参数：

locust -f ai_inference_test.py --headless -u 100 -r 10 --run-time 5m --csv=result

这样会生成 result_stats.csv 和 result_failures.csv。用 Pandas 读进来，画个图，一目了然。

一个小技巧：压测结束后，别急着关。等个 30 秒，让 Locust 把最后的统计数据刷完。不然 CSV 里可能缺最后几秒的数据。

好了，Locust 实战的核心内容就这些。从安装到脚本编写，从 Web UI 到分布式压测，每一步我都把实际项目中踩过的坑告诉你了。你照着做，应该能少走不少弯路。

想要更多资料，联系可可

← 上一章 📖 返回目录 下一章 →