压力测试 Locust

第1章：Locust 概述与核心概念

1.1 性能测试基础概念

概念名称	说明	注意事项
并发用户数（Concurrent Users）	同时向系统发送请求的虚拟用户数量，模拟真实用户并发访问。	并发数过高可能导致压测机资源耗尽，需合理设置。
请求率（Requests per Second, RPS）	系统每秒处理的请求数量，反映系统吞吐能力。	RPS 受网络、服务器性能、客户端资源等多因素影响。
响应时间（Response Time）	从发送请求到接收到完整响应所花费的时间，通常以毫秒（ms）为单位。	应关注平均值、中位数、95% 和 99% 分位值，避免被极端值误导。
吞吐量（Throughput）	单位时间内系统处理的数据量，如 MB/s 或请求数/秒。	与 RPS 相关，但更侧重于数据传输量。
错误率（Error Rate）	请求失败的比例，如 HTTP 5xx、超时、断言失败等。	高错误率通常意味着系统存在性能瓶颈或逻辑缺陷。
资源利用率	被测系统在压力下的 CPU、内存、磁盘 I/O、网络等资源使用情况。	需结合监控工具（如 Prometheus）观察系统瓶颈。
压力测试（Stress Testing）	逐步增加负载，直到系统崩溃，以确定系统极限。	用于发现系统在极端情况下的表现和恢复能力。
负载测试（Load Testing）	在预期负载下测试系统性能，验证是否满足性能需求。	是最常见的性能测试类型，用于容量规划。

1.2 Locust 简介与特点

概念名称	说明	注意事项
Locust 定义	一个基于 Python 的开源负载测试工具，使用协程（gevent）实现高并发。	需了解 Python 基础语法才能高效编写测试脚本。
编程方式定义测试	用户行为通过编写 Python 脚本定义，灵活且可复用。	相比 GUI 工具，学习曲线略高，但扩展性强。
分布式支持	支持 Master-Worker 架构，可横向扩展压测能力。	需确保网络通畅，Master 与 Worker 版本一致。
实时 Web UI	提供直观的 Web 界面，实时查看测试进度和性能指标。	Web UI 仅用于监控，实际压测由 Python 脚本驱动。
高可扩展性	可自定义用户行为、协议、事件处理等。	可结合其他 Python 库（如 requests、Faker）增强功能。
轻量级	无需复杂配置，安装后即可快速启动测试。	对压测机资源有一定要求，尤其是高并发场景。

1.3 Locust 与其他工具对比（JMeter、Gatling等）

对比维度	Locust	JMeter	Gatling
编写方式	Python 脚本（代码驱动）	GUI 配置 + XML 存储（可视化驱动）	Scala DSL（代码驱动）
学习难度	中等（需 Python 基础）	低（图形化操作）	高（需 Scala 基础）
并发模型	基于 gevent 协程，轻量高效	基于 Java 线程，资源消耗较高	基于 Akka Actor 模型，高并发能力强
扩展性	高（Python 生态丰富）	高（支持插件）	高（支持自定义组件）
分布式支持	支持 Master-Worker 模式	支持 Master-Slave 模式	支持分布式压测
实时监控	内置 Web UI，实时图表	需插件或外部工具	内置 HTML 报告，实时性一般
适用场景	灵活定制、CI/CD 集成、开发人员使用	功能测试与性能测试结合、非开发人员使用	高性能、复杂场景压测
脚本维护	易于版本控制（纯代码）	XML 文件，版本控制较难	代码形式，易于维护
社区支持	活跃，文档完善	非常活跃，资料丰富	活跃，但相对小众

1.4 Locust 架构原理（Master-Worker 模式）

组件名称	说明	注意事项
Master 节点	控制整个测试流程，接收 Worker 的数据，汇总结果，并提供 Web UI。	必须先启动 Master；可运行在任意机器上。
Worker 节点	执行实际的压测任务，模拟用户行为，将性能数据发送给 Master。	可部署多个 Worker 以提升并发能力；需连接到 Master。
协程（Greenlet）	Locust 使用 gevent 实现协程，单线程内模拟大量并发用户。	协程轻量，避免阻塞操作（如 time.sleep），应使用 self.wait()。
消息通信	Master 与 Worker 通过 TCP 或 ZMQ 通信（默认 TCP）。	确保防火墙开放对应端口（默认 5557、5558）。
数据聚合	Worker 定期将请求统计、错误信息等发送给 Master 进行汇总。	数据同步有轻微延迟，但不影响整体趋势分析。
测试启动流程	1. 启动 Master；2. 启动一个或多个 Worker；3. 通过 Web UI 或命令行启动测试。	Worker 必须在 Master 启动后启动，否则无法注册。

第2章：环境搭建与快速入门

2.1 安装 Locust

操作步骤	操作细节	注意事项
步骤1：安装 Python	确保系统已安装 Python 3.7+，推荐使用虚拟环境。	建议使用 python --version 验证版本。
步骤2：创建虚拟环境（可选）	python -m venv locustenv source locustenv/bin/activate（Linux/Mac） locustenv\Scripts\activate（Windows）	使用虚拟环境可避免依赖冲突。
步骤3：安装 Locust	pip install locust	可使用清华源加速：pip install locust -i https://pypi.tuna.tsinghua.edu.cn/simple
步骤4：验证安装	locust --help 或 locust --version	若命令未找到，检查 pip 是否安装到正确环境。
步骤5：安装可选依赖（如 fasthttp）	pip install locust[fasthttp]	使用 FastHttpUser 可提升 HTTP 性能。

2.2 第一个 Locust 脚本（Hello World）

方法/元素	语法	用途	代码示例	注意事项
HttpUser	class MyUser(HttpUser):	定义基于 HTTP 的用户类，继承后可发起 HTTP 请求。	class WebsiteUser(HttpUser): wait_time = between(1, 5)	必须继承 HttpUser 才能使用 client 属性。
@task	@task def hello_world(self):	标记一个方法为可执行任务，Locust 会随机调用这些任务。	@task def load_page(self): self.client.get("/")	可设置权重：@task(3) 表示该任务被调用概率是其他任务的 3 倍。
between()	wait_time = between(1, 5)	设置用户在任务之间等待的随机时间范围（秒）。	wait_time = between(2, 8)	需导入：from locust import between
self.client.get()	self.client.get("/path")	发起 HTTP GET 请求。	self.client.get("/api/users")	client 由 HttpUser 提供，自动管理会话和 Cookie。
self.client.post()	self.client.post("/path", data={})	发起 HTTP POST 请求。	self.client.post("/login", json={"user":"admin"})	支持 data、json、headers 等参数。

