Locust性能测试实战：从零构建高并发压测脚本与分布式部署

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

如果你正在寻找一个能让你快速上手、用Python代码就能定义复杂压测场景的工具，Locust绝对值得你花时间研究。我最早接触它是在一个电商大促前的全链路压测项目里，当时团队需要一个能模拟真实用户购物行为（登录、浏览、加购、下单）的工具，并且希望压测脚本本身能像业务代码一样被版本管理和持续集成。我们对比了JMeter、Gatling等，最终选择了Locust，原因很简单：它用Python写脚本，对我们团队来说几乎没有学习成本；它基于事件驱动，单机就能模拟极高的并发；最关键的是，它的测试报告清晰直观，能直接定位到接口的响应时间和吞吐量瓶颈。

这几年，随着微服务和云原生架构的普及，性能测试不再是“上线前跑一下”的仪式，而是贯穿研发流程的必备环节。Locust凭借其轻量、灵活和可编程性，在DevOps和SRE团队中越来越受欢迎。这篇实战教程，我会结合我踩过的坑和积累的经验，从零开始带你搭建Locust环境，编写一个完整的压测脚本，分析测试结果，并分享一些在生产环境中大规模使用的进阶技巧。无论你是刚入行的测试工程师，还是需要为自己开发的服务做性能验证的后端开发，这篇文章都能给你提供一套可直接复用的“作战方案”。

2. Locust核心设计与思路拆解

2.1 事件驱动与协程：高并发的基石

很多人第一次用Locust，会被它单机轻松模拟数千上万用户的能力震惊。这背后的核心是事件驱动（Event-driven）和协程（Coroutine）模型。传统的多线程压测工具（如早期版本的JMeter）为每个虚拟用户（VU）创建一个操作系统线程或进程，当并发数达到几千时，线程切换和内存开销会成为瓶颈。Locust则不同，它使用gevent库，这是一个基于libev的高性能并发库。

简单来说，gevent通过“猴子补丁”（monkey patching）将Python标准库中的阻塞式I/O操作（如socket、time.sleep）替换成非阻塞版本。当一个虚拟用户在执行HTTP请求等待响应时（这是一个I/O等待操作），gevent会自动把CPU时间让给其他就绪的虚拟用户。这些虚拟用户都在同一个操作系统线程内运行，通过协程进行切换，切换成本极低。这就是为什么你用一个普通的笔记本电脑，就能轻松驱动几千个并发用户，而CPU和内存占用还很低。

注意：理解这一点至关重要。这意味着你在Locust脚本里写的time.sleep()，实际效果是“让出CPU给其他用户”，而不是真的阻塞住。但如果你在脚本中执行了长时间的计算密集型任务（比如一个复杂的加密算法），那就会真的阻塞整个事件循环，导致所有虚拟用户“卡住”。压测脚本的逻辑应该以发起网络请求和等待为主。

2.2 面向用户的测试思维：Task与User

Locust的脚本模型非常直观，它模拟的是用户行为，而不是单纯的“发请求”。这是它和很多其他工具哲学上的不同。在Locust中，核心是两个类：HttpUser（或其基类User）和TaskSet。

• User类：代表一类用户。你可以把它想象成一个“用户模板”。在这个类里，你定义这类用户的属性，比如他们等待每个任务执行完的间隔时间（wait_time），以及他们需要执行哪些任务（tasks）。
• TaskSet类：代表一组任务的集合。一个用户可以在不同场景下执行不同的任务集。例如，一个“浏览型用户”的任务集可能只包含查看商品列表和详情的任务，而一个“购买型用户”的任务集则包含登录、加购、下单等一系列任务。
• Task（任务）：用户具体做的动作，通常就是一个HTTP接口请求。任务通过Python的@task装饰器来定义，你可以给任务赋予权重，权重越高，被随机执行的概率就越大。

这种设计让编写贴近真实场景的压测脚本变得非常自然。你不再是在配置一个个孤立的HTTP请求，而是在描述“用户会怎么做”。例如，一个用户进入APP，他可能先浏览（权重3），然后搜索（权重1），偶尔才会下单（权重0.5）。这种权重比例可以直接映射到@task装饰器的参数上。

2.3 分布式与无状态：横向扩展的艺术

