当前位置：首页 > news >正文

Locust性能测试进阶：自定义用户行为与实时Web UI监控实战

news 2026/7/27 6:17:45

1. 项目概述：为什么选择 Locust 做性能测试？

如果你正在寻找一个轻量级、代码驱动、并且能提供实时监控的性能测试工具，那么 Locust 绝对值得你花时间研究。我最初接触性能测试时，也用过一些图形化工具，它们上手快，但一旦遇到复杂的业务逻辑或者需要高度定制化的压测场景，就感觉有点力不从心。脚本维护困难、资源消耗大、报告不够直观等问题接踵而至。后来转向 Locust，它用 Python 代码定义用户行为，这种“一切皆代码”的理念，让性能测试脚本变得和开发业务代码一样清晰、可维护，并且能轻松集成到 CI/CD 流程中。

Locust 的核心魅力在于它的简洁和强大。它不像一些重型工具那样需要复杂的配置，一个简单的 Python 文件就能启动一场分布式压测。更重要的是，它内置了一个基于 Web 的实时监控界面，你可以在压测过程中，实时观察每秒请求数（RPS）、响应时间、失败率等关键指标的变化曲线，这种即时反馈对于快速定位性能瓶颈至关重要。本次我们要深入探讨的，正是 Locust 的两个核心进阶能力：如何灵活地自定义模拟用户（User）的复杂行为，以及如何充分利用其实时 Web UI 来掌控整个压测过程。无论你是想模拟用户登录后浏览商品、加购、下单的电商流程，还是想测试一个带有思考时间、随机跳转的 API 接口链，Locust 都能通过清晰的代码结构帮你实现。

2. 环境准备与 Locust 核心概念解析

在开始编写复杂的用户行为之前，我们需要先把基础环境搭建好，并理解 Locust 里几个关键名词的含义。这能帮助我们在后续设计测试场景时，思路更加清晰。

2.1 安装与基础配置

Locust 的安装极其简单，它只依赖 Python 环境。我强烈建议使用 Python 3.8 或更高版本，并且为每个项目创建独立的虚拟环境，以避免包依赖冲突。

# 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，或者存在多个 Python 环境，一定要确认当前激活的虚拟环境是正确的，否则可能会遇到ModuleNotFoundError或者命令不识别的问题。

2.2 理解 User、TaskSet 与 Tasks

这是 Locust 脚本的基石，理解它们的关系至关重要。

User 类：这是你要模拟的“虚拟用户”的蓝图。每个虚拟用户在测试运行时，都会成为一个独立的 User 类实例。在 User 类中，你需要定义两个关键属性：
- wait_time：用户在执行任务之间的等待时间。这用于模拟真实用户的思考或浏览间隔。常用的是between(min_wait, max_wait)，表示在最小和最大秒数之间随机等待。
- tasks：这是一个列表，定义了用户要执行哪些任务（TaskSet 类或普通函数），以及它们的执行权重。权重越高，被选中的概率越大。
TaskSet 类：你可以把它理解为一组任务的集合或一个“场景”。比如，“浏览商品”是一个 TaskSet，里面可能包含“查看商品列表”、“搜索商品”、“查看商品详情”等子任务。TaskSet 可以嵌套，从而构建出非常复杂的用户行为路径。
Task：任务是最小的执行单元，就是一个 Python 函数（或用@task装饰器装饰的方法）。它里面包含了具体的 HTTP 请求（或其他客户端操作）和断言。Locust 的HttpUser类提供了便捷的client属性（一个requests.Session的封装）来发起请求。

一个最基础的脚本骨架长这样：

from locust import HttpUser, task, between class QuickstartUser(HttpUser): wait_time = between(1, 2.5) # 每次任务后等待1-2.5秒 @task(3) # 权重为3 def view_items(self): self.client.get("/items") @task(1) # 权重为1 def view_item(self): for item_id in range(10): self.client.get(f"/item?id={item_id}", name="/item") # name参数用于聚合统计 def on_start(self): self.client.post("/login", json={"username":"foo", "password":"bar"})

在这个例子里，QuickstartUser继承自HttpUser。每个虚拟用户启动后，会先执行on_start方法（模拟登录），然后根据权重反复执行view_items和view_item任务。