2.3 运行模式：Web UI 模式 vs 命令行模式

运行模式	操作细节	用途	注意事项
Web UI 模式	在脚本所在目录执行 locust，打开浏览器访问 http://localhost:8089	交互式启动测试，适合调试和演示。	可动态设置用户数、spawn rate；实时查看图表。
命令行模式（无 UI）	locust -f locustfile.py --headless -u 100 -r 10 --run-time 10m	自动化执行，适合 CI/CD 集成。	必须指定 -u（用户数）、-r（每秒启动用户数）、--run-time（运行时长）。
headless 模式退出码	测试结束后返回退出码：0 表示成功，非 0 表示失败（如错误率超限）。	用于自动化判断测试是否通过。	可结合 --exit-code-on-error 控制失败行为。
分布式运行（Master）	locust -f locustfile.py --master	启动 Master 节点，协调多个 Worker。	默认端口 5557（客户端）和 5558（Worker）。
分布式运行（Worker）	locust -f locustfile.py --worker --master-host=192.168.1.100	启动 Worker 节点，执行压测任务。	--master-host 指定 Master 的 IP 地址。

2.4 查看测试报告与结果解读

指标名称	说明	注意事项
Type	请求类型，如 GET、POST。	用于区分不同接口的性能表现。
Name	请求路径或自定义名称。	可在请求中使用 name="login" 参数统一命名动态 URL。
Request Count	总请求数。	结合 RPS 可分析系统处理能力。
Failures	失败请求数及错误率。	错误率 > 0% 需排查原因（网络、服务、断言等）。
Median (ms)	响应时间中位数。	比平均值更能反映典型用户感受。
90% / 95% / 99% Percentile	90%/95%/99% 的请求响应时间低于该值。	用于评估极端情况下的用户体验。
Avg (ms)	平均响应时间。	易受异常值影响，需结合分位数分析。
Min / Max (ms)	最小和最大响应时间。	Max 值过高可能表示存在性能瓶颈。
Average Size (bytes)	响应体平均大小。	用于估算带宽消耗。
Current RPS	当前每秒请求数。	实时波动，观察是否稳定。
Total RPS	整体平均请求率。	反映系统整体吞吐能力。
Errors 列表	显示具体错误类型和次数（如 500、Timeout）。	点击可查看错误详情，便于定位问题。
Charts（Web UI）	包括 RPS、响应时间、用户数等实时图表。	用于直观分析性能趋势。

第3章：核心类与基础语法

3.1 User 类：用户行为的抽象

方法/属性	语法	用途	代码示例	注意事项
User	class MyUser(User):	Locust 中所有用户类的基类，用于定义虚拟用户行为。	from locust import User class ApiUser(User): wait_time = between(1, 3)	通常不直接使用，推荐继承 HttpUser。
wait_time	wait_time = between(1, 5)	定义用户在执行任务之间的等待时间策略。	wait_time = constant(2)	必须设置，否则用户将无间隔连续执行任务。
tasks	tasks = [task1, task2] 或 tasks = {task1: 3, task2: 1}	定义用户可执行的任务列表或字典（含权重）。	tasks = [view_page, login]	使用字典可设置任务调用概率权重。
fixed_count	class AdminUser(User): fixed_count = 5	指定该用户类的固定实例数量（在分布式中使用）。	fixed_count = 10	需配合 --users 总数合理设置，避免资源浪费。

3.2 TaskSet 类（已弃用）与 @task 装饰器

概念	说明	注意事项
TaskSet 类	旧版本中用于组织用户任务的类，支持嵌套任务集。	自 Locust 1.0 起已弃用，不推荐在新项目中使用。
@task 装饰器替代 TaskSet	现在直接在 User 子类中使用 @task 定义任务，无需 TaskSet。	所有任务方法直接定义在 User 类中，结构更简洁。
迁移建议	将原 TaskSet 中的任务方法移至 User 子类，并添加 @task。	避免混合使用旧模式，确保代码兼容性。

3.3 @task 装饰器的使用

方法	语法	用途	代码示例	注意事项
@task	@task def my_task(self):	标记一个方法为 Locust 可调度的任务。	@task def browse_home(self): self.client.get("/")	必须定义在 User 子类中。
@task(weight)	@task(3) def login(self):	设置任务执行的相对权重，权重越高越频繁执行。	@task(2) def search(self): self.client.get("/search?q=test")	权重为整数，默认为 1。
多个 @task	@task(1) def read(self): ... @task(2) def write(self): ...	定义多个任务，Locust 随机选择执行。任务按权重比例调度，如 1:2，则 write 调用概率是 read 的 2 倍。	任务之间应独立，避免强依赖。
无参数任务	@task def ping(self): self.client.get("/ping")	最常见的任务形式，执行单一操作。	适合用于简单接口压测。可结合 wait_time 模拟真实用户间隔。

3.4 wait_time 函数：设置用户等待时间

方法	语法	用途	代码示例	注意事项
between(min, max)	wait_time = between(1, 5)	用户在 min 到 max 秒之间随机等待。	wait_time = between(2, 8)	最常用策略，模拟真实用户思考时间。
constant(seconds)	wait_time = constant(3)	用户每次任务后固定等待指定秒数。	wait_time = constant(1)	适用于节奏一致的自动化场景。
constant_pacing(seconds)	wait_time = constant_pacing(5)	确保每个任务周期至少为指定秒数，若任务执行快则等待补足。	wait_time = constant_pacing(10)	保证最小执行间隔，避免过快请求。
constant_throughput(rps)	wait_time = constant_throughput(2)	控制用户每秒最多执行 1/rps 秒的间隔，实现恒定吞吐量。	wait_time = constant_throughput(0.5)	rps 为每秒请求数，如 0.5 表示每 2 秒一个请求。
自定义 wait_time	def my_wait_time(self): return 1 + random.random() wait_time = my_wait_time	自定义函数返回等待时间（秒）。	def slow_down(self): return random.expovariate(1/10) wait_time = slow_down	函数必须返回数值，可实现复杂行为模型。

