Python Locust性能测试实战：从入门到分布式压测与瓶颈分析

1. 项目概述：为什么是Locust？

如果你做过性能测试，大概率用过JMeter或者LoadRunner。它们功能强大，但有时候你会觉得有点“重”——配置复杂、脚本编写不够灵活、资源消耗大。特别是当你需要快速验证一个API接口的并发能力，或者想用更贴近业务逻辑的代码来描述用户行为时，这些工具就显得不那么顺手了。

这就是Locust框架的价值所在。它是一个用Python编写的开源负载测试工具，其核心思想是“用代码定义用户行为”。这意味着，你的性能测试脚本就是一段纯粹的Python代码，你可以利用Python生态中所有的库（如requests, aiohttp, BeautifulSoup等）来模拟任何你能想到的复杂场景。从简单的HTTP请求到需要处理登录态、依赖前后接口、解析响应并做条件判断的完整业务流程，Locust都能优雅地支持。

我选择Locust作为深度实战的主题，是因为它在敏捷开发和DevOps流程中越来越受欢迎。它轻量、灵活，能与CI/CD管道无缝集成，并且其分布式压测能力非常强悍。更重要的是，对于已经熟悉Python的开发和测试同学来说，学习成本极低，几乎可以立刻上手产出价值。本次实战，我将带你从零开始，不仅学会如何使用Locust，更会深入其原理，分享我在大量压测实践中积累的“踩坑”经验和调优技巧，目标是让你能独立设计并执行一次专业的、数据可信的性能测试。

2. 核心需求解析与测试策略制定

性能测试不是漫无目的地“跑一下看看”，在动手写第一行Locust脚本之前，我们必须明确目标。这直接决定了后续的脚本设计、场景构造和结果分析。

2.1 明确性能测试目标与指标

通常，一次性能测试会关注以下几个核心目标中的一个或多个：

1. 负载测试：在预期的正常负载下，验证系统是否满足性能要求。例如，模拟100个用户同时浏览商品列表，持续10分钟，观察响应时间和错误率。
2. 压力测试：逐步增加负载，找到系统的性能瓶颈或最大处理能力。例如，从50用户开始，每2分钟增加50用户，直到错误率超过5%或响应时间超过阈值，此时的最大并发用户数即为一个关键瓶颈点。
3. 稳定性/耐力测试：在一定的负载压力下，长时间运行（如12小时、24小时），检查系统是否存在内存泄漏、资源耗尽等问题。
4. 尖峰测试：模拟负载在短时间内突然激增的场景，检验系统的弹性恢复能力。

对应这些目标，我们需要关注的关键性能指标包括：

• 吞吐量：系统每秒处理的请求数（Requests per Second, RPS）。这是衡量系统处理能力的核心指标。
• 响应时间：用户从发起请求到收到完整响应所经历的时间。我们通常关注其分布，如平均响应时间、90分位（P90）、95分位（P95）、99分位（P99）响应时间。P95响应时间为200ms意味着95%的请求都在200ms内完成，这比平均响应时间更能反映用户体验。
• 并发用户数：同时向系统发起请求的虚拟用户数量。注意，Locust中的“用户”是一个持续执行任务循环的实体，其并发数取决于孵化速率和总用户数设置。
• 错误率：失败请求数占总请求数的比例。任何非2xx或3xx的HTTP状态码，或者我们在代码中手动标记为失败的请求，都会计入错误。

注意：千万不要只盯着“平均响应时间”。一个平均响应时间很好看的系统，可能因为少数请求的严重延迟（长尾效应）而导致大量用户投诉。P95/P99响应时间才是保障大多数用户体验的黄金指标。

2.2 设计合理的测试场景

有了目标，就需要将其转化为Locust可执行的场景。场景设计的关键在于模拟真实的用户行为。

• 思考用户旅程：一个电商用户可能先访问首页，然后搜索商品，查看商品详情，加入购物车，最后下单支付。每个步骤的占比、思考时间（等待时间）都是不同的。
• 参数化与数据关联：不能让所有用户都搜索同一个关键词或使用同一个账号登录。你需要准备测试数据池（如CSV文件、数据库），让虚拟用户从中随机或顺序取用数据，避免缓存带来的性能假象。
• 设置合理的等待时间：使用Locust的wait_time属性来模拟用户操作间隔。between(1, 5)表示等待1到5秒之间的随机时间，这比固定间隔更贴近真实情况。

一个常见的误区是设计过于理想的场景，所有用户都一秒不停地发送请求。这会产生远超真实情况的压力，发现的瓶颈可能没有实际参考价值。我们的目标是用真实的负载模式，去发现真实的系统瓶颈。