注意：很多新手会忽略name参数。在上面的view_item任务中，我们循环请求了10个不同的URL，但通过name="/item"将它们统一归到 “/item” 这个统计条目下。否则，Locust 的报告中会出现10个独立的条目，使得数据非常碎片化，不利于分析。这是编写高效 Locust 脚本的第一个重要技巧。

3. 深度自定义 User 行为模式

掌握了基础之后，我们来挑战更真实的业务场景。真实的用户行为很少是简单、固定权重的随机选择，他们可能按顺序操作，可能根据上一个请求的结果决定下一步，也可能在失败后重试。Locust 通过其灵活的 Python 特性，可以轻松模拟这些复杂行为。

3.1 使用 TaskSet 构建顺序与分支逻辑

单纯的@task装饰器是随机选择，要模拟固定流程，我们需要在 TaskSet 内部定义执行顺序。

from locust import HttpUser, TaskSet, task, between class UserBehavior(TaskSet): # 任务也可以按顺序执行，只需在方法中调用self.interrupt()来跳出 @task(1) def sequence_flow(self): # 1. 浏览首页 self.client.get("/") # 2. 登录 resp_login = self.client.post("/api/login", json={"user":"test", "pwd":"123"}) if resp_login.status_code == 200: token = resp_login.json().get("token") self.headers = {"Authorization": f"Bearer {token}"} # 3. 获取用户信息（依赖登录态） self.client.get("/api/user/profile", headers=self.headers) # 4. 执行完这个序列后，这个用户实例会回到User类中，重新选择任务 self.interrupt() # 关键：中断当前TaskSet，返回父级 class WebsiteUser(HttpUser): wait_time = between(2, 5) tasks = [UserBehavior] # 将TaskSet类作为任务 def on_start(self): """每个用户开始时的初始化，比TaskSet的on_start更早""" self.headers = {}

这里，UserBehavior这个 TaskSet 定义了一个顺序流。self.interrupt()是关键，它告诉 Locust 这个序列执行完毕，虚拟用户应该跳出当前的 TaskSet，回到 User 的 tasks 列表中重新选择任务（虽然这里只有一个任务，但实践中可能有多个）。如果不调用interrupt()，用户会一直在这个 TaskSet 内循环执行其定义的任务。

3.2 模拟用户思考时间与步调

wait_time属性控制任务间的间隔，但有时我们需要在单个任务内部的多个步骤间也加入等待，以更精确地模拟用户操作。

from locust import HttpUser, task, between import time class PacingUser(HttpUser): # 全局的任务间等待时间 wait_time = between(1, 3) @task def complex_journey(self): # 步骤1：搜索商品 self.client.get("/search?q=laptop") # 模拟用户查看搜索结果列表的思考时间 time.sleep(2) # 使用标准的time.sleep # 步骤2：点击第一个商品 self.client.get("/product/123") # 模拟用户阅读商品详情页的时间 time.sleep(5) # 步骤3：加入购物车 self.client.post("/cart/add", json={"product_id": 123, "quantity": 1}) # 模拟用户犹豫时间 time.sleep(1) # 步骤4：去结算（可能只有部分用户会做） if random.random() < 0.7: # 70%的用户会点击结算 self.client.get("/checkout")

直接在任务中使用time.sleep()是完全可以的。这会让这个虚拟用户的执行线程暂停，从而精确控制每个步骤的间隔。需要注意的是，这会占用一个并发用户数。如果你设置了1000个用户，有大量用户同时sleep，实际上对服务器的压力会间歇性下降。这种模式适合模拟“节奏型”场景，而非极限并发场景。

3.3 参数化与数据驱动测试

压测时使用固定的数据（如固定的用户ID、商品ID）会导致缓存命中率异常高，测试结果失真。我们必须让数据动态化。

方法一：使用列表循环（适用于小规模数据）

class DataDrivenUser(HttpUser): wait_time = between(1, 2) # 假设我们有一批测试账号 user_credentials = [ {"username": "user1", "password": "pass1"}, {"username": "user2", "password": "pass2"}, # ... 更多数据 ] def on_start(self): # 每个用户实例从列表中取一组凭证 cred = random.choice(self.user_credentials) resp = self.client.post("/login", json=cred) if resp.ok: self.token = resp.json()["token"] else: # 登录失败，可以标记此用户停止运行或重试 self.stop(force=True) @task def get_profile(self): self.client.get("/profile", headers={"Auth": self.token})