3.5 on_start() 与 on_stop() 生命周期方法

方法	语法	用途	代码示例	注意事项
on_start()	def on_start(self):	用户启动时执行一次，通常用于登录、初始化等操作。	def on_start(self): self.login() def login(self): self.client.post("/login", json={"u":"admin","p":"123"})	仅执行一次，适合建立会话状态。
on_stop()	def on_stop(self):	用户停止前执行一次，可用于清理资源、登出等。	def on_stop(self): self.client.post("/logout")	并非总是被调用（如强制终止时），不保证执行。
调用时机	on_start() 在第一个任务前调用 on_stop() 在最后一个任务后调用	控制用户生命周期的关键节点。	结合 @task 使用，构建完整用户行为流。	避免在 on_start 中执行耗时过长操作，影响压测启动。

第4章：HTTP 请求测试实战

4.1 使用 client 发起 HTTP 请求（HttpUser）

方法/属性	语法	用途	代码示例	注意事项
HttpUser	class MyUser(HttpUser):	继承后可使用 self.client 发起 HTTP 请求，自动管理 Session。	from locust import HttpUser class WebUser(HttpUser): host = "https://api.example.com"	是最常用的用户基类。
self.client	self.client.get("/path")	HttpUser 提供的客户端实例，用于发送 HTTP 请求。	response = self.client.get("/")	自动处理 Cookie、连接池、重定向等。
host 属性	host = "https://example.com"	设置默认主机地址，避免在每次请求中重复写域名。	host = "https://jsonplaceholder.typicode.com"	请求路径为相对路径时使用。
catch_response	self.client.get("/", catch_response=True)	手动控制请求成功/失败判断，用于自定义断言。	with self.client.get("/health", catch_response=True) as r: if r.status_code == 200 and "OK" in r.text: r.success() else: r.failure("Health check failed")	必须配合 with 语句和 r.success()/r.failure() 使用。

4.2 GET 请求：参数传递与响应处理

方法	语法	用途	代码示例	注意事项
client.get()	self.client.get(path)	发起 GET 请求。	self.client.get("/users")	最基础的 HTTP 请求方式。
params 参数	self.client.get("/search", params={"q": "test", "page": 1})	传递 URL 查询参数，自动编码。	params = {"category": "books", "sort": "price"} self.client.get("/products", params=params)	推荐使用字典形式传递参数。
响应状态码	response.status_code	获取 HTTP 响应状态码。	r = self.client.get("/user/1") assert r.status_code == 200	可用于断言或条件判断。
响应文本	response.text	获取响应体文本内容（字符串）。	if "welcome" in r.text: ...	适用于 HTML、JSON 字符串等。
响应 JSON	response.json()	解析响应体为 Python 对象（dict/list）。	data = r.json() assert data["id"] == 1	若响应非 JSON 格式会抛出异常，建议加 try-except。
响应时间	response.elapsed.total_seconds()	获取请求耗时（秒）。	duration = r.elapsed.total_seconds()	可用于日志记录或性能分析。

4.3 POST 请求：表单、JSON、文件上传

方法	语法	用途	代码示例	注意事项
data 参数	self.client.post("/login", data={"username": "admin", "password": "123"})	发送表单数据（application/x-www-form-urlencoded）。	r = self.client.post("/submit", data={"name": "test", "email": "t@t.com"})	类似 HTML 表单提交。
json 参数	self.client.post("/api/users", json={"name": "John"})	发送 JSON 数据，自动设置 Content-Type 为 application/json。	payload = {"title": "New Post", "body": "Content"} self.client.post("/posts", json=payload)	推荐用于 API 测试。
files 参数	self.client.post("/upload", files={"file": open("test.txt", "rb")})	上传文件，Content-Type 为 multipart/form-data。	files = {"avatar": ("avatar.jpg", open("a.jpg", "rb"), "image/jpeg")} self.client.post("/upload", files=files)	注意文件打开模式为二进制（rb），并指定 MIME 类型。
多文件上传	files={"file1": ..., "file2": ...}	同时上传多个文件。	files = {"f1": ("1.txt", open("1.txt","rb")), "f2": ("2.txt", open("2.txt","rb"))}	确保文件路径正确且可读。

4.4 设置请求头（headers）与认证（Authentication）

方法	语法	用途	代码示例	注意事项
headers 参数	self.client.get("/api", headers={"Authorization": "Bearer token123"})	为单个请求设置自定义请求头。	headers = {"X-API-Key": "abc123", "User-Agent": "Locust"} self.client.get("/data", headers=headers)	优先级高于全局 headers。
全局 headers	default_headers = {"Authorization": "..."}	在 User 类中设置默认请求头。	class ApiUser(HttpUser): default_headers = {"Authorization": "Basic abc"}	所有请求自动携带，适合认证信息。
Basic Auth	auth=("user", "pass")	使用 HTTP Basic 认证。	self.client.get("/secure", auth=("admin", "pass"))	自动编码为 Base64 放入 Authorization 头。
Bearer Token	headers={"Authorization": "Bearer <token>"}	使用 Token 认证（如 JWT）。	token = "eyJhbGciOiJIUzI1NiIs..." self.client.get("/profile", headers={"Authorization": f"Bearer {token}"})	通常在 on_start 中获取 token 后设置。

4.5 处理 Cookie 与 Session

方法	语法	用途	代码示例	注意事项
Session 管理	self.client.get("/login") self.client.get("/dashboard")	HttpUser 默认使用 Session，自动管理 Cookie。	登录后 Cookie 自动保存，后续请求自动携带。	无需手动处理 Cookie，适合有状态应用测试。
查看 Cookie	print(self.client.cookies)	查看当前会话的 Cookie 集合。	for cookie in self.client.cookies: print(cookie.name, cookie.value)	用于调试或提取特定 Cookie 值。
手动设置 Cookie	self.client.cookies.set("session_id", "abc123")	强制设置 Cookie 值。	self.client.cookies.set("theme", "dark")	少用，通常由服务器自动设置。
清除 Cookie	self.client.cookies.clear() self.client.get("/login")	清除当前会话的所有 Cookie。	self.client.cookies.clear()	用于模拟新用户或重新登录。