当需要模拟的用户数超过单台机器的能力（通常受限于网络带宽或CPU），Locust可以轻松地以主从（Master-Slave）模式进行分布式运行。Master节点负责分发测试任务、收集汇总数据；Slave节点（也叫Worker节点）负责运行虚拟用户，发起实际请求。

这里有一个关键点：Locust的分布式是无状态的。每个Slave节点独立运行自己的用户实例，它们之间不共享任何状态（如登录后的Session）。这意味着，如果你的测试场景依赖于用户登录态，你必须在每个虚拟用户实例中独立完成登录操作。这听起来有额外开销，但它带来了巨大的好处——扩展性极强。你可以动态地增加或减少Slave节点，Master会自动协调。在实际生产中，我们常用Docker或K8s来快速部署一整套分布式的Locust集群。

3. 从零开始：环境搭建与第一个脚本

3.1 环境准备与安装避坑

安装Locust非常简单，但有些细节不注意就会踩坑。强烈建议使用虚拟环境（venv或conda）来管理依赖，避免污染全局Python环境。

# 1. 创建并激活虚拟环境（以venv为例） python -m venv locust_env # Windows locust_env\Scripts\activate # Linux/Mac source locust_env/bin/activate # 2. 安装Locust pip install locust

安装完成后，在命令行输入locust --help，如果能看到帮助信息，说明安装成功。

实操心得：如果你在Windows上安装遇到关于gevent或greenlet的编译错误，大概率是因为缺少C++构建工具。最简单的解决方法是安装预编译的轮子（wheel）。可以先尝试升级pip，并使用pip install locust --prefer-binary。如果还不行，可以去 这个非官方网站 下载对应Python版本和系统位数的gevent和greenlet的.whl文件，用pip install xxx.whl本地安装，再安装Locust。

3.2 编写第一个压测脚本：模拟API访问

我们来创建一个最简单的脚本，压测一个假设的待测系统（SUT）的两个API：获取首页和查询用户信息。文件命名为locustfile.py，这是Locust默认寻找的入口文件。

from locust import HttpUser, task, between class QuickstartUser(HttpUser): # 用户执行完一个任务后，等待1到5秒（均匀分布） wait_time = between(1, 5) # 每个User实例启动时都会执行一次，用于初始化，如登录 def on_start(self): # 假设登录接口，获取token # response = self.client.post("/login", json={"username":"test", "password":"123"}) # self.token = response.json().get("token") # 这里我们先注释掉，因为待测系统可能没有 pass @task(3) # 权重为3，执行频率更高 def view_homepage(self): # self.client 是HttpUser内置的HttpSession实例，用法和requests库几乎一样 with self.client.get("/", catch_response=True) as response: # 可以自定义成功/失败的判断逻辑 if response.status_code == 200: response.success() else: response.failure(f"Unexpected status code: {response.status_code}") @task(1) # 权重为1 def view_user_profile(self): # 假设需要传递token在header中 headers = {"Authorization": f"Bearer {self.token}"} if hasattr(self, 'token') else {} with self.client.get("/api/user/1", headers=headers, catch_response=True) as response: if response.elapsed.total_seconds() > 0.5: # 如果响应时间超过500ms，标记为失败 response.failure(f"Response too slow: {response.elapsed}") elif response.status_code != 200: response.failure(f"Bad status code: {response.status_code}") else: response.success() # 每个User实例停止时执行一次 def on_stop(self): # 可以做清理工作，如登出 pass

代码关键点解析：

1. HttpUser：因为我们主要做HTTP测试，所以继承这个类最方便。
2. wait_time：定义了用户思考时间。between(1,5)是最常用的，表示均匀分布。还有constant(3)（固定3秒）和constant_pacing(2)（固定节奏，确保任务间隔至少2秒）。
3. @task：核心装饰器。权重比是3:1，意味着在长时间运行中，view_homepage被调用的次数大约是view_user_profile的3倍。
4. self.client：这是发起请求的核心对象。它支持get，post，put，delete等方法，接口设计和requests库高度一致，学习成本为零。
5. catch_response=True：这个上下文管理器允许你更精细地控制请求的成功与失败。默认情况下，HTTP状态码非2xx/3xx会被标记为失败。但有时业务上200返回错误码也算失败，或者5xx但你希望忽略，都可以在这里处理。
6. response.elapsed：这是一个timedelta对象，表示请求的耗时。这是性能测试中最重要的指标之一。