3. Locust环境搭建与核心脚本编写

3.1 环境准备与安装

Locust的安装极其简单。强烈建议使用虚拟环境来管理依赖，避免污染全局Python环境。

# 1. 创建并进入项目目录 mkdir locust-performance-test && cd locust-performance-test # 2. 创建虚拟环境（以venv为例） python -m venv venv # 3. 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 4. 安装Locust pip install locust

安装完成后，可以通过locust -V检查版本。我当前使用的是Locust 2.x版本，它与1.x在Web UI和部分API上有较大变化，功能更强大，本文也将基于2.x进行讲解。

3.2 编写你的第一个Locust脚本

一个最基本的Locust脚本包含两个核心类：HttpUser（或其父类User）和定义在其中的TaskSet或@task装饰器。

我们来创建一个测试某个待办事项（Todo）API的例子，文件名为locustfile.py。

from locust import HttpUser, task, between class QuickstartUser(HttpUser): # 设置用户在每个任务执行后的等待时间范围（秒） wait_time = between(1, 3) # 标记为一个任务，weight权重表示任务执行的概率 @task(3) # 权重为3 def view_items(self): # 模拟查看待办事项列表 with self.client.get("/api/todos", catch_response=True) as response: # 可以自定义成功/失败的判断逻辑 if response.status_code == 200: # 甚至可以检查响应体内容 if "error" in response.text: response.failure("Found 'error' in response body") else: response.success() else: response.failure(f"Status code: {response.status_code}") @task(1) # 权重为1 def create_item(self): # 模拟创建一个新的待办事项 todo_data = { "title": f"Locust Test Task {self.user_id}", "completed": False } headers = {"Content-Type": "application/json"} self.client.post("/api/todos", json=todo_data, headers=headers) # 每个虚拟用户启动时执行一次，常用于登录等初始化操作 def on_start(self): # 假设登录接口，获取token login_response = self.client.post("/api/login", json={"username": "test", "password": "test"}) if login_response.status_code == 200: self.token = login_response.json().get("token") self.client.headers = {"Authorization": f"Bearer {self.token}"} else: # 如果登录失败，可以停止这个用户的任务执行 self.stop(force=True)

代码解析与实操要点：

1. HttpUser类：代表一类虚拟用户。self.client是其内置的HTTP客户端，基于requests.Session，会自动保持cookies，类似于一个浏览器会话。
2. @task装饰器：定义了用户要执行的具体任务。weight参数控制任务执行的相对频率。在上例中，view_items被调用的概率是create_item的3倍（3:1）。
3. wait_time：between(1, 3)使得每个任务执行后，用户会随机等待1到3秒，模拟用户思考或阅读时间。
4. on_start方法：每个用户实例开始运行前只执行一次。这里是放置登录逻辑的绝佳位置。登录成功后，我们将Token存入用户实例属性，并更新client的默认请求头，这样后续所有请求都会自动携带认证信息。
5. catch_response=True：这个参数允许我们手动控制请求的成功与失败。默认情况下，HTTP状态码非2xx的请求会被标记为失败。但有时服务器可能返回200，但响应体里包含业务错误信息。使用catch_response后，我们可以通过response.success()或response.failure()来更精细地定义请求结果，这对于测试API的健壮性至关重要。

4. 执行测试与Web UI深度解读

4.1 启动测试与命令行参数

在脚本所在目录下，执行以下命令启动Locust的Web界面：

locust

默认会启动在http://localhost:8089。你也可以通过命令行参数进行更多控制：

# 指定主机和端口 locust --host=https://your-api-server.com --web-host=0.0.0.0 --web-port=8090 # 无头模式运行（不启动Web UI，用于CI/CD） locust --headless --users 100 --spawn-rate 10 --run-time 5m --host=https://your-api-server.com # 指定不同的locustfile和用户类 locust -f my_locustfile.py MyCustomUserClass

常用参数说明：

• --host：被测试系统的基地址。
• --web-host,--web-port：控制Web UI绑定的主机和端口。
• --headless：无头模式，适合自动化。
• --users：要模拟的总用户数（峰值）。
• --spawn-rate：每秒孵化的用户数（用户增长速率）。
• --run-time：测试运行时间，如5m（5分钟）、1h30m。

4.2 Web UI界面核心功能详解

启动后，打开Web UI，你需要先填写参数才能开始测试。

1. 启动界面参数：

   • Number of users：模拟的最大总用户数。
   • Spawn rate：每秒启动多少个用户，直到达到总用户数。例如，用户数100，孵化率10，意味着需要10秒来启动所有用户。
   • Host：被测试系统的URL。如果命令行已指定，这里会自动填充。