4.6 提取响应数据（JSON/Text）用于后续请求

方法	语法	用途	代码示例	注意事项
提取 JSON 字段	data = r.json() user_id = data["id"]	从响应中提取数据用于后续请求参数。	r = self.client.post("/users", json={"name": "Test"}) user_id = r.json()["id"] self.client.get(f"/users/{user_id}")	确保响应结构稳定，避免 KeyError。
提取文本中的数据	import re match = re.search(r"id=(\d+)", r.text)	使用正则从 HTML 或文本中提取信息。	r = self.client.get("/form") token = re.search('csrf_token" value="(.+?)"', r.text).group(1)	适用于传统 Web 应用。
存储为实例变量	self.user_id = data["id"]	将提取的数据保存为用户实例变量，供其他任务使用。	def on_start(self): r = self.client.post("/login") self.token = r.json()["token"] @task def access_api(self): self.client.get("/data", headers={"Authorization": f"Bearer {self.token}"})	实现任务间数据传递。
错误处理	try: data = r.json() except Exception as e: ...	防止解析失败导致脚本中断。	if r.status_code == 200: try: user_id = r.json()["id"] except (KeyError, ValueError): self.logger.warning("Failed to parse user ID")	增强脚本健壮性。

第5章：测试场景设计

5.1 多用户行为建模（多个 User 子类）

方法/属性	语法	用途	代码示例	注意事项
定义多个 User 类	class BrowserUser(HttpUser): ... class ApiUser(HttpUser): ...	模拟不同类型的用户行为（如普通用户、管理员、API 客户端）。	class MobileUser(HttpUser): wait_time = between(2, 5) @task def view_feed(self): ... class AdminUser(HttpUser): wait_time = constant(1) @task def audit_logs(self): ...	每个类代表一种用户角色或终端类型。
启动时指定用户类	locust -f locustfile.py --headless -u 100 -r 10	Locust 自动发现所有 User 子类并按比例启动。	若文件中有 BrowserUser 和 ApiUser，则两者都会参与压测。	所有 User 子类在同一文件或模块中均会被加载。
使用 fixed_count 控制数量	fixed_count = 5	固定该用户类的实例数量，用于关键角色控制。	class AdminUser(HttpUser): fixed_count = 3 host = "https://admin.example.com"	适合模拟少量高权限用户。
host 属性差异化	host = "https://api.example.com"	不同用户类访问不同服务或环境。	class ExternalApiUser(HttpUser): host = "https://thirdparty-api.com" @task def call_service(self): ...	实现跨系统联合压测。

5.2 权重控制（weight 参数）

方法	语法	用途	代码示例	注意事项
User 类权重	class RegularUser(HttpUser): weight = 10 class PremiumUser(HttpUser): weight = 2	控制不同用户类在总用户数中的占比。	weight = 8 表示该类用户数量是 weight=1 类的 8 倍。	总用户数按权重比例分配，如 total=100，Regular:Premium ≈ 80:20。
@task 权重	@task(3) def browse(self): ... @task(1) def checkout(self): ...	控制同一用户内不同任务的执行频率。	浏览任务执行概率是结账任务的 3 倍。	权重为相对值，总和不强制为 100。
组合使用	User.weight + @task.weight	实现多层级流量分布控制。	普通用户多浏览，VIP 用户多下单。	设计时应基于真实业务流量比例。
权重为 0	@task(0) 或 weight=0	禁用某个任务或用户类（调试用）。	@task(0) def deprecated_task(self): ...	不推荐用于生产脚本，可用注释替代。

5.3 任务执行顺序与随机性

方法	语法	用途	代码示例	注意事项
随机调度	默认行为	Locust 随机选择 @task 标记的方法执行，符合权重比例。	多个 @task 任务自动随机执行。	无法保证顺序，适合模拟真实用户行为。
强制顺序执行	在单个 @task 中调用其他方法	实现固定流程的任务序列。	@task def complete_purchase(self): self.add_to_cart() self.checkout() self.pay()	适用于必须按序执行的业务流。
使用循环控制	for i in range(3): self.browse_category() self.wait()	在任务中实现重复操作。	模拟用户连续浏览多个类别。	注意 wait_time 已存在时避免双重等待。
禁用随机性	只定义一个 @task	所有用户只执行单一任务，无随机选择。	@task def ping_heartbeat(self): self.client.get("/ping")	适用于接口级性能基准测试。

5.4 模拟不同用户路径（业务流）

方法	语法	用途	代码示例	注意事项
on_start() 初始化路径	def on_start(self): self.user_type = random.choice(["new", "returning"])	在用户启动时决定其行为路径。	if self.user_type == "new": self.signup() else: self.login()	实现用户分群测试。
路径分支任务	@task(2) def user_flow(self): if hasattr(self, 'is_vip') and self.is_vip: self.access_vip_content() else: self.view_ads()	根据用户状态执行不同逻辑。	结合实例变量实现个性化路径。	—
模拟注册→登录→购物流	def on_start(self): self.register() @task def shop(self): self.browse() self.buy()	完整用户生命周期建模。	注意错误处理，注册失败时不应继续后续步骤。	—
模拟游客 vs 登录用户	class GuestUser(HttpUser): ... class AuthUser(HttpUser): ...	用不同类区分访问模式。	GuestUser 只能浏览，AuthUser 可下单。	更清晰，优于在单个类中做条件判断。

5.5 使用 between()、constant() 等 wait_time 策略

方法	语法	用途	代码示例	注意事项
between(min, max)	wait_time = between(1, 5)	用户在 min 到 max 秒间随机等待，模拟自然行为。	wait_time = between(2, 10)	最常用，推荐用于大多数场景。
constant(seconds)	wait_time = constant(1)	固定间隔等待，产生稳定请求流。	wait_time = constant(0.5) # 每秒2次请求	适合吞吐量敏感测试。
constant_pacing(seconds)	wait_time = constant_pacing(5)	确保每轮任务至少耗时指定秒数，若任务快则等待补足。	wait_time = constant_pacing(3)	防止过快请求淹没服务器。
constant_throughput(rps)	wait_time = constant_throughput(2)	控制用户每秒最多发起 rps 次请求。	wait_time = constant_throughput(0.2) # 每5秒1次	实现恒定吞吐量模型。
自定义函数	def my_wait(self): return random.uniform(1, 3) wait_time = my_wait	实现复杂等待逻辑（如指数分布、高峰模拟）。	def peak_hour_wait(self): return random.expovariate(1/2) # 模拟高峰到达率	函数必须返回数值（秒）。