3.3 运行测试与Web UI解读

保存好locustfile.py后，在终端进入该文件所在目录，运行：

locust

默认会启动Web UI在http://localhost:8089。打开浏览器，你会看到启动界面。

1. 填写压测参数：

• Number of users：要模拟的总用户数。Locust会逐步启动这些用户，直到达到目标数。
• Spawn rate：每秒启动多少个用户。设置为10，意味着每秒新增10个用户，直到达到总用户数。
• Host：被测试系统的根地址，例如http://your-api-server.com。注意，脚本中的请求路径是相对于这个Host的。

填写后点击“Start swarming”，压测就开始了。

2. 核心数据看板解读：Web UI的“Statistics”标签页是核心，表格里的每一行代表一个被请求的接口（端点）。

列名	含义与解读
Type	请求方法（GET/POST等）。
Name	你在脚本中给请求定义的名称（如/api/user/1），如果没定义，会用URL路径。建议用name参数明确指定，如self.client.get("/api/user/1", name="/api/user/[id]")，这样相同模式的请求会被聚合统计。
Requests	当前总请求数。
Fails	失败请求数。失败率是衡量系统稳定性的黄金指标。
Median, 90%, 95%, 99%	响应时间的百分位数。重点关注90%或95%分位值（P90/P95），它表示90%/95%的请求响应时间低于这个值。这个指标比平均响应时间更能反映用户体验，因为它能剔除极端慢请求的影响。
Average	平均响应时间。参考价值一般，容易受极端值影响。
Min/Max	最小/最大响应时间。偶尔看看最大值，排查是否有“拖尾”请求。
Average size	平均响应体大小（字节）。
Current RPS	当前每秒请求数。这是系统吞吐量的直接体现。

3. 其他重要标签页：

• Charts：实时图表，可以看到RPS和响应时间随时间的变化趋势。理想状态是RPS曲线平稳，响应时间曲线平稳且低位。如果随着用户数增加，RPS上不去而响应时间飙升，说明系统遇到瓶颈。
• Failures：详细列出每个失败的请求、错误类型和发生时间，是排查问题的第一现场。
• Exceptions：脚本运行中抛出的Python异常。
• Download Data：可以下载CSV格式的测试报告，用于后续分析或归档。

4. 进阶实战：构建复杂业务场景

4.1 参数化与数据驱动：模拟真实用户多样性

上面的例子中，我们查询的用户ID是固定的1。现实中，用户ID、商品ID、搜索关键词都应该是变化的。Locust脚本是纯Python，因此我们可以用任何Python方式来实现参数化。

方法一：从列表/队列中取数据

from locust import HttpUser, task, between import queue class SearchUser(HttpUser): wait_time = between(2, 5) # 在类级别初始化一个队列，所有用户实例共享 search_keywords = queue.Queue() # 在模块加载时填充数据（实际项目中可从文件读取） for word in ["手机", "笔记本电脑", "耳机", "咖啡", "书籍"] * 100: search_keywords.put(word) @task def search_product(self): # 从队列中取出一个关键词，如果队列为空，则任务结束 try: keyword = self.search_keywords.get_nowait() except queue.Empty: # 可以在这里停止这个用户，或者跳过任务 self.stop(force=True) return with self.client.get(f"/api/search?q={keyword}", name="/api/search", catch_response=True) as resp: # ... 处理响应 # 可选：如果测试需要循环使用数据，可以把数据再放回队列尾部 # self.search_keywords.put(keyword)

方法二：使用Python内置的随机函数

import random from locust import HttpUser, task, between class RandomUser(HttpUser): user_ids = list(range(1000, 2000)) # 假设有1000个用户ID @task def view_random_user(self): uid = random.choice(self.user_ids) with self.client.get(f"/api/user/{uid}", name="/api/user/[id]", catch_response=True) as resp: if resp.status_code == 404: # 对于404，我们可以选择标记为成功（因为用户可能不存在），也可以标记为失败 resp.success() # 取决于你的测试目标