这种方式简单，但数据是预加载到内存的，且所有用户共享。如果数据量很大，或者希望每个用户数据唯一，就不太合适。

方法二：从外部文件动态读取（推荐）更常见的做法是从 CSV、JSON 文件或数据库中读取数据。Locust 本身不提供内置的数据池管理，但我们可以利用 Python 轻松实现。

import csv import queue from locust import HttpUser, task, between class CSVDataUser(HttpUser): wait_time = between(1, 3) # 在类级别初始化一个线程安全的数据队列 data_queue = queue.Queue() with open('test_accounts.csv', 'r') as f: reader = csv.DictReader(f) for row in reader: data_queue.put(row) def on_start(self): # 每个用户从队列中取一条数据，取完后可以循环或停止 try: self.user_data = self.data_queue.get_nowait() except queue.Empty: # 数据用尽，停止这个用户 self.stop(force=True) # 使用取到的数据登录 resp = self.client.post("/login", json=self.user_data) self.token = resp.json().get("token") @task def view_private_page(self): user_id = self.user_data['id'] self.client.get(f"/user/{user_id}/data", headers={"Auth": self.token})

这里使用了queue.Queue，它是线程安全的，可以确保在分布式压测时，多个 Worker 进程也不会取到重复的数据（前提是文件被正确共享）。当队列为空时，新产生的虚拟用户会自动停止。这是一种非常经典和实用的数据驱动模式。

4. 实时 Web UI 监控与结果分析实战

Locust 的 Web UI 是其一大亮点，它提供了一个直观的仪表盘，让我们在压测过程中就能实时掌握系统状态，而不是等到测试结束后再看报告。

4.1 启动测试与界面详解

首先，将你的脚本保存为locustfile.py（这是默认文件名），然后在命令行启动 Locust：

locust

默认会启动一个 Web 服务在http://localhost:8089。打开浏览器访问这个地址，你会看到启动界面。

在启动界面，你需要填写：

Number of users (peak concurrency)：模拟的总用户数（峰值并发数）。Locust 会以指定的孵化速率逐渐增加到这个数量。
Spawn rate (users started/second)：孵化速率，每秒启动多少个用户。
Host：被测试系统的根地址，如http://www.example.com。你的脚本中的相对路径（如/api/login）会拼接到这个 Host 之后。

点击 “Start swarming” 就开始压测了。主界面主要分为以下几个区域：

Statistics（统计表格）：这是核心数据区。显示每个请求类型的详细信息。
- Type：请求路径（由name参数决定）。
- Requests：总请求数。
- Fails：失败数。
- Median, 95%, 99% Response Time：响应时间的中位数、95分位值和99分位值（单位毫秒）。95%响应时间是评估系统性能的关键指标，它表示95%的请求都在这个时间内完成。
- Average Response Time：平均响应时间。
- Requests/s：每秒请求数（RPS），即吞吐量。
- Failures/s：每秒失败数。
Charts（图表）：包含多个实时曲线图。
- Total Requests per Second：总 RPS 变化曲线。
- Response Times (ms)：平均响应时间和95分位响应时间曲线。
- Number of Users：当前活跃用户数曲线。
Failures（失败记录）：列出所有失败的请求，包括错误类型、发生时间和详细信息，点击可以查看异常追踪，是排查问题的第一现场。
Exceptions（异常记录）：记录代码运行中抛出的 Python 异常。
Download Data：可以下载 CSV 格式的报告数据，用于后续离线分析。

4.2 关键指标解读与性能瓶颈定位

看 UI 不能光看热闹，要能从数据中读出故事。

RPS 上不去，但用户数在涨：这是最典型的瓶颈信号。意味着系统吞吐量已经达到极限，即使增加并发用户，系统也无法处理更多请求。此时响应时间通常会急剧上升。你需要去检查服务器的 CPU、内存、网络 I/O 或数据库连接池等资源是否耗尽。
95% 响应时间与平均响应时间差距过大：如果95%响应时间远高于平均响应时间（比如平均200ms，95%值却高达2000ms），说明系统存在长尾问题。大部分请求很快，但有一小部分请求异常慢。这可能是因为某些请求触发了慢查询、Full GC，或者遇到了资源竞争（如数据库锁）。你需要结合失败和异常日志，分析那5%的慢请求有什么共同特征。
失败率突然飙升：伴随失败率飙升的，往往是响应时间暴涨和 RPS 骤降。这通常是系统某个组件彻底崩溃（如数据库连接超时、第三方服务不可用）或达到限流阈值的结果。立即查看 “Failures” 标签页，确定错误类型（5xx 服务器错误？4xx 客户端错误？连接超时？）。

