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: fBearer {token}} # 3. 获取用户信息依赖登录态 self.client.get(/api/user/profile, headersself.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?qlaptop) # 模拟用户查看搜索结果列表的思考时间 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, jsoncred) if resp.ok: self.token resp.json()[token] else: # 登录失败可以标记此用户停止运行或重试 self.stop(forceTrue) 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(forceTrue) # 使用取到的数据登录 resp self.client.post(/login, jsonself.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这是默认文件名然后在命令行启动 Locustlocust默认会启动一个 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%响应时间远高于平均响应时间比如平均200ms95%值却高达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?qbook) 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 --hosthttp://your-target-system.com默认情况下Master 会监听 5557 端口用于 Worker 连接和 8089 端口用于 Web UI。启动 Worker(在压测机可以是同一台或多台机器)locust -f locustfile.py --worker --master-host192.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(fws://{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_typeWS, namesend_msg, response_timetotal_time, response_lengthlen(result)) except Exception as e: total_time int((time.time() - start_time) * 1000) events.request_failure.fire(request_typeWS, namesend_msg, response_timetotal_time, exceptione) 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大量出现。排查思路检查目标服务 首先确认被测试的服务本身是否存活、网络是否可达。用curl或浏览器手动访问一下。检查 Locust 机器资源 在压测机上运行top或htop。如果 Locust Worker 进程的 CPU 或内存占用很高可能是单机模拟的用户数太多了达到了压测机本身的瓶颈。此时需要增加 Worker 数量或使用分布式压测。调整超时设置 Locust 的HttpUser底层使用requests.Session。你可以在on_start中为self.client配置更长的超时时间。def on_start(self): self.client.timeout (10, 30) # (连接超时 读取超时) 单位秒检查端口限制 单机模拟大量用户时可能会耗尽本地端口。可以调整系统的本地端口范围。# Linux 临时调整 sysctl -w net.ipv4.ip_local_port_range1024 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: ftest_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_responseTrue) 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_responseTrue) 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, jsonorder_data, name提交订单, catch_responseTrue) 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(forceTrue) return self.username creds[username] with self.client.post(/api/login, jsoncreds, name用户登录, catch_responseTrue) as resp: if resp.status_code 200: self.token resp.json()[token] self.client.headers.update({Authorization: fBearer {self.token}}) resp.success() # 登录后将凭证放回队列尾部实现数据循环使用根据测试目的选择 # user_data_queue.put(creds) else: resp.failure(登录失败) self.stop(forceTrue) # 登录失败此用户停止活动 tasks [BrowseBehavior, OrderBehavior] # 用户会随机执行浏览或下单任务集 # 可以通过权重控制用户更倾向于浏览还是下单 # task_weight {BrowseBehavior: 3, OrderBehavior: 1} def on_stop(self): 用户停止时可选的清理操作 # 例如登出 # self.client.post(/api/logout) pass这个脚本展示了数据池、标签、嵌套 TaskSet、响应验证、条件判断等多项技术的综合运用。你可以根据实际的业务 API 对其进行修改和扩展。记住一个好的性能测试脚本其核心目标是真实、可控、可度量地模拟线上用户行为从而为系统性能评估和容量规划提供可靠依据。