注意事项：参数化数据源的选择很重要。如果所有用户都从一个共享列表里随机取，可能会造成“热点”数据被频繁访问，不符合真实分布。对于大规模压测，更真实的做法是为每类用户预分配一个独立的数据池，或者从文件中按行读取。

4.2 处理关联接口：保持用户会话

很多业务流需要保持状态，比如先登录拿到token，后续请求都要带上。我们可以在on_start方法中完成登录，并将凭证保存在用户实例的属性中。

from locust import HttpUser, task, between import json class AuthenticatedUser(HttpUser): wait_time = between(1, 3) host = "http://your-auth-server.com" def on_start(self): # 登录，获取token login_payload = {"username": "test_user", "password": "secure_pass"} with self.client.post("/api/login", json=login_payload, name="01_Login") as resp: if resp.status_code == 200: self.auth_token = resp.json()["data"]["token"] self.headers = {"Authorization": f"Bearer {self.auth_token}"} else: # 如果登录失败，记录异常并停止这个用户 resp.failure("Login failed") self.stop(force=True) # 强制停止该用户实例 @task(5) def get_profile(self): # 使用保存在实例中的headers self.client.get("/api/profile", headers=self.headers, name="02_GetProfile") @task(1) def create_order(self): order_data = {"product_id": 123, "quantity": 1} with self.client.post("/api/order", json=order_data, headers=self.headers, name="03_CreateOrder") as resp: if resp.status_code == 201: self.order_id = resp.json()["order_id"] else: resp.failure(f"Create order failed: {resp.text}") # 可以基于上一个请求的结果发起新请求 @task def view_my_order(self): if hasattr(self, 'order_id'): self.client.get(f"/api/order/{self.order_id}", headers=self.headers, name="04_ViewOrder")

4.3 使用TaskSet组织复杂场景

当用户行为有多组不同的模式时，TaskSet就派上用场了。例如，一个用户可能大部分时间在“浏览模式”，偶尔进入“购买模式”。

from locust import HttpUser, task, between, TaskSet class BrowseTasks(TaskSet): # 浏览任务集内的任务权重是独立的 @task(10) def view_item_list(self): self.client.get("/items") @task(2) def view_item_detail(self): self.client.get("/items/123") # 可以嵌套跳出到父级User的其他TaskSet @task(1) def stop_browsing(self): self.interrupt() # 关键！用于跳出当前TaskSet class PurchaseTasks(TaskSet): def on_start(self): # 进入购买流程前，先确保有商品在购物车 self.client.post("/cart", json={"item_id": 456}) @task def checkout(self): self.client.post("/checkout") @task(1) def stop_purchasing(self): self.interrupt() class WebsiteUser(HttpUser): wait_time = between(1, 5) # tasks是一个列表，元素可以是可调用对象，或者是(TaskSet类, 权重)的元组 tasks = [BrowseTasks, PurchaseTasks] # 用户有50%概率进入BrowseTasks，50%进入PurchaseTasks # 如果想赋予不同权重：tasks = [(BrowseTasks, 3), (PurchaseTasks, 1)] # 3:1的概率

关键点：在TaskSet内部使用self.interrupt()是跳出当前任务集、返回父级调度器的标准方式。如果不调用interrupt()，用户将一直执行当前TaskSet内的任务，不会切换到其他TaskSet或User级别的task。

5. 生产级部署与性能调优

5.1 无Web UI运行与结果导出

在CI/CD流水线或服务器上运行时，我们通常不需要Web UI，而是以无头（headless）模式运行，并将结果导出。

# 基本无头模式运行，指定用户数、孵化率、运行时间 locust -f locustfile.py --headless --users 100 --spawn-rate 10 --run-time 1m --host=http://your-server.com # 更常用的：指定运行时间，并导出结果文件 locust -f locustfile.py \ --headless \ --users 500 \ --spawn-rate 20 \ --run-time 5m \ --csv=results/my_test \ # 生成CSV报告的前缀 --csv-full-history \ # 记录整个运行期间的历史数据 --host=http://your-server.com

运行后，会生成一系列CSV文件：