实操心得：我习惯在压测时同时打开服务器的资源监控（如htop,nmon或云平台监控）和 Locust 的图表。当 Locust 上看到 RPS 曲线走平或响应时间曲线陡增时，立刻切换到资源监控，看是 CPU 先打满，还是内存先耗尽，或者是磁盘 IO 等待异常高。这种关联性分析能快速定位瓶颈所在层级（应用层、数据库层、网络层）。

4.3 使用标签（Tags）进行精细化统计

当你的脚本有很多任务时，你可能只想关注其中一部分的统计数据，比如所有“下单流程”相关的请求。Locust 的tag装饰器可以帮我们实现这个功能。

from locust import HttpUser, task, tag, between class TaggedUser(HttpUser): wait_time = between(1, 3) @task @tag('browse') def index(self): self.client.get("/") @task @tag('browse', 'search') def search_product(self): self.client.get("/search?q=book") @task @tag('order') def add_to_cart(self): self.client.get("/cart/add") @task @tag('order', 'payment') def checkout(self): self.client.get("/checkout")

在 Web UI 的统计表格上方，会出现一个 “Filter by tag” 的输入框。输入order，表格就只会显示被打上order标签的请求（即add_to_cart和checkout）的统计数据。输入browse,search（逗号分隔）则会显示带有browse或search标签的请求。这个功能在分析复杂场景下特定业务链路的性能时非常有用。

5. 高级配置与分布式压测

单机运行 Locust 可能受限于机器本身的网络、端口和 CPU 资源，无法模拟足够高的并发。要发起大规模压测，就需要使用分布式模式。

5.1 分布式压测架构与配置

Locust 采用主从（Master-Worker）架构。

Master 节点：负责分发测试任务、收集汇总来自所有 Worker 的测试数据，并托管 Web UI。它本身不模拟用户。
Worker 节点：接收 Master 指令，实际创建和运行虚拟用户，发起请求，并将统计数据实时汇报给 Master。你可以启动多个 Worker 进程，甚至分布在多台物理机器上。

启动命令示例：

启动 Master(在控制机)：
```
locust -f locustfile.py --master --host=http://your-target-system.com
```
默认情况下，Master 会监听 5557 端口（用于 Worker 连接）和 8089 端口（用于 Web UI）。
启动 Worker(在压测机，可以是同一台或多台机器)：
```
locust -f locustfile.py --worker --master-host=192.168.1.100
```
将--master-host指向 Master 节点的 IP 地址。每个 Worker 会启动多个进程（取决于 CPU 核心数）来执行任务。

注意事项与避坑指南：

代码与数据同步：所有 Master 和 Worker 节点上的locustfile.py以及任何它引用的数据文件（如 CSV）必须完全一致。否则会出现不可预知的行为。建议使用版本控制（如 Git）或共享存储（如 NFS）来管理脚本。
依赖包一致：所有节点的 Python 环境和 Locust 等依赖包版本也需要一致。
网络与防火墙：确保 Master 节点的 5557 和 8089 端口对所有 Worker 节点和你的浏览器访问机器开放。防火墙规则是分布式压测最常见的“坑”。
单机多 Worker：即使在一台机器上，你也可以启动一个 Master 和多个 Worker（例如，Worker 数量等于 CPU 逻辑核心数），以充分利用多核性能。命令中的--worker和--master参数可以同时使用。
查看 Worker 连接状态：在 Web UI 的 “Workers” 标签页下，可以看到所有已连接的 Worker 及其状态。

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

Locust 默认提供HttpUser，但它的架构是开放的，你可以通过继承User类并定义client属性，来测试任何协议，例如 WebSocket、gRPC、TCP Socket 等。