第6章：参数化与数据驱动

6.1 使用外部数据文件（CSV）进行参数化

方法	语法	用途	代码示例	注意事项
读取 CSV 文件	import csv with open("users.csv") as f: reader = csv.DictReader(f) data = list(reader)	加载外部测试数据，实现数据驱动测试。	users = [] with open("data/users.csv") as f: for row in csv.DictReader(f): users.append((row["username"], row["password"]))	建议在 on_init 或 init 中预加载。
每用户独享数据	使用 next() 从迭代器取数据	避免多个用户争用同一数据行。	def on_start(self): self.username, self.password = next(self.user_iterator)	需确保数据量 >= 用户数，否则抛 StopIteration。
共享数据池	类变量存储数据列表	多用户共享同一数据集，可重复使用。	class MyUser(HttpUser): credentials = [...] # 预加载 def on_start(self): cred = random.choice(self.credentials)	适合用户名密码等可复用凭证。
动态切换数据	在任务中切换账号或输入	模拟多账户操作。	@task def post_content(self): post_as_user(self.current_user)	结合状态管理实现复杂行为。

6.2 数据共享与线程安全问题

概念	说明	注意事项
Locust 并发模型	基于 gevent 协程，非多线程，但存在并发执行。	协程间仍可能竞争资源，需注意共享数据安全。
实例变量（self.xxx）	每个 User 实例独享，天然线程安全。	推荐用于存储用户私有数据（如 session_id、token）。
类变量（cls.xxx）	所有实例共享，存在并发修改风险。	如需修改，应使用锁或原子操作。
全局变量	所有用户共享，高风险区域。	避免写操作，或使用 threading.Lock（gevent 兼容）。
读写共享数据	读安全，写不安全	可读取类变量配置，但避免多用户同时 append 到同一列表。
推荐做法	尽量使用实例变量，预加载只读数据	数据初始化放在类级别，运行时状态放实例级别。

6.3 使用类变量或全局变量传递数据

方法	语法	用途	代码示例	注意事项
类变量存储配置	CREDENTIALS = [("u1","p1"), ("u2","p2")]	定义静态测试数据供所有实例使用。	class LoginUser(HttpUser): USERS = load_from_csv("users.csv")	初始化后不应修改。
全局变量跨文件共享	USERS = [] # 在 locustfile.py 中定义	在多个 User 类或辅助函数间共享数据。	global USER_QUEUE USER_QUEUE = iter(load_users())	使用 global 关键字声明。
预加载大数据集	@classmethod def setup_class(cls): cls.data = load_large_file()	在测试开始前一次性加载，提升性能。	class DataUser(HttpUser): products = load_json("products.json")	避免在 on_start 中重复加载。
共享计数器	total_success = 0	统计全局指标（如成功交易数）。	with counter_lock: MyUser.total_orders += 1	必须加锁保护写操作。

6.4 动态生成测试数据（Faker 等库）

方法	语法	用途	代码示例	注意事项
安装 Faker	pip install faker	生成逼真的虚拟数据（姓名、邮箱、地址等）。	from faker import Faker fake = Faker()	支持多国语言和数据类型。
生成用户信息	fake.name(), fake.email()	用于注册、资料填写等场景。	name = fake.name() email = fake.email() self.client.post("/register", json={"name": name, "email": email})	每次调用生成唯一数据。
生成地理位置	fake.city(), fake.country()	模拟全球用户访问。	location = f"{fake.city()}, {fake.country()}"	增强测试真实性。
生成时间数据	fake.date_this_year(), fake.time()	用于日期字段测试。	dob = fake.date_of_birth(minimum_age=18)	支持范围控制。
多语言支持	Faker('zh_CN'), Faker('fr_FR')	生成特定语言的数据。	cn_fake = Faker('zh_CN') name = cn_fake.name() # 中文名	适合国际化应用测试。
性能考虑	预生成 or 按需生成	高并发下频繁调用 Faker 可能成为瓶颈。	建议预生成一批数据循环使用，而非每次实时生成。	—

第7章：高级功能与扩展

7.1 自定义客户端（非 HTTP 协议支持）

方法	语法	用途	代码示例	注意事项
继承 User 类	class CustomUser(User): def init(self, *args, **kwargs): super().init(*args, **kwargs)	创建自定义客户端类，实现非 HTTP 协议。	class MQTTUser(User): def init(self, *args, **kwargs): super().init(*args, **kwargs) self.client = MqttClient()	需实现自己的客户端逻辑。
定义自定义客户端	from some_library import Client class MyCustomClient(Client): ...	实现具体的协议通信逻辑。	class RedisClient(Client): def connect(self): self.conn = redis.Redis(host="localhost") def get_data(self, key): return self.conn.get(key)	确保连接池等资源管理得当。
使用自定义客户端	@task def fetch_data(self): data = self.client.get_data("key")	在任务中使用自定义客户端执行操作。	@task def send_message(self): self.client.publish("topic", "payload")	注意异常处理，避免脚本中断。

7.2 WebSocket 压测示例

方法	语法	用途	代码示例	注意事项
使用 websocket-client 库	pip install websocket-client	发起 WebSocket 连接并进行压测。	from websocket import create_connection ws = create_connection("ws://example.com/ws")	注意 WebSocket 连接状态维护。
WebSocketUser 示例	class WebSocketUser(HttpUser): wait_time = between(1, 5) def on_start(self): self.ws = create_connection("ws://example.com/ws") @task def send_message(self): self.ws.send("Hello, World!") response = self.ws.recv() print(response)	模拟 WebSocket 客户端行为。	注意在 on_stop 中关闭连接，避免资源泄漏。
断言与响应处理	if "success" in response: self.environment.events.request_success.fire(...) else: self.environment.events.request_failure.fire(...)	处理 WebSocket 响应并记录成功或失败。	通过事件机制通知 Locust 记录结果。	—