2. 测试运行中的主面板：

   • Statistics：核心数据标签页。显示所有请求的聚合数据，包括请求类型、路径、请求数、失败数、平均响应时间、分位响应时间、最小/最大响应时间以及当前的RPS。
   • Charts：图表标签页。实时展示总RPS和响应时间随时间变化的曲线图。这是观察测试过程趋势最直观的方式。
   • Failures：失败请求详情。列出所有失败的请求，包括URL、错误信息和发生时间，方便快速定位问题。
   • Exceptions：Locust运行过程中捕获的Python异常。
   • Download Data：可以下载CSV格式的报告数据，用于后续更细致的分析。

3. 关键数据解读技巧：

   • 观察RPS曲线：在用户数达到预设值后，RPS应该趋于稳定。如果RPS持续下降，很可能说明系统正在恶化，处理能力在降低。
   • 关注失败率和异常：即使只有0.1%的失败率，在百万级请求下也是上千次错误，必须深究原因。
   • 对比分位响应时间：如果P99响应时间远高于P50，说明系统存在长尾延迟问题，需要优化慢请求。

5. 高级场景设计与实战技巧

5.1 复杂用户行为模拟

真实的用户行为远不止简单的GET/POST。Locust的灵活性在此体现得淋漓尽致。

场景一：顺序执行任务流使用SequentialTaskSet可以强制用户按顺序执行一系列任务。

from locust import HttpUser, SequentialTaskSet, task, between class UserJourney(SequentialTaskSet): def on_start(self): self.login() @task def login(self): # 登录逻辑 pass @task def browse_product(self): # 浏览商品 pass @task def add_to_cart(self): # 加入购物车，可能需要上一步的商品ID pass @task def checkout(self): # 结算 pass class WebsiteUser(HttpUser): tasks = [UserJourney] wait_time = between(2, 5)

场景二：参数化与数据关联从文件中读取测试数据，实现不同用户使用不同数据。

import csv from locust import HttpUser, task, between class ApiUser(HttpUser): wait_time = between(1, 2) def on_start(self): # 假设我们有一个CSV文件，包含username和password with open('user_credentials.csv', 'r') as f: reader = csv.DictReader(f) self.user_credentials = list(reader) self.current_user = None @task def login_and_get_profile(self): if not self.user_credentials: self.stop(force=True) return # 每次任务从列表中取一个用户（或随机取） self.current_user = self.user_credentials.pop() username = self.current_user['username'] password = self.current_user['password'] # 登录 login_resp = self.client.post("/login", json={"u": username, "p": password}) if login_resp.ok: token = login_resp.json()['token'] # 使用获取的token访问个人资料 headers = {"Authorization": f"Bearer {token}"} self.client.get("/profile", headers=headers)

场景三：处理异步请求或复杂逻辑Locust的任务函数是同步的，但你可以利用gevent（Locust底层使用的协程库）来封装异步操作，或者将复杂的、耗时的计算逻辑放在任务中。

from locust import HttpUser, task, between import random import time class ComplexUser(HttpUser): wait_time = between(0.5, 1.5) @task def complex_scenario(self): # 步骤1: 获取一个资源列表 list_resp = self.client.get("/api/resources") if list_resp.ok: resources = list_resp.json() if resources: # 随机选择一个资源 selected = random.choice(resources) resource_id = selected['id'] # 步骤2: 模拟一些“思考”或本地计算（这不会增加服务器压力，但占用用户时间） time.sleep(random.uniform(0.1, 0.5)) # 模拟客户端处理 # 步骤3: 对选中的资源进行操作 self.client.put(f"/api/resources/{resource_id}", json={"status": "updated"})

5.2 分布式压测实战

单台机器由于网络、CPU、端口数限制，可能无法产生足够大的压力。Locust原生支持分布式运行。

架构：一个主节点（Master） + 多个从节点（Worker）。

• Master：负责分发测试任务、收集汇总数据、提供Web UI。
• Worker：负责执行任务，生成实际的负载。

启动步骤：

1. 在主节点机器上启动Master（--master），并指定期望的Worker数量（可选）：

   locust --master --expect-workers=4 --host=https://your-api.com

2. 在每台从节点机器上启动Worker（--worker），并指定Master的地址（--master-host）：

   locust --worker --master-host=192.168.1.100 # 假设主节点IP是192.168.1.100

   • 确保所有Worker机器上都有相同的locustfile.py和Python环境。
   • Worker启动后，会在Master的Web UI上显示为已连接。