from locust import User, task, between import websocket import json import time class WebSocketClient: def __init__(self, host): self.ws = websocket.create_connection(f"ws://{host}/ws") def send(self, message): start_time = time.time() try: self.ws.send(json.dumps(message)) result = self.ws.recv() # 假设服务端会立即回复 total_time = int((time.time() - start_time) * 1000) # 毫秒 events.request_success.fire(request_type="WS", name="send_msg", response_time=total_time, response_length=len(result)) except Exception as e: total_time = int((time.time() - start_time) * 1000) events.request_failure.fire(request_type="WS", name="send_msg", response_time=total_time, exception=e) def close(self): self.ws.close() class WebSocketUser(User): abstract = True # 这是一个抽象基类，不会被直接实例化 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.client = WebSocketClient(self.host) def on_stop(self): self.client.close() class MyWebSocketUser(WebSocketUser): wait_time = between(1, 3) @task def send_ping(self): self.client.send({"action": "ping", "data": "hello"})

在这个例子中，我们创建了一个自定义的WebSocketClient类，并在WebSocketUser中将其赋值给self.client。在任务中，我们调用self.client.send()，并手动使用events.request_success.fire()和events.request_failure.fire()来向 Locust 报告成功和失败，这样这些请求的响应时间才会被统计到 Locust 的报告中。这是测试非 HTTP 协议服务的标准模式。

6. 常见问题排查与实战技巧

在实际使用 Locust 的过程中，你肯定会遇到各种各样的问题。这里我总结了一些高频问题和解决技巧。

6.1 连接与资源类问题

问题：ConnectionError或Timeout大量出现。
- 排查思路：
  1. 检查目标服务：首先确认被测试的服务本身是否存活、网络是否可达。用curl或浏览器手动访问一下。
  2. 检查 Locust 机器资源：在压测机上运行top或htop。如果 Locust Worker 进程的 CPU 或内存占用很高，可能是单机模拟的用户数太多了，达到了压测机本身的瓶颈。此时需要增加 Worker 数量或使用分布式压测。
  3. 调整超时设置： Locust 的HttpUser底层使用requests.Session。你可以在on_start中为self.client配置更长的超时时间。
```
def on_start(self): self.client.timeout = (10, 30) # (连接超时， 读取超时) 单位秒
```
  4. 检查端口限制：单机模拟大量用户时，可能会耗尽本地端口。可以调整系统的本地端口范围。
```
# Linux 临时调整 sysctl -w net.ipv4.ip_local_port_range="1024 65535"
```
问题：压测过程中，RPS 逐渐下降，响应时间逐渐上升。
- 排查思路：这很可能是内存泄漏的迹象。首先检查被测试应用的内存使用情况。其次，检查 Locust 脚本本身：是否在任务中不断追加数据到某个全局列表或字典？是否没有正确关闭连接（比如自定义客户端）？使用分布式模式时，观察 Master 节点的内存是否持续增长，有时大量数据汇聚也可能导致 Master 内存不足。

6.2 脚本与逻辑类问题

问题：所有请求的响应时间都非常平均，且异常地低（如始终<10ms），不符合预期。
- 原因与解决：这通常是请求被缓存了。可能是应用层缓存（如 Redis）、数据库查询缓存，或者是 Locust 脚本中使用了固定参数，导致后端直接返回缓存结果。确保你的测试数据是充分参数化和随机的，并且在测试前清理缓存。对于需要验证性能的接口，可以考虑在请求头中加入缓存禁用标识（如Cache-Control: no-cache）。
问题：如何让一部分用户只执行 A 流程，另一部分用户只执行 B 流程？
- 解决方案：定义多个不同的 User 类，并在启动 Locust 时通过--user-class参数指定。或者，更灵活的方法是在一个 User 类中，通过on_start方法为用户分配一个“角色”，然后在任务选择逻辑中根据这个角色来决定行为。
```
class MultiRoleUser(HttpUser): wait_time = between(1, 3) def on_start(self): self.role = random.choice(["viewer", "buyer"]) @task def task_flow(self): if self.role == "viewer": self.client.get("/browse") else: # buyer self.client.post("/order", json={...})
```
问题：Locust 报告中的 RPS 比我用其他工具（如wrk,ab）测出来的低很多。
- 原因分析： Locust 的 RPS 是每个请求从发起到收到响应的完整周期时间计算出来的，并且包含了任务之间的wait_time。而wrk这类工具通常使用异步 IO，以尽可能高的速率发送请求，不包含思考时间。Locust 模拟的是真实用户的行为节奏，其 RPS 反映的是在特定用户行为和思考时间模型下，系统实际能处理的请求速率，这个值通常低于系统的绝对极限吞吐量。两者视角不同，Locust 的数据对于评估真实用户体验更有价值。