7.3 断言与失败处理（catch_response, ResponseContextManager）

方法	语法	用途	代码示例	注意事项
catch_response 参数	with self.client.get("/", catch_response=True) as response: if response.status_code == 200 and "OK" in response.text: response.success() else: response.failure("Not OK")	手动控制请求的成功/失败判断。	with self.client.post("/login", json={"u":"admin","p":"123"}, catch_response=True) as r: if r.status_code != 200 or "token" not in r.json(): r.failure("Login failed")	必须配合 with 语句和 response.success() 或 failure() 使用。
使用 ResponseContextManager	from locust import ResponseContextManager	提供更灵活的响应处理方式。	with ResponseContextManager(self.client.get("/")) as response: if "expected_text" in response.text: response.success() else: response.failure("Expected text not found")	适合复杂的断言逻辑。
事件监听器	self.environment.events.request_success.add_listener(my_success_handler) self.environment.events.request_failure.add_listener(my_failure_handler)	监听成功和失败事件，触发自定义处理逻辑。	def my_success_handler(request_type, name, response_time, response_length, **kw): print(f"Success: {name} took {response_time}ms")	可用于日志记录或报警。

7.4 自定义事件（events）与钩子函数

方法	语法	用途	代码示例	注意事项
request_success 事件	self.environment.events.request_success.add_listener(on_request_success)	监听请求成功事件，执行自定义逻辑。	def on_request_success(request_type, name, response_time, response_length): print(f"Request succeeded: {name}")	适用于统计或日志记录。
request_failure 事件	self.environment.events.request_failure.add_listener(on_request_failure)	监听请求失败事件，执行自定义逻辑。	def on_request_failure(request_type, name, response_time, exception): print(f"Request failed: {name}, Exception: {exception}")	有助于快速定位问题。
init 事件	self.environment.events.init.add_listener(on_init)	在 Locust 启动时执行初始化逻辑。	def on_init(environment, **kw): print("Locust initialized")	适合全局配置或资源预加载。
quitting 事件	self.environment.events.quitting.add_listener(on_quitting)	在 Locust 结束前执行清理逻辑。	def on_quitting(environment, **kw): print("Locust is quitting")	关闭数据库连接或其他资源。

7.5 分布式压测：Master 与 Worker 配置

方法	语法	用途	代码示例	注意事项
启动 Master	locust -f locustfile.py --master	启动 Master 节点，协调多个 Worker。	启动命令：locust -f locustfile.py --master	默认端口 5557（客户端）和 5558（Worker）。
启动 Worker	locust -f locustfile.py --worker --master-host=<MASTER_IP>	启动 Worker 节点，执行实际压测任务。	启动命令：locust -f locustfile.py --worker --master-host=192.168.1.100	确保网络通畅，防火墙开放相应端口。
配置文件	在 locustfile.py 中设置共享变量或配置	实现多节点间数据同步或共享配置。	class SharedConfig: base_url = "https://api.example.com" class MyUser(HttpUser): host = SharedConfig.base_url	避免直接硬编码 URL 或其他重复信息。
日志与监控	使用 logging 模块记录日志	收集分布式测试中的运行日志。	import logging logging.info("Starting worker node")	便于调试和故障排查。

第8章：性能监控与结果分析

8.1 Web UI 界面详解（图表、RPS、响应时间等）

指标名称	说明	注意事项
RPS 图表	每秒请求数，反映系统吞吐能力。	观察峰值和波动情况，评估系统极限。
响应时间分布	包括平均值、中位数、90%/95%/99% 分位值。	特别关注高分位值，代表极端情况下的用户体验。
用户数图表	当前活跃用户数，随时间变化趋势。	结合 RPS 和响应时间，分析负载对性能的影响。
错误率	请求失败比例，如 HTTP 5xx、超时等。	高错误率提示系统存在瓶颈或逻辑缺陷。
CPU/Memory 监控	若集成监控工具（如 Prometheus），可显示服务器资源使用情况。	了解系统资源消耗，辅助容量规划。
实时日志	显示最近的请求日志，包括成功和失败。	快速定位具体请求的问题。

8.2 实时监控指标解读

指标名称	说明	注意事项
平均响应时间	所有请求的平均响应时长。	易受极端值影响，需结合分位数分析。
最大/最小响应时间	单个请求的最大和最小响应时间。	Max 值过高可能表示存在性能瓶颈。
总请求数	测试期间累计发起的请求数量。	与预期数量对比，确保测试覆盖完整。
成功请求数	成功完成的请求数量。	结合总请求数计算成功率。
失败请求数	失败的请求数量及类型（如 500、Timeout）。	点击可查看错误详情，便于定位问题。
用户数	当前并发用户数量。	结合 RPS 和响应时间，分析负载对性能的影响。

8.3 生成 HTML 报告

方法	语法	用途	代码示例	注意事项
使用 —html 参数	locust -f locustfile.py --headless -u 100 -r 10 --run-time 10m --html report.html	运行测试后生成 HTML 格式的报告。	locust -f locustfile.py --headless -u 50 -r 5 --run-time 5m --html output/report.html	报告包含图表和关键性能指标。
查看报告	打开生成的 HTML 文件，查看测试结果。	无需额外工具，浏览器即可查看。	适合分享给团队成员或客户。
自定义报告模板	可通过修改源码实现定制化报告。	适合高级用户根据需求调整报告样式。	修改源码需谨慎，建议备份原版。

8.4 命令行模式下的结果输出（—csv 参数）

方法	语法	用途	代码示例	注意事项
使用 —csv 参数	locust -f locustfile.py --headless -u 100 -r 10 --run-time 10m --csv results	将测试结果导出为 CSV 文件。	locust -f locustfile.py --headless -u 50 -r 5 --run-time 5m --csv output/results	导出文件名可自定义。
输出文件	自动生成三个 CSV 文件：stats.csv、failures.csv、requests.csv	分别存储统计信息、失败请求、所有请求详情。	stats.csv 包含整体性能指标；failures.csv 列出失败请求；requests.csv 详尽记录每次请求。	—
数据分析	使用 Excel 或 Pandas 分析导出的数据。	适合后续深入分析或绘制自定义图表。	—	—