分布式压测的核心注意事项：

• 数据一致性：如果测试脚本需要读取外部数据文件（如CSV），必须确保该文件在所有Worker节点上的路径和内容一致。更好的做法是使用共享存储，或者在Master上启动一个简单的文件服务，让Worker在启动时拉取。
• 系统时间同步：所有Master和Worker节点的系统时间应基本同步，否则日志时间戳会混乱。
• 网络与防火墙：确保Master节点和所有Worker节点之间的网络通畅，特别是Master的端口（默认5557用于Worker通信，8089用于Web UI）需要能被Worker访问。

5.3 自定义客户端与测试非HTTP协议

Locust的核心是User类，HttpUser只是其一个特例。你可以继承User类，使用任何Python客户端库来测试其他协议，如WebSocket、gRPC、TCP/UDP自定义协议等。

from locust import User, task, between import websocket import json import time class WebSocketUser(User): wait_time = between(1, 5) def on_start(self): # 建立WebSocket连接 self.ws = websocket.WebSocket() self.ws.connect("ws://echo.websocket.org") self.ws.settimeout(5) @task def send_message(self): message = json.dumps({"time": time.time(), "user": self.user_id}) try: self.ws.send(message) response = self.ws.recv() # 接收响应 # 可以在这里根据响应内容判断成功/失败 # self.environment.events.request.fire(...) except Exception as e: # 标记请求失败 # self.environment.events.request.fire(...) pass def on_stop(self): # 关闭连接 if self.ws: self.ws.close()

关键点：对于非HTTP协议，你需要手动触发Locust的事件来记录请求的成功与失败，以便在统计中显示。这涉及到使用self.environment.events.request.fire()方法，传入request_type,name,response_time,response_length,exception等参数。

6. 结果分析与性能瓶颈定位

测试执行完毕，拿到一堆数据后，如何分析？

6.1 数据分析方法论

1. 看整体，定基调：首先看总RPS和平均响应时间是否达到预期目标。如果没达到，再看是哪个具体接口拖了后腿。
2. 找短板，定瓶颈：在Statistics页面，按平均响应时间或P95响应时间排序，找出最慢的几个接口。它们就是首要的优化目标。
3. 关联分析：结合系统监控（如服务器的CPU、内存、磁盘I/O、网络带宽、数据库连接数、慢查询日志）一起看。当Locust显示响应时间飙升时，服务器监控指标发生了什么变化？
   • 响应时间变长，同时CPU使用率100% -> 可能是应用代码性能问题或计算资源不足。
   • 响应时间变长，同时数据库服务器CPU高或慢查询增多 -> 瓶颈很可能在数据库。
   • 错误率突然升高，同时网络连接数或错误数飙升 -> 可能是网络问题、连接池耗尽或下游服务故障。

6.2 常见性能瓶颈模式与Locust表现

6.3 生成与解读测试报告

Locust Web UI提供了实时图表，但对于归档和分享，我们需要更正式的报告。

1. 导出CSV：在Web UI的“Download Data”选项卡，可以下载requests.csv和failures.csv等。你可以用Excel、Python Pandas或任何数据分析工具进行深入分析，比如绘制更美观的趋势图，或者计算在不同压力阶段的具体指标。
2. 使用--html报告：在无头模式运行时，可以生成一个HTML报告。

   locust --headless --users 1000 --spawn-rate 100 --run-time 10m --host=https://api.example.com --html report.html

   生成的report.html包含了关键指标的汇总和图表，比CSV更直观，适合直接附在测试报告邮件中。
3. 集成到CI/CD：在Jenkins、GitLab CI等工具中，你可以将上述无头模式命令作为构建步骤。通过分析命令的退出码（Locust在测试失败率超过某个阈值时可以非零退出）或解析生成的报告文件，来自动判断性能测试是否通过。

7. 常见问题排查与实战避坑指南

这一部分是我多年压测经验中积累的“血泪史”，很多问题文档里不会写，但一旦遇到就会耗费大量时间。

7.1 Locust自身相关问题

问题1：“Address already in use”或无法启动大量用户。

• 原因：操作系统可用端口耗尽。每个Locust用户（协程）会使用一个本地端口连接服务器。当并发数很高时（如几万），可能耗光端口范围。
• 解决：
  • 增加本地端口范围：

    # Linux sysctl -w net.ipv4.ip_local_port_range="1024 65535" # 永久生效需修改 /etc/sysctl.conf

  • 启用端口复用（对Locust客户端）：

    sysctl -w net.ipv4.tcp_tw_reuse=1

  • 最重要的：使用分布式压测。将负载分摊到多台Worker机器上，这是解决单机资源限制的根本方法。