6.3 一个完整的电商场景压测脚本示例

最后，让我们整合以上所有知识点，来看一个模拟电商用户行为的、相对完整的 Locust 脚本示例。它包含了登录、浏览、搜索、加购、下单等行为，并使用了参数化数据和标签。

from locust import HttpUser, task, tag, between, TaskSet import random import queue import csv # 模拟一个商品ID池和用户凭证池 PRODUCT_IDS = [1001, 1002, 1003, 1004, 1005, 1006, 1007, 1008, 1009, 1010] # 从CSV读取用户数据（示例） user_data_queue = queue.Queue() try: with open('users.csv', 'r') as f: reader = csv.DictReader(f) for row in reader: user_data_queue.put(row) except FileNotFoundError: # 如果文件不存在，创建一些模拟数据 for i in range(100): user_data_queue.put({"username": f"test_user_{i}", "password": "default_pass"}) class BrowseBehavior(TaskSet): """浏览行为任务集""" @tag('browse') @task(5) def view_homepage(self): self.client.get("/", name="首页") @tag('browse', 'search') @task(3) def search_product(self): keyword = random.choice(["手机", "笔记本", "耳机", "书籍"]) with self.client.get(f"/search?q={keyword}", name="/search", catch_response=True) as resp: if resp.status_code == 200 and len(resp.json()['list']) > 0: resp.success() else: resp.failure("搜索无结果或失败") @tag('browse', 'detail') @task(2) def view_product_detail(self): product_id = random.choice(PRODUCT_IDS) with self.client.get(f"/product/{product_id}", name="/product/[id]", catch_response=True) as resp: if resp.status_code == 200: # 可以在这里解析响应，获取库存等信息，用于后续判断 self.product_info = resp.json() resp.success() else: resp.failure("获取商品详情失败") class OrderBehavior(TaskSet): """下单行为任务集，通常依赖于浏览后""" def on_start(self): # 进入下单流程，先确保有商品在购物车，这里简单模拟 self.cart_items = [random.choice(PRODUCT_IDS) for _ in range(random.randint(1, 3))] @tag('order') @task def checkout(self): # 假设购物车API返回需要结算的信息 cart_resp = self.client.get("/api/cart", name="获取购物车") if cart_resp.status_code != 200: self.interrupt() # 购物车为空或异常，跳出下单流程 # 提交订单 order_data = {"items": self.cart_items, "address_id": 1} with self.client.post("/api/order/create", json=order_data, name="提交订单", catch_response=True) as resp: if resp.status_code == 200 and resp.json().get('order_id'): resp.success() # 订单创建成功，可以继续支付等，这里简化处理 self.order_id = resp.json()['order_id'] else: resp.failure(f"创建订单失败: {resp.text}") self.interrupt() # 完成一次下单，跳出 class WebsiteUser(HttpUser): host = "http://shop.demo.com" # 在Web UI或命令行中可以覆盖 wait_time = between(2, 5) # 用户操作间隔较长 def on_start(self): """用户初始化：登录""" try: creds = user_data_queue.get_nowait() except queue.Empty: print("测试用户数据已用尽，停止生成新用户") self.stop(force=True) return self.username = creds['username'] with self.client.post("/api/login", json=creds, name="用户登录", catch_response=True) as resp: if resp.status_code == 200: self.token = resp.json()['token'] self.client.headers.update({'Authorization': f'Bearer {self.token}'}) resp.success() # 登录后，将凭证放回队列尾部，实现数据循环使用（根据测试目的选择） # user_data_queue.put(creds) else: resp.failure("登录失败") self.stop(force=True) # 登录失败，此用户停止活动 tasks = [BrowseBehavior, OrderBehavior] # 用户会随机执行浏览或下单任务集 # 可以通过权重控制用户更倾向于浏览还是下单 # task_weight = {BrowseBehavior: 3, OrderBehavior: 1} def on_stop(self): """用户停止时，可选的清理操作""" # 例如：登出 # self.client.post("/api/logout") pass

这个脚本展示了数据池、标签、嵌套 TaskSet、响应验证、条件判断等多项技术的综合运用。你可以根据实际的业务 API 对其进行修改和扩展。记住，一个好的性能测试脚本，其核心目标是真实、可控、可度量地模拟线上用户行为，从而为系统性能评估和容量规划提供可靠依据。

查看全文

http://www.jsqmd.com/news/1272263/