8.5 结果文件字段说明（stats, failures, requests 等）

文件	字段	说明	注意事项
stats.csv	Type	请求类型（GET、POST 等）。	用于区分不同接口的性能表现。
	Name	请求路径或自定义名称。	可在请求中使用 name="login" 参数统一命名动态 URL。
	Request Count	总请求数。	结合 RPS 可分析系统处理能力。
	Failures	失败请求数及错误率。	错误率 > 0% 需排查原因（网络、服务、断言等）。
	Median (ms)	响应时间中位数。	比平均值更能反映典型用户感受。
	90% / 95% / 99% Percentile	90%/95%/99% 的请求响应时间低于该值。	用于评估极端情况下的用户体验。
failures.csv	Method	请求方法（GET、POST 等）。	用于区分不同类型的失败请求。
	Name	请求路径或自定义名称。	与 stats.csv 对应。
	Error	具体错误信息（如 500 内部服务器错误）。	便于定位具体问题。
requests.csv	Timestamp	请求发生的时间戳。	用于时间序列分析。
	Method	请求方法（GET、POST 等）。	用于区分不同类型的请求。
	Path	请求路径。	与 Name 字段对应。

第9章：最佳实践与常见问题

9.1 编写可维护的 Locust 脚本

方法	语法	用途	代码示例	注意事项
模块化组织	将不同业务拆分为多个文件或类	提高脚本可读性和复用性。	users/ ├── browser_user.py ├── api_user.py └── common_tasks.py	使用 from .common_tasks import login_task 导入共享任务。
使用常量配置	HOST = "https://api.example.com" HEADERS = {"Content-Type": "application/json"}	集中管理配置，便于环境切换。	class ApiUser(HttpUser): host = config.HOST default_headers = config.HEADERS	避免硬编码，推荐使用配置文件（如 JSON、YAML）。
任务职责单一	每个 @task 只做一件事	便于调试、权重控制和结果分析。	@task def view_product(self): ... @task def add_to_cart(self): ...	避免在一个任务中混合多个操作。
添加注释与文档	# 模拟用户登录流程 def on_start(self): ...	帮助团队成员理解脚本逻辑。	使用 docstring 描述类或复杂方法用途。	特别是 on_start/on_stop 和复杂路径逻辑。
使用日志	import logging logging.info("User logged in")	记录关键行为，便于调试和监控。	self.logger.info(f"User {self.username} started")	避免过度日志影响性能。

9.2 避免常见性能测试陷阱

陷阱	说明	解决方案	注意事项
脚本自身成为瓶颈	Locust 压测机 CPU/内存/网络耗尽，无法生成足够压力。	使用分布式压测（Master/Worker）、优化脚本、升级压测机。	监控压测机资源使用率，确保其不成为瓶颈。
忘记设置 wait_time	用户无等待连续请求，产生非真实流量。	显式设置 wait_time = between(1, 5) 或类似策略。	即使是接口级测试，也建议设置最小等待。
错误使用 catch_response	忘记调用 success()/failure()，导致请求不被记录。	确保在 with 块中根据条件调用对应方法。	建议使用 try-except 包裹，防止异常中断。
动态 URL 未命名	不同 ID 的请求被视为不同接口，统计分散。	使用 name="/users/[id]" 统一命名。	self.client.get(f"/users/{user_id}", name="/users/[id]")
并发修改共享数据	多用户同时修改类变量导致数据竞争。	使用只读数据、实例变量或加锁保护。	尽量避免在运行时修改共享状态。

9.3 资源限制与压测机优化

方法	说明	代码/命令示例	注意事项
监控压测机资源	使用 top、htop、nmon 等工具监控 CPU、内存、网络。	htop nethogs	确保压测机资源充足，避免成为瓶颈。
调整 gevent 协程数	默认协程数可能受限，可调整以提升并发能力。	通常无需手动设置，Locust 自动管理。	若需调整，可通过环境变量或源码修改。
使用轻量级等待策略	避免 constant(0) 或过短 wait_time 导致 CPU 空转。	wait_time = between(0.1, 1) 比 constant(0) 更合理。	平衡压力与资源消耗。
分布式部署	多台 Worker 分担负载，突破单机限制。	Master: --master Worker: --worker --master-host=x.x.x.x	确保网络延迟低，Master 不参与压测。
关闭不必要的服务	压测机上关闭无关进程，释放资源。	systemctl stop docker（若不用）	保证 Locust 获得最大系统资源。

9.4 如何模拟真实用户行为

方法	说明	代码示例	注意事项
设置合理的 wait_time	模拟用户”思考时间”或页面停留时间。	wait_time = between(2, 10) # 浏览页面	不同页面停留时间不同，可差异化设置。
混合用户类型	定义不同 User 类模拟游客、会员、管理员等。	class GuestUser(HttpUser): ... class MemberUser(HttpUser): ...	按真实比例设置 weight。
模拟用户路径（User Journey）	实现注册 → 登录 → 浏览 → 下单 → 支付完整流程。	on_start() 中登录，任务中按序执行。	使用 @task 权重控制各环节频率。
使用真实数据或 Faker	避免使用固定参数，增加请求多样性。	name=fake.name(), email=fake.email()	提升测试真实性，避免缓存干扰。
模拟不同终端行为	移动端请求频率低、等待长；API 客户端频率高。	MobileUser 的 wait_time 更长，ApiClient 更短。	结合业务场景设计。
引入随机失败或中断	模拟用户放弃操作、网络中断等异常。	if random.random() < 0.1: return # 10% 概率跳出	更贴近真实用户行为。

9.5 故障排查：常见错误与解决方案

错误现象	可能原因	解决方案	注意事项
ConnectionError: [Errno 111] Connection refused	目标服务未启动或端口错误	检查 host 配置、服务状态、防火墙	使用 ping 或 telnet 测试连通性
SSLError: [SSL: CERTIFICATE_VERIFY_FAILED]	HTTPS 证书验证失败	添加 verify=False 或设置正确 CA 证书	仅测试环境使用 verify=False，生产慎用
KeyError: ‘token’	响应 JSON 中缺少预期字段	检查接口返回结构，添加异常处理	使用 try-except 或 dict.get() 安全访问
响应时间异常高（>10s）	网络延迟、服务过载、脚本阻塞	检查压测机与目标网络、服务负载、脚本逻辑	排除压测机自身性能瓶颈
RPS 上不去（远低于预期）	wait_time 过长、压测机资源不足、连接池限制	缩短 wait_time、优化脚本、使用分布式	查看 Locust 日志是否有警告
分布式模式下 Worker 无法连接 Master	网络不通、防火墙、IP/端口配置错误	检查 --master-host、开放 5557/5558 端口	使用 telnet master_ip 5557 测试
CSV 数据文件读取失败	文件路径错误、编码问题、权限不足	使用绝对路径、检查文件编码（UTF-8）、权限	建议将数据文件与脚本放同一目录
ResponseTimeout 错误过多	服务响应慢或网络延迟高	增加 timeout 参数，如 self.client.get("/", timeout=30)	默认超时时间可能过短

第10章：集成与自动化

10.1 与 CI/CD 集成（如 Jenkins、GitHub Actions）

方法	语法/配置	用途	代码示例（CI 配置片段）	注意事项
命令行执行 Locust	locust -f locustfile.py --headless --users 10 --spawn-rate 2 --run-time 2m	在 CI 环境中非交互式运行压测。	yaml # GitHub Actions 示例 - name: Run Locust Test run: locust -f test_api.py --headless -u 5 -r 1 --run-time 1m --exit-code-on-error=1	必须使用 --headless 模式。
Jenkins Pipeline 集成	使用 sh 步骤调用 Locust 命令	将性能测试嵌入构建流程。	groovy stage('Performance Test') { steps { sh 'locust -f locustfile.py --headless -u 10 -r 1 --run-time 5m' } }	确保 Jenkins 节点已安装 Python 和 Locust。
GitHub Actions 集成	在 .github/workflows/ 中定义 workflow	实现 Pull Request 触发性能回归测试。	yaml on: [push, pull_request] jobs: load-test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Set up Python uses: actions/setup-python@v4 with: python-version: '3.11' - run: pip install locust - run: locust -f test_user.py --headless -u 5 -r 1 --run-time 2m --exit-code-on-error=1	可结合 --exit-code-on-error 控制流程中断。
环境变量注入	LOCUST_HOST=https://staging.example.com	动态指定测试环境，避免硬编码。	yaml env: LOCUST_HOST: ${{ secrets.STAGING_URL }}	在 Locust 脚本中使用 os.getenv("LOCUST_HOST") 读取。

10.2 自动化执行与阈值校验（exit code）

方法	语法	用途	代码示例	注意事项
—exit-code-on-error	--exit-code-on-error=1	当存在失败请求时，Locust 进程返回非零退出码，用于 CI 中断。	locust -f test.py --headless -u 10 --run-time 5m --exit-code-on-error=1	推荐在 CI 中使用，确保失败时构建失败。
自定义阈值检查	结合脚本逻辑判断指标是否达标	实现更精细的断言，如”99%响应时间 < 1s”。	python def check_thresholds(environment): stats = environment.stats.total if stats.get_response_time_percentile(0.99) > 1000: print("99th percentile too high!") environment.process_exit_code = 1 # 在 on_stop 或事件中调用	需通过事件钩子注入，如 quitting。
结合 —csv 进行后处理	导出 CSV 后用脚本分析结果	实现复杂阈值校验（如错误率 < 0.1%）。	bash locust -f test.py --headless -u 10 --run-time 2m --csv=results python analyze.py results_stats.csv # 自定义分析脚本	适合需要统计多个指标组合判断的场景。
environment.process_exit_code	设置该属性为非零值	强制 Locust 返回指定退出码。	environment.process_exit_code = 1 # 表示失败	通常在 quitting 事件中设置。

10.3 结合 Grafana + InfluxDB 实时监控

方法	说明	配置示例	注意事项
使用 locust-influxdb-listener	将 Locust 指标实时发送到 InfluxDB。	python from locust_influxdb_listener import InfluxDBListener class MyUser(HttpUser): def on_start(self): InfluxDBListener(self.environment, influxdb_host="localhost", influxdb_port=8086, database="locust")	需安装 pip install locust-influxdb-listener。
手动发送指标到 InfluxDB	使用 influxdb-client 库自定义发送。	python from influxdb_client import Point from influxdb_client.client.write_api import SYNCHRONOUS def on_request_success(...): point = Point("request").tag("name", name).field("response_time", response_time) write_api.write(bucket="locust", record=point)	灵活性高，但需自行管理连接和错误。
Grafana 面板配置	添加 InfluxDB 数据源，创建可视化面板。	数据源 URL: http://localhost:8086 Query: from(bucket: "locust") |> range(start: -5m)	—
监控指标	包括 RPS、响应时间、用户数、错误率等。	实时反映压测过程中的系统表现。	可叠加服务器资源指标（CPU、内存）进行关联分析。
性能开销	持续写入 InfluxDB 可能影响压测机性能。	建议使用异步写入、批量提交。	避免在高并发下频繁写入单条数据。

10.4 使用 locust.conf 配置文件

方法	语法	用途	配置文件示例	注意事项
创建 locust.conf 文件	在项目根目录或用户目录创建配置文件。	集中管理 Locust 运行参数，避免命令行冗长。	ini [locust] host = https://api.example.com users = 50 spawn-rate = 5 run-time = 10m headless = true csv = results/output exit-code-on-error = true	支持 .conf 或 .ini 后缀。
配置优先级	命令行参数 > locust.conf > 默认值	命令行可覆盖配置文件中的设置。	locust -f test.py --users 100 会覆盖配置文件中的 users = 50。	便于环境差异化配置。
支持的配置项	大多数命令行参数均可写入配置文件。	包括 host, users, spawn-rate, run-time, headless, csv, master, worker 等。	完整列表参考 Locust 官方文档。	—
环境隔离	为不同环境（dev/staging/prod）使用不同配置文件。	locust -c locust.staging.conf	可通过 -c 指定配置文件路径。	—
敏感信息处理	避免在配置文件中明文存储密码或密钥。	使用环境变量注入敏感信息。	如 host = ${STAGING_HOST}，需结合其他工具支持。	—