问题2：测试过程中RPS上不去，但CPU/内存占用很低。

• 原因A：wait_time设置过长。如果每个用户执行一个任务后要等待好几秒，那么即使有大量用户，实际每秒发起的请求数（RPS）也会被限制。
• 排查：检查脚本中的wait_time。对于压力测试，可以暂时将其设置为constant(0)或很小的值，以产生最大压力。
• 原因B：被测试服务器响应太慢，导致Locust用户大量时间阻塞在等待响应上。
• 排查：查看Locust统计中的响应时间。如果平均响应时间接近或大于wait_time，说明系统处理不过来，用户都在排队等待响应，而不是在等待思考。这时需要先优化服务端性能。

问题3：出现“ConnectionResetError: [Errno 104] Connection reset by peer”等大量连接错误。

• 原因：服务端或中间件（如Nginx、负载均衡器）主动断开了连接。可能由于：
  1. 服务端连接超时设置过短。
  2. 负载均衡器的空闲连接超时。
  3. 服务端应用崩溃或重启。
• 解决：
  • 检查服务端和中间件的超时配置，适当调大。
  • 在Locust的HTTP客户端配置中，可以增加重试机制（需自定义客户端或使用requests.Session的适配器）。
  • 更重要的是，这些错误本身是性能测试要发现的问题，它指示了系统在高并发下的不稳定表现。

7.2 测试脚本与场景设计陷阱

陷阱1：数据未正确参数化，导致缓存命中率虚高。

• 现象：测试时性能极好，但上线后真实用户访问时性能骤降。
• 案例：所有虚拟用户都查询product_id=1的商品详情。这个数据很可能被数据库或应用缓存，所有请求都命中了缓存，没有对数据库产生真实压力。
• 避坑：务必使用足够多样化的测试数据。可以从生产环境脱敏后导入，或者用脚本生成海量仿真数据。

陷阱2：忽略资源清理，导致测试间相互影响。

• 现象：第一次测试结果正常，连续跑第二次、第三次测试时，错误率越来越高。
• 案例：测试脚本包含了“创建订单”的任务，但每次测试后没有清理测试数据。导致数据库中的订单表越来越大，后续查询和插入操作变慢。
• 避坑：
  • 为性能测试准备独立的数据库或表，测试前后有清理和初始化脚本。
  • 在测试数据中加入特定前缀或时间戳，便于识别和清理。
  • 使用on_stop钩子或单独的清理脚本在测试后执行清理工作。

陷阱3：断言过于严格或逻辑错误，误报大量失败。

• 现象：错误率异常高，但服务监控显示一切正常。
• 案例：在catch_response中，对响应JSON的某个字段做了硬编码断言，但该字段在生产环境中本就是可变的。

  # 错误示例 if response.json()['status'] != 'success': # ‘status’字段可能返回‘ok’、‘SUCCESS’等 response.failure("Status not success")

• 避坑：性能测试的断言应关注接口可用性和基本业务正确性，避免过于细致的业务逻辑校验。或者使用更灵活的匹配方式。

7.3 环境与配置问题

问题：分布式压测时，Worker节点数据不同步或启动失败。

• 检查清单：
  1. 网络：确保所有Worker能ping通Master的IP和端口（默认5557）。
  2. 防火墙：关闭或配置好Master和Worker机器上的防火墙规则。
  3. 代码与数据：所有Worker机器上的locustfile.py必须完全一致。如果脚本中通过相对路径读取文件，要确保该文件在所有Worker的相同路径下都存在。
  4. Python环境：Locust主版本、以及脚本依赖的第三方库（如requests）版本应尽量一致。
  5. 系统资源：检查Worker机器本身的CPU、内存、网络是否已成为瓶颈。可以用htop,iftop等工具监控。

一个实用的调试技巧：在Master启动时加上--logfile和--loglevel参数，在Worker启动时也加上，将日志输出到文件，这样能更清晰地看到连接和任务分发过程中的任何错误信息。

最后，性能测试是一个“测试-分析-优化-再测试”的循环过程。Locust给了你一把强大而灵活的锤子，但找准钉子（瓶颈）并正确地敲下去，更需要你对系统架构、业务逻辑和运维知识的综合理解。不要把性能测试当成一个一次性任务，而应将其作为持续交付流程中不可或缺的质量关卡。每次代码变更、基础设施调整后，都跑一下核心场景的性能测试，建立性能基线并监控其变化，这才是保障系统长期稳定运行的秘诀。