• my_test_stats.csv: 总体统计信息（与Web UI的表格对应）。
• my_test_failures.csv: 失败请求详情。
• my_test_history.csv: 如果指定了--csv-full-history，会生成每秒的RPS和响应时间数据，用于绘制趋势图。

5.2 分布式部署实战

单机Locust受限于网络端口和CPU，一般能模拟几千到上万的并发。要模拟十万级并发，必须分布式部署。

步骤1：启动Master节点Master节点不模拟用户，只负责协调和收集数据。

locust -f locustfile.py --master --host=http://your-server.com

步骤2：启动一个或多个Worker节点Worker节点需要能访问到locustfile.py和Master节点（默认端口8089）。

# 在另一台机器或另一个终端 locust -f locustfile.py --worker --master-host=<MASTER_IP_ADDRESS>

你可以启动任意多个Worker，它们会自动向Master注册。

步骤3：在Web UI或命令行控制压测此时，访问Master节点的Web UI（http://<MASTER_IP>:8089），启动压测。你会发现，你可以设置的总用户数上限变大了，因为资源由所有Worker共同提供。

踩坑记录：分布式运行时，确保所有Worker节点上的locustfile.py是完全一致的，特别是共享数据源（如队列）的初始化逻辑。否则可能出现不可预知的行为。建议将压测脚本放在版本库中，在所有节点统一拉取。

5.3 Locust脚本性能调优

即使有gevent加持，写得不好的Locust脚本也可能成为瓶颈。以下是一些优化点：

1. 避免在任务函数中进行阻塞操作或繁重计算：如前所述，这会阻塞整个事件循环。如果必须做，使用gevent.spawn或gevent.pool将其放到单独的greenlet中执行。
2. 谨慎使用time.sleep：虽然gevent将其替换为非阻塞的，但大量虚拟用户频繁sleep也会消耗资源。对于精确的节奏控制，考虑使用constant_pacing。
3. 优化请求验证逻辑：catch_response=True和复杂的响应体解析（如解析大的JSON）会有开销。在高压下，如果只关心请求是否成功发出，可以简化验证逻辑，甚至对某些接口关闭catch_response。
4. 使用连接池：requests.Session（self.client底层使用）会自动保持HTTP连接复用。确保你没有在每个请求中创建新的Session。
5. 监控Locust Runner进程本身：在压测过程中，用top或htop观察Locust进程的CPU和内存占用。如果单个Worker的CPU持续超过80%，可能就需要增加Worker节点了。

6. 常见问题排查与调试技巧

6.1 性能测试结果分析“红宝书”

看到测试报告，如何判断系统是否有问题？以下是一个快速自查清单：

现象	可能原因	排查方向
失败率（Fails）高	1. 被测系统容量不足，返回5xx错误。 2. 脚本断言条件过于严格（如对响应时间要求极高）。 3. 网络问题或测试机资源耗尽。	1. 查看失败详情（Failures标签），看具体错误信息（如502 Bad Gateway, 504 Timeout）。 2. 检查被测系统监控（CPU、内存、数据库连接池、线程池）。 3. 检查Locust Worker节点资源使用情况。
响应时间（P95）随用户数增长而线性飙升	系统存在明显瓶颈，无法处理当前并发。可能是： 1. 数据库慢查询。 2. 外部依赖服务响应慢。 3. 应用服务器线程池打满。 4. 代码中存在同步锁或资源竞争。	1. 关联查看系统监控，定位是CPU、内存、I/O还是数据库指标先达到瓶颈。 2. 使用APM工具（如SkyWalking, Pinpoint）或Profiler分析应用内部调用链耗时。 3. 检查数据库监控，看是否有全表扫描或死锁。
RPS（每秒请求数）上不去，但响应时间正常	1. 压测端达到瓶颈（网络带宽、端口数、Locust单机性能）。 2. 被压测接口本身处理能力有限（可能是单线程处理或有限队列）。 3. 脚本中wait_time设置过长，用户“思考”时间太多。	1. 检查Locust Worker的CPU和网络流量。尝试分布式压测。 2. 检查被测服务是否部署了多个实例，负载是否均衡。 3. 调整wait_time，使用constant_pacing模式控制最小请求间隔。
响应时间波动大（毛刺多）	1. 垃圾回收（GC）停顿。 2. 系统中有后台定时任务或批处理作业干扰。 3. 网络抖动。 4. 依赖的缓存服务（如Redis）偶尔慢查询。	1. 观察系统GC日志。 2. 检查在压测期间是否有其他运维操作或批处理。 3. 对比多轮测试，看毛刺是否在固定时间出现。

6.2 Locust脚本调试技巧

1. 打印日志：在脚本中使用Python标准的logging模块或直接print。注意，在分布式模式下，日志会输出在各自Worker的终端。可以通过--logfile参数将日志重定向到文件。

   import logging logger = logging.getLogger(__name__) @task def my_task(self): logger.info("Starting a task") # ... 请求 logger.debug(f"Response time: {response.elapsed}")

   运行Locust时，可以通过--loglevel DEBUG来输出更详细的日志。

2. 使用catch_response进行断言调试：这是最强大的调试手段之一。你可以在with块内对响应内容做任何检查，并灵活地标记成功或失败。

   with self.client.post("/api", json=data, catch_response=True) as resp: # 检查状态码 if resp.status_code != 200: resp.failure(f"Status code error: {resp.status_code}") return # 检查JSON结构 try: json_data = resp.json() if json_data.get("code") != 0: resp.failure(f"Business logic error: {json_data.get('msg')}") else: resp.success() except json.JSONDecodeError: resp.failure("Response is not valid JSON")

3. 先以单个用户、慢速模式运行：使用--users 1 --spawn-rate 1启动，在Web UI中观察单个用户的请求流是否正确，所有关联逻辑（登录、带Token请求、参数传递）是否正常。这是验证脚本逻辑最安全的方式。

6.3 与持续集成（CI）集成

将Locust集成到CI中，可以为每次代码提交或每日构建提供性能基线。一个简单的GitLab CI / GitHub Actions配置示例如下：

# .gitlab-ci.yml 示例 performance_test: stage: test image: python:3.9-slim before_script: - pip install locust script: - | locust -f locustfile.py \ --headless \ --users 50 \ --spawn-rate 5 \ --run-time 2m \ --csv=perf_report \ --host=$TEST_ENVIRONMENT_HOST \ --only-summary # 只输出最终摘要，减少日志噪音 artifacts: paths: - perf_report*.csv expire_in: 1 week rules: - if: $CI_PIPELINE_SOURCE == "merge_request_event" # 仅在合并请求时运行

在CI中，你可以设定性能合格线（如P95响应时间<500ms，失败率<0.1%），如果测试结果不达标，则让CI任务失败，阻止代码合并。

7. 超越基础：定制化与扩展

Locust的强大之处在于它的可扩展性。因为核心逻辑就是Python，你可以做任何事。

自定义客户端：如果你需要测试非HTTP协议，比如WebSocket、gRPC、TCP，你可以继承User类，并自己实现client属性。社区已经有很多相关插件，如locust-grpc、locust-socketio。

生成自定义统计信息：你可以使用事件钩子（events）来收集自定义指标。例如，记录每个订单的金额，并在测试结束后计算总交易额。

from locust import events from locust.runners import MasterRunner order_amounts = [] @events.request.add_listener def on_request(request_type, name, response_time, response_length, exception, context, **kwargs): # 可以在这里记录每一次请求的详细信息 pass @events.test_stop.add_listener def on_test_stop(environment, **kwargs): # 测试结束时触发 if isinstance(environment.runner, MasterRunner): print(f"Total orders simulated: {len(order_amounts)}") print(f"Total transaction volume: {sum(order_amounts)}")

集成到监控系统：你可以将Locust的实时统计数据通过events.report_to_master或events.worker_report发送到你的监控系统（如Prometheus），实现压测数据的可视化大盘。

经过这些年的实战，Locust对我来说已经不仅仅是一个压测工具，更像是一个用代码定义“用户行为模型”的仿真平台。它的学习曲线平缓，但上限极高。从简单的API压测到复杂的全链路业务场景模拟，从单机运行到千级节点的分布式集群，它都能很好地胜任。最关键的是，它让性能测试脚本化、工程化，成为了开发流程中自然的一环。下次当你需要评估系统容量或寻找性能瓶颈时，不妨从写一个简单的locustfile.py开始。