1. 项目概述为什么我们需要一个“蝗虫”来压测如果你做过性能测试大概率用过JMeter或者LoadRunner。这些工具功能强大但配置复杂脚本编写尤其是JMeter的BeanShell对很多开发者和测试同学来说体验并不友好。当我们需要模拟成千上万个用户并且希望测试脚本能像写业务代码一样灵活时Locust就进入了视野。Locust中文直译是“蝗虫”寓意其能像蝗虫群一样发起海量并发请求。它是一个用Python编写的开源负载测试工具。和传统工具最大的不同在于它完全采用代码来定义用户行为。这意味着你可以用你熟悉的Python语法结合Requests库、异步库甚至任何自定义的客户端来模拟任何你想模拟的业务场景。测试脚本就是普通的Python文件这为版本管理、参数化、逻辑复用带来了极大的便利。它的架构也很有意思采用主从Master-Worker模式一个Master节点负责协调和收集数据多个Worker节点可以分布在不同的机器上负责真正执行测试脚本、产生负载。这种分布式设计让它能轻松发起百万级别的并发。对于开发、测试和运维来说Locust提供了一种更“程序员友好”的性能验证方式尤其适合在敏捷开发和持续集成流程中快速验证API、微服务乃至整个Web应用的承压能力。2. Locust核心设计哲学与架构拆解2.1 事件驱动与协程高并发的基石Locust能轻松支持单机数千并发其核心秘密在于它基于gevent库。gevent是一个基于协程的Python网络库。要理解它的优势得先看看传统多线程/多进程模型的瓶颈。在传统的压测工具中每一个虚拟用户VU通常对应一个操作系统线程或进程。当你有几千个VU时就需要创建几千个线程。线程的创建、销毁和上下文切换会消耗大量CPU和内存资源这就是为什么很多工具单机并发数上不去的原因。Locust采用了不同的思路。它使用gevent提供的greenlet微线程即协程。成千上万个协程可以在同一个操作系统线程内运行。当一个协程遇到I/O操作比如等待HTTP响应时它会主动让出控制权gevent的调度器会立刻切换到另一个就绪的协程继续执行。对于负载测试这种绝大部分时间都在等待网络I/O的场景来说这种模型效率极高。单机一个进程就能轻松承载数千个活跃的虚拟用户资源消耗远小于线程模型。注意gevent通过猴子补丁monkey patch来修改Python标准库的socket等模块使其变为非阻塞式。在Locust脚本中你通常会在文件开头看到from gevent import monkey; monkey.patch_all()。这行代码至关重要它确保了像requests.get()这样的同步HTTP调用也能被gevent正确调度实现异步效果。如果你忘记添加所有虚拟用户将会变成串行执行完全达不到并发效果。2.2 Master-Worker分布式架构解析当单机性能无法满足超大规模并发需求时就需要分布式执行。Locust采用了经典的主从架构。Master节点是大脑。它不产生任何负载只负责三件事协调启动测试、停止测试、告诉Worker要模拟多少用户、以多快的速率孵化。收集接收所有Worker节点发回的实时统计数据请求数、响应时间、失败率等。展示运行一个Web UI默认端口8089提供实时图表和控制界面。Worker节点是肌肉。每个Worker节点都会完整加载你的测试脚本并独立运行其中的用户类User。它们根据Master的指令创建和管理指定数量的协程虚拟用户执行定义好的任务流并将请求结果实时发送回Master进行汇总。这种架构的优势在于线性扩展。理论上只要增加Worker节点就能近乎线性地提升能产生的总并发用户数。在实际部署时你需要确保Master和Worker网络互通并且Worker能够访问到被测试系统。2.3 核心概念User、TaskSet与Events理解Locust脚本关键是理解三个核心类。HttpUser (以及基类User)这是虚拟用户的蓝图。每个模拟用户都是这个类的一个实例。HttpUser是User的子类它内置了一个client属性这是一个HttpSession对象类似requests.Session用于发送HTTP请求。你可以继承HttpUser来定义一类用户的行为。from locust import HttpUser, task, between class MyWebsiteUser(HttpUser): wait_time between(1, 5) # 每个任务执行后等待1-5秒 host https://www.example.com # 被测系统的基础URL task(3) # task装饰器权重为3 def view_homepage(self): self.client.get(/) task(1) # 权重为1执行频率是view_homepage的1/3 def view_about(self): self.client.get(/about)Tasks 与 task装饰器用户做什么就由task来定义。task装饰器将一个方法标记为一个任务。权重参数决定了该任务被执行的相对频率。在上面的例子中view_homepage任务被执行的概率是view_about任务的3倍。wait_time定义了任务执行之间的等待时间用于模拟用户思考或浏览时间使负载更真实。TaskSet 类用于将任务分组和嵌套模拟复杂的用户操作流程。例如一个电商用户的行为可以分组为“浏览行为”和“购买行为”。from locust import HttpUser, task, TaskSet, between class BrowseProducts(TaskSet): task(4) def view_product_list(self): self.client.get(/api/products) task(1) def search_product(self): self.client.get(/api/products?searchlocust) task(1) def stop(self): self.interrupt() # 关键中断当前TaskSet返回父级 class PurchaseFlow(TaskSet): task def add_to_cart(self): self.client.post(/api/cart, json{product_id: 1}) task def checkout(self): self.client.post(/api/checkout) class ECommerceUser(HttpUser): wait_time between(2, 8) tasks [BrowseProducts, PurchaseFlow] # 直接声明TaskSet类 # 或者通过task装饰器 # task(5) # def browse(self): # self.execute_task_set(BrowseProducts)Events事件钩子机制是Locust的扩展点。你可以在测试生命周期的特定时刻注入自定义逻辑例如在测试开始时准备测试数据在请求成功后校验复杂的响应体或在测试停止时清理环境。from locust import events from locust.runners import MasterRunner events.test_start.add_listener def on_test_start(environment, **kwargs): if isinstance(environment.runner, MasterRunner): print(测试在Master上启动可以在这里初始化全局数据) # 例如清空测试数据库准备一批测试账号 events.request.add_listener def on_request(request_type, name, response_time, response_length, exception, context, **kwargs): if exception: print(f请求失败: {name}, 异常: {exception}) elif response_time 5000: # 自定义断言响应时间超过5秒记录警告 print(f警告请求 {name} 响应缓慢: {response_time}ms)3. 从零到一编写你的第一个Locust压测脚本3.1 环境搭建与基础脚本结构首先通过pip安装Locust。建议使用虚拟环境。pip install locust验证安装locust --version。一个最基础的Locust脚本比如命名为locustfile.py包含以下要素必要的导入from locust import HttpUser, task, between用户类定义继承HttpUser。host属性指定被测系统的根URL。也可以在命令行用--host参数覆盖。wait_time属性定义任务间隔。常用between(min, max)随机等待或constant(n)固定等待。task装饰的方法定义用户行为。一个完整的“Hello World”脚本如下from locust import HttpUser, task, between class QuickstartUser(HttpUser): wait_time between(1, 2.5) host https://httpbin.org # 一个用于测试的公共API task def get_status(self): # self.client是HttpSession用法同requests response self.client.get(/status/200) # 你可以对响应进行断言 if response.status_code ! 200: response.failure(fUnexpected status code: {response.status_code}) task(3) # 这个任务执行频率是get_status的3倍 def get_delay(self): self.client.get(/delay/1) # 模拟一个延迟1秒的接口这个脚本定义了一类用户他们以1到2.5秒的间隔随机执行两个任务一个检查状态码另一个调用一个延迟接口。get_delay任务由于权重为3被选中的概率更高。3.2 模拟真实场景参数化、关联与断言真实的业务场景不会总是请求固定的URL。我们需要参数化、处理Cookie/Session、从上一个请求的响应中提取数据用于下一个请求关联并对响应进行断言。参数化数据可以使用Python的任何方式来准备数据例如列表、从文件读取、从数据库查询等。import csv from locust import HttpUser, task, between class ParameterizedUser(HttpUser): wait_time between(1, 3) host https://your-api.com def on_start(self): 每个虚拟用户开始运行时执行一次用于初始化 self.product_ids [1001, 1002, 1003, 1004] self.current_index 0 task def view_product(self): # 简单的轮询参数化 pid self.product_ids[self.current_index % len(self.product_ids)] self.current_index 1 with self.client.get(f/api/product/{pid}, name/api/product/[id], catch_responseTrue) as response: # catch_responseTrue允许我们手动控制请求的成功/失败 if response.status_code 200 and stock in response.text: response.success() # 可以在这里提取数据例如库存量 # stock response.json().get(stock) else: response.failure(fFailed to get product {pid}) # 更常见的做法是使用队列或随机选择 # from random import choice # pid choice(self.product_ids)处理关联提取Token很多API需要先登录获取token后续请求携带此token。class UserWithAuth(HttpUser): host https://your-api.com wait_time between(2, 5) def on_start(self): 用户登录获取token login_resp self.client.post(/api/login, json{username: test, password: test}) if login_resp.status_code 200: self.token login_resp.json().get(access_token) self.headers {Authorization: fBearer {self.token}} else: self.token None # 如果登录失败可以标记这个用户实例为失败或者停止任务 self.environment.runner.stop_user(self) task def get_profile(self): if not self.token: return # 在请求中带上认证头 self.client.get(/api/profile, headersself.headers)复杂断言与响应校验使用catch_responseTrue上下文管理器可以精细控制请求的成功与失败。task def create_order(self): payload {product_id: 1001, quantity: 2} with self.client.post(/api/order, jsonpayload, catch_responseTrue, name/api/order) as resp: # 校验HTTP状态码 if resp.status_code ! 201: resp.failure(f创建订单失败状态码{resp.status_code}) return # 校验JSON响应体结构 try: data resp.json() if not data.get(order_id): resp.failure(响应中缺少order_id字段) elif data.get(total_price) 0: resp.failure(订单总价异常) else: resp.success() self.order_id data.get(order_id) # 保存供后续使用 except JSONDecodeError: resp.failure(响应不是有效的JSON)3.3 使用WaitTime与Weight控制负载模型负载的“形状”由wait_time和任务权重共同决定。wait_time策略between(min, max)最常用模拟用户随机思考时间。constant(n)每个任务后固定等待n秒适用于节奏固定的轮询场景。constant_pacing(n)更推荐用于稳定性测试。它试图确保每个任务包括执行时间和等待时间的总耗时至少为n秒。如果任务执行很快它就等待剩余时间如果任务执行超过n秒它就不等待立即开始下一个。这能更精确地控制每秒发出的请求数RPS。自定义函数你可以实现任何复杂的等待逻辑。from locust import constant_pacing import random class MyUser(HttpUser): # 确保每个任务周期至少5秒 wait_time constant_pacing(5) # 或者自定义80%的情况等1-3秒20%的情况等10-15秒模拟用户离开 # wait_time lambda self: random.uniform(1, 3) if random.random() 0.2 else random.uniform(10, 15)任务权重与动态任务task装饰器的权重参数是静态的。有时我们需要根据运行时状态动态决定下一个任务。可以通过重写User的tasks属性为一个返回任务列表的方法来实现。from locust import HttpUser, task, between class DynamicUser(HttpUser): wait_time between(1, 3) # 这是一个属性返回一个任务列表 property def tasks(self): task_list [self.view_item] if self.is_vip: # 假设有一个vip标志位 task_list.append(self.vip_discount) return task_list def on_start(self): self.is_vip random.random() 0.8 # 20%的用户是VIP task def view_item(self): self.client.get(/item) def vip_discount(self): self.client.get(/vip-only)4. 执行、监控与结果分析实战4.1 单机与分布式执行命令详解单机无Web UI模式适合CI/CDlocust -f locustfile.py --headless -u 100 -r 10 --run-time 1h30m --hosthttps://your-target.com--headless: 无头模式不启动Web UI。-u 100: 设置总用户数为100。-r 10: 设置孵化速率Hatch rate为每秒10个用户直到达到总用户数。--run-time: 设置测试运行时长格式如1h30m20s。--host: 覆盖脚本中定义的host。输出会在控制台测试结束后会生成摘要。单机启动Web UIlocust -f locustfile.py --hosthttps://your-target.com访问http://localhost:8089即可打开控制界面。在UI中你可以动态设置用户数、孵化速率并点击开始。分布式执行启动Master节点不产生负载locust -f locustfile.py --master --hosthttps://your-target.com在每台Worker机器上启动Worker节点假设Master IP是192.168.1.100locust -f locustfile.py --worker --master-host192.168.1.100打开Master节点的Web UIhttp://192.168.1.100:8089你会看到连接的Worker数量。然后像单机一样启动测试即可。实操心得在分布式执行时确保所有Worker节点上的locustfile.py脚本和Python环境特别是第三方库版本完全一致。否则可能会出现不可预知的行为。建议使用Docker容器或配置管理工具来保证环境一致性。4.2 Web UI界面深度解读Locust的Web UI虽然简洁但信息密度很高。状态标签页显示当前用户数、孵化速率、RPS、失败率以及实时图表。图表包括总RPS、响应时间平均、中位数、P95、P99和用户数。重点关注P95/P99响应时间它们比平均响应时间更能反映尾部延迟对用户体验影响更大。图表响应时间和RPS的趋势图是定位性能问题的关键。如果随着用户数增加响应时间曲线陡然上升而RPS曲线持平甚至下降说明系统已经达到瓶颈。失败率任何非零的失败率都需要警惕。点击“失败”标签页可以查看具体的失败请求和异常信息。下载数据测试结束后务必从“Download Data”选项卡下载CSV报告。包含请求统计、响应时间分位数、异常统计等用于后续详细分析。4.3 关键性能指标KPI与结果分析运行完测试Locust会在控制台输出一个摘要并可以生成HTML报告。你需要学会解读这些数据。Requests/s (RPS)每秒请求数。这是吞吐量的核心指标。在负载增加时观察RPS是否线性增长。当系统达到瓶颈时RPS会达到峰值并开始波动或下降。Response Times (ms)Average平均响应时间参考价值一般。Median (50%)中位数响应时间有一半的请求比它快。95th / 99th percentile (P95/P99)最重要的指标之一。95%的请求响应时间低于此值。它反映了绝大多数用户的体验。P99则反映了最慢的那1%请求的情况对于高要求系统尤其关键。Failure %失败请求的百分比。在性能测试中即使响应慢只要没超时或返回错误码就不算失败。你需要根据业务定义合理的超时时间和断言条件。Number of Users并发用户数。注意Locust中的“用户”是活跃的、正在执行任务的协程。它不等于在线用户数更接近同时发起请求的用户数。如何分析寻找拐点逐步增加负载用户数绘制RPS和P95响应时间随用户数变化的曲线。当P95响应时间开始非线性急剧上升而RPS增长放缓或停滞时这个点就是系统的性能拐点即最大有效吞吐量点。对比基线每次代码发布或配置变更后用相同的负载模型进行测试对比关键指标如P95响应时间、RPS。如果出现显著退化就需要排查。关注错误分析失败请求的日志看是超时、4xx/5xx错误还是自定义断言失败。这往往是系统bug或配置问题的直接体现。5. 高级技巧与生产环境最佳实践5.1 自定义客户端测试非HTTP协议Locust的魔力在于其User类不限于HTTP。你可以继承基类User使用任何Python客户端库。示例使用WebSocket客户端from locust import User, task, between, events import websocket import json import time class WebSocketUser(User): wait_time between(1, 5) def on_start(self): # 建立WebSocket连接 ws_url ws://your-websocket-server/ws self.ws websocket.create_connection(ws_url) # 可以监听连接事件 events.user_started.fire(user_instanceself) def on_stop(self): # 关闭连接 if self.ws: self.ws.close() events.user_stopped.fire(user_instanceself) task def send_message(self): message json.dumps({type: ping, data: int(time.time())}) start_time time.time() try: self.ws.send(message) result self.ws.recv() # 接收响应 response_time int((time.time() - start_time) * 1000) # 毫秒 # 手动触发请求成功事件用于统计 events.request.fire( request_typeWS, nameping, response_timeresponse_time, response_lengthlen(result), exceptionNone, context{} ) # 可以在这里对result进行断言 except Exception as e: response_time int((time.time() - start_time) * 1000) events.request.fire( request_typeWS, nameping, response_timeresponse_time, response_length0, exceptione, context{} )通过手动触发events.request事件你可以将任何协议的请求纳入Locust的统计系统中。同理你可以测试gRPC、TCP、MQTT等任何协议。5.2 利用Events钩子实现复杂逻辑Events系统是Locust的瑞士军刀。除了之前提到的test_start和request还有几个常用钩子init在Locust环境初始化时触发可用于解析命令行参数、注册自定义参数。quitting在Locust进程退出前触发用于清理资源。user_error当用户代码如on_start或task方法发生未处理异常时触发。实战案例在每个Worker上独立准备测试数据在分布式模式下test_start事件只在Master上触发一次。如果每个Worker需要独立准备数据比如从数据库读取一批独立的测试账号可以使用init事件。from locust import events, User import pymysql import random test_accounts [] # 全局变量但每个Worker进程独立 events.init.add_listener def on_locust_init(environment, **kwargs): 在每个进程Master和每个Worker初始化时执行 global test_accounts # 连接数据库读取测试账号 db pymysql.connect(host..., user..., password..., database...) cursor db.cursor() cursor.execute(SELECT username, password FROM test_users WHERE statusactive) test_accounts cursor.fetchall() cursor.close() db.close() print(f[初始化] 进程 {os.getpid()} 加载了 {len(test_accounts)} 个测试账号) class ApiUser(User): def on_start(self): if test_accounts: self.username, self.password random.choice(test_accounts) # 使用这个账号登录...5.3 集成到CI/CD流水线将性能测试左移集成到CI/CD中是保障线上稳定的重要手段。基本集成步骤编写稳定的测试脚本脚本要能独立运行不依赖外部不确定状态。使用--headless模式。定义性能基准确定核心接口的P95响应时间、RPS等合格线。创建CI任务在Jenkins、GitLab CI、GitHub Actions中增加一个性能测试阶段。执行与断言运行Locust命令并解析其输出或生成的报告进行断言。示例使用pytest和locust的插件可以使用pytest来组织和管理Locust测试并利用locust-plugins等第三方库增强功能。# .github/workflows/performance.yml 示例 (GitHub Actions) name: Performance Test on: [push] jobs: load-test: runs-on: ubuntu-latest steps: - uses: actions/checkoutv2 - name: Set up Python uses: actions/setup-pythonv2 with: { python-version: 3.9 } - name: Install dependencies run: pip install locust - name: Run load test run: | locust -f locustfile.py \ --headless \ -u 50 -r 5 \ --run-time 2m \ --host ${{ secrets.TEST_HOST }} \ --csvresults \ --htmlreport.html # 注意这里需要先启动被测试服务或确保服务已就绪 - name: Check performance thresholds run: | # 使用一个Python脚本解析results_stats.csv检查P95是否小于200ms python check_perf.py - name: Upload report uses: actions/upload-artifactv2 with: { name: locust-report, path: report.html }check_perf.py是一个自定义脚本用于解析Locust生成的CSV结果文件并判断是否通过。# check_perf.py import csv import sys def check_performance(): with open(results_stats.csv, r) as f: reader csv.DictReader(f) for row in reader: if row[Name] Aggregated: # 查看汇总行 p95 float(row[95%]) failure_rate float(row[Failure Rate]) print(fAggregated P95: {p95}ms, Failure Rate: {failure_rate*100}%) if p95 200: # 阈值设为200ms print(fERROR: P95响应时间 {p95}ms 超过阈值200ms) sys.exit(1) if failure_rate 0.01: # 失败率阈值1% print(fERROR: 失败率 {failure_rate*100}% 超过阈值1%) sys.exit(1) print(性能测试通过) sys.exit(0) print(未找到聚合数据) sys.exit(1) if __name__ __main__: check_performance()6. 常见问题排查与性能调优指南6.1 Locust自身性能瓶颈与优化有时候压测工具本身会成为瓶颈。以下迹象可能表明是Locust Worker资源不足Worker节点的CPU或内存使用率接近100%。增加Worker节点总RPS却没有线性增长。Master的Web UI出现卡顿或数据更新延迟。优化策略增加Worker节点这是最直接的方法。确保Worker机器有足够的CPU和网络资源。优化测试脚本避免在任务中执行繁重的计算或同步I/O比如频繁读写大文件、进行复杂的字符串处理。这些操作会阻塞协程影响并发能力。将数据预加载到内存中。谨慎使用catch_responseTrue和手动触发事件这些操作有一定开销。对于简单的成功/失败判断直接依赖HTTP状态码更高效。使用更高效的HTTP客户端Locust默认的HttpSession基于requests。对于极高并发可以尝试使用aiohttp作为客户端需要自定义User类。调整Locust配置--expect-workers在Master启动时指定期望的Worker数量Master会等待所有Worker连接后才允许启动测试避免数据不一致。监控Master和Worker的日志查看是否有错误或警告。6.2 被测系统监控与瓶颈定位当Locust报告性能不佳时下一步是定位被测系统的瓶颈。你需要结合系统监控来看。应用服务器指标CPU使用率、内存使用量、线程池状态、GC频率。如果CPU持续高位可能是计算瓶颈或低效代码如果内存持续增长可能有内存泄漏。数据库指标查询QPS、慢查询数量、连接数、锁等待。性能问题常常出在数据库。关注Locust运行期间出现的慢查询日志。中间件指标Redis/Memcached的命中率、Kafka的堆积情况、Nginx的活跃连接数。网络与操作系统指标网络带宽、TCP重传率、磁盘IO等待。一个典型的排查流程现象随着用户数增加响应时间变长RPS上不去。检查Locust Worker资源使用正常排除工具瓶颈。检查应用服务器发现CPU使用率很低但线程池活跃线程数很高很多线程处于BLOCKED或WAITING状态。推断线程在等待某个外部资源可能是数据库连接。检查数据库发现数据库连接池已满大量活跃连接执行缓慢的SQL。定位分析慢查询日志找到没有使用索引的查询或锁竞争激烈的表。解决优化SQL增加索引或调整业务逻辑。6.3 脚本编写中的典型“坑”与避坑指南坑数据竞争导致断言失败或逻辑错误场景多个虚拟用户共享一个全局列表来获取测试数据导致同一数据被多个用户使用。解决为每个虚拟用户实例独立准备数据或在on_start中分配。对于需要全局唯一且消耗性的数据如订单号使用线程安全的队列queue.Queue或分布式锁如果Worker跨机器。坑连接未关闭导致资源泄漏场景在自定义客户端如直接使用requests或数据库连接时没有正确关闭连接。解决在User类的on_stop方法中确保清理资源。对于HTTPLocust的client会自动管理连接池。坑wait_time设置不当导致负载不真实场景使用constant(0)即无等待连续发送请求。这会产生“脉冲式”压力可能瞬间打满系统队列掩盖了系统在稳定流下的真实表现。解决始终设置合理的wait_time如between(1, 3)以模拟真实用户的思考时间。对于稳定性测试使用constant_pacing来控制稳定的RPS。坑断言过于宽松或严格导致失败率失真场景只检查HTTP状态码为200但业务上响应体可能表示失败如{code: 500, msg: error}。解决结合catch_responseTrue对响应体进行充分校验。但同时也要注意不要将一些非核心的字段校验失败定义为请求完全失败这可能会高估系统的功能故障率。可以区分“性能失败”和“业务失败”。坑分布式运行时测试启动时间不一致场景Master点击启动后各个Worker加载脚本、连接Master、接收指令有微小延迟导致初始时刻负载不均衡。解决对于短时测试如1分钟影响较大。可以适当增加--step-load步进加载的步长时间或忽略测试开始前几秒的数据。对于长时间稳定性测试此影响可忽略。我个人在实际使用Locust进行全链路压测时最深的一点体会是脚本的可靠性和可维护性比追求极致的并发数更重要。一个结构清晰、数据隔离良好、断言完备的脚本虽然可能在单机并发上稍微损失一点性能但它能让你在分析结果时充满信心快速定位问题是出在脚本还是被测系统。尤其是在CI/CD中稳定、可重复的测试才是王道。花时间设计好测试数据模型和用户行为流往往比盲目调高并发用户数能发现更多有价值的性能问题。
Locust性能测试实战：从原理到分布式压测与CI/CD集成
发布时间：2026/6/18 21:04:06
1. 项目概述为什么我们需要一个“蝗虫”来压测如果你做过性能测试大概率用过JMeter或者LoadRunner。这些工具功能强大但配置复杂脚本编写尤其是JMeter的BeanShell对很多开发者和测试同学来说体验并不友好。当我们需要模拟成千上万个用户并且希望测试脚本能像写业务代码一样灵活时Locust就进入了视野。Locust中文直译是“蝗虫”寓意其能像蝗虫群一样发起海量并发请求。它是一个用Python编写的开源负载测试工具。和传统工具最大的不同在于它完全采用代码来定义用户行为。这意味着你可以用你熟悉的Python语法结合Requests库、异步库甚至任何自定义的客户端来模拟任何你想模拟的业务场景。测试脚本就是普通的Python文件这为版本管理、参数化、逻辑复用带来了极大的便利。它的架构也很有意思采用主从Master-Worker模式一个Master节点负责协调和收集数据多个Worker节点可以分布在不同的机器上负责真正执行测试脚本、产生负载。这种分布式设计让它能轻松发起百万级别的并发。对于开发、测试和运维来说Locust提供了一种更“程序员友好”的性能验证方式尤其适合在敏捷开发和持续集成流程中快速验证API、微服务乃至整个Web应用的承压能力。2. Locust核心设计哲学与架构拆解2.1 事件驱动与协程高并发的基石Locust能轻松支持单机数千并发其核心秘密在于它基于gevent库。gevent是一个基于协程的Python网络库。要理解它的优势得先看看传统多线程/多进程模型的瓶颈。在传统的压测工具中每一个虚拟用户VU通常对应一个操作系统线程或进程。当你有几千个VU时就需要创建几千个线程。线程的创建、销毁和上下文切换会消耗大量CPU和内存资源这就是为什么很多工具单机并发数上不去的原因。Locust采用了不同的思路。它使用gevent提供的greenlet微线程即协程。成千上万个协程可以在同一个操作系统线程内运行。当一个协程遇到I/O操作比如等待HTTP响应时它会主动让出控制权gevent的调度器会立刻切换到另一个就绪的协程继续执行。对于负载测试这种绝大部分时间都在等待网络I/O的场景来说这种模型效率极高。单机一个进程就能轻松承载数千个活跃的虚拟用户资源消耗远小于线程模型。注意gevent通过猴子补丁monkey patch来修改Python标准库的socket等模块使其变为非阻塞式。在Locust脚本中你通常会在文件开头看到from gevent import monkey; monkey.patch_all()。这行代码至关重要它确保了像requests.get()这样的同步HTTP调用也能被gevent正确调度实现异步效果。如果你忘记添加所有虚拟用户将会变成串行执行完全达不到并发效果。2.2 Master-Worker分布式架构解析当单机性能无法满足超大规模并发需求时就需要分布式执行。Locust采用了经典的主从架构。Master节点是大脑。它不产生任何负载只负责三件事协调启动测试、停止测试、告诉Worker要模拟多少用户、以多快的速率孵化。收集接收所有Worker节点发回的实时统计数据请求数、响应时间、失败率等。展示运行一个Web UI默认端口8089提供实时图表和控制界面。Worker节点是肌肉。每个Worker节点都会完整加载你的测试脚本并独立运行其中的用户类User。它们根据Master的指令创建和管理指定数量的协程虚拟用户执行定义好的任务流并将请求结果实时发送回Master进行汇总。这种架构的优势在于线性扩展。理论上只要增加Worker节点就能近乎线性地提升能产生的总并发用户数。在实际部署时你需要确保Master和Worker网络互通并且Worker能够访问到被测试系统。2.3 核心概念User、TaskSet与Events理解Locust脚本关键是理解三个核心类。HttpUser (以及基类User)这是虚拟用户的蓝图。每个模拟用户都是这个类的一个实例。HttpUser是User的子类它内置了一个client属性这是一个HttpSession对象类似requests.Session用于发送HTTP请求。你可以继承HttpUser来定义一类用户的行为。from locust import HttpUser, task, between class MyWebsiteUser(HttpUser): wait_time between(1, 5) # 每个任务执行后等待1-5秒 host https://www.example.com # 被测系统的基础URL task(3) # task装饰器权重为3 def view_homepage(self): self.client.get(/) task(1) # 权重为1执行频率是view_homepage的1/3 def view_about(self): self.client.get(/about)Tasks 与 task装饰器用户做什么就由task来定义。task装饰器将一个方法标记为一个任务。权重参数决定了该任务被执行的相对频率。在上面的例子中view_homepage任务被执行的概率是view_about任务的3倍。wait_time定义了任务执行之间的等待时间用于模拟用户思考或浏览时间使负载更真实。TaskSet 类用于将任务分组和嵌套模拟复杂的用户操作流程。例如一个电商用户的行为可以分组为“浏览行为”和“购买行为”。from locust import HttpUser, task, TaskSet, between class BrowseProducts(TaskSet): task(4) def view_product_list(self): self.client.get(/api/products) task(1) def search_product(self): self.client.get(/api/products?searchlocust) task(1) def stop(self): self.interrupt() # 关键中断当前TaskSet返回父级 class PurchaseFlow(TaskSet): task def add_to_cart(self): self.client.post(/api/cart, json{product_id: 1}) task def checkout(self): self.client.post(/api/checkout) class ECommerceUser(HttpUser): wait_time between(2, 8) tasks [BrowseProducts, PurchaseFlow] # 直接声明TaskSet类 # 或者通过task装饰器 # task(5) # def browse(self): # self.execute_task_set(BrowseProducts)Events事件钩子机制是Locust的扩展点。你可以在测试生命周期的特定时刻注入自定义逻辑例如在测试开始时准备测试数据在请求成功后校验复杂的响应体或在测试停止时清理环境。from locust import events from locust.runners import MasterRunner events.test_start.add_listener def on_test_start(environment, **kwargs): if isinstance(environment.runner, MasterRunner): print(测试在Master上启动可以在这里初始化全局数据) # 例如清空测试数据库准备一批测试账号 events.request.add_listener def on_request(request_type, name, response_time, response_length, exception, context, **kwargs): if exception: print(f请求失败: {name}, 异常: {exception}) elif response_time 5000: # 自定义断言响应时间超过5秒记录警告 print(f警告请求 {name} 响应缓慢: {response_time}ms)3. 从零到一编写你的第一个Locust压测脚本3.1 环境搭建与基础脚本结构首先通过pip安装Locust。建议使用虚拟环境。pip install locust验证安装locust --version。一个最基础的Locust脚本比如命名为locustfile.py包含以下要素必要的导入from locust import HttpUser, task, between用户类定义继承HttpUser。host属性指定被测系统的根URL。也可以在命令行用--host参数覆盖。wait_time属性定义任务间隔。常用between(min, max)随机等待或constant(n)固定等待。task装饰的方法定义用户行为。一个完整的“Hello World”脚本如下from locust import HttpUser, task, between class QuickstartUser(HttpUser): wait_time between(1, 2.5) host https://httpbin.org # 一个用于测试的公共API task def get_status(self): # self.client是HttpSession用法同requests response self.client.get(/status/200) # 你可以对响应进行断言 if response.status_code ! 200: response.failure(fUnexpected status code: {response.status_code}) task(3) # 这个任务执行频率是get_status的3倍 def get_delay(self): self.client.get(/delay/1) # 模拟一个延迟1秒的接口这个脚本定义了一类用户他们以1到2.5秒的间隔随机执行两个任务一个检查状态码另一个调用一个延迟接口。get_delay任务由于权重为3被选中的概率更高。3.2 模拟真实场景参数化、关联与断言真实的业务场景不会总是请求固定的URL。我们需要参数化、处理Cookie/Session、从上一个请求的响应中提取数据用于下一个请求关联并对响应进行断言。参数化数据可以使用Python的任何方式来准备数据例如列表、从文件读取、从数据库查询等。import csv from locust import HttpUser, task, between class ParameterizedUser(HttpUser): wait_time between(1, 3) host https://your-api.com def on_start(self): 每个虚拟用户开始运行时执行一次用于初始化 self.product_ids [1001, 1002, 1003, 1004] self.current_index 0 task def view_product(self): # 简单的轮询参数化 pid self.product_ids[self.current_index % len(self.product_ids)] self.current_index 1 with self.client.get(f/api/product/{pid}, name/api/product/[id], catch_responseTrue) as response: # catch_responseTrue允许我们手动控制请求的成功/失败 if response.status_code 200 and stock in response.text: response.success() # 可以在这里提取数据例如库存量 # stock response.json().get(stock) else: response.failure(fFailed to get product {pid}) # 更常见的做法是使用队列或随机选择 # from random import choice # pid choice(self.product_ids)处理关联提取Token很多API需要先登录获取token后续请求携带此token。class UserWithAuth(HttpUser): host https://your-api.com wait_time between(2, 5) def on_start(self): 用户登录获取token login_resp self.client.post(/api/login, json{username: test, password: test}) if login_resp.status_code 200: self.token login_resp.json().get(access_token) self.headers {Authorization: fBearer {self.token}} else: self.token None # 如果登录失败可以标记这个用户实例为失败或者停止任务 self.environment.runner.stop_user(self) task def get_profile(self): if not self.token: return # 在请求中带上认证头 self.client.get(/api/profile, headersself.headers)复杂断言与响应校验使用catch_responseTrue上下文管理器可以精细控制请求的成功与失败。task def create_order(self): payload {product_id: 1001, quantity: 2} with self.client.post(/api/order, jsonpayload, catch_responseTrue, name/api/order) as resp: # 校验HTTP状态码 if resp.status_code ! 201: resp.failure(f创建订单失败状态码{resp.status_code}) return # 校验JSON响应体结构 try: data resp.json() if not data.get(order_id): resp.failure(响应中缺少order_id字段) elif data.get(total_price) 0: resp.failure(订单总价异常) else: resp.success() self.order_id data.get(order_id) # 保存供后续使用 except JSONDecodeError: resp.failure(响应不是有效的JSON)3.3 使用WaitTime与Weight控制负载模型负载的“形状”由wait_time和任务权重共同决定。wait_time策略between(min, max)最常用模拟用户随机思考时间。constant(n)每个任务后固定等待n秒适用于节奏固定的轮询场景。constant_pacing(n)更推荐用于稳定性测试。它试图确保每个任务包括执行时间和等待时间的总耗时至少为n秒。如果任务执行很快它就等待剩余时间如果任务执行超过n秒它就不等待立即开始下一个。这能更精确地控制每秒发出的请求数RPS。自定义函数你可以实现任何复杂的等待逻辑。from locust import constant_pacing import random class MyUser(HttpUser): # 确保每个任务周期至少5秒 wait_time constant_pacing(5) # 或者自定义80%的情况等1-3秒20%的情况等10-15秒模拟用户离开 # wait_time lambda self: random.uniform(1, 3) if random.random() 0.2 else random.uniform(10, 15)任务权重与动态任务task装饰器的权重参数是静态的。有时我们需要根据运行时状态动态决定下一个任务。可以通过重写User的tasks属性为一个返回任务列表的方法来实现。from locust import HttpUser, task, between class DynamicUser(HttpUser): wait_time between(1, 3) # 这是一个属性返回一个任务列表 property def tasks(self): task_list [self.view_item] if self.is_vip: # 假设有一个vip标志位 task_list.append(self.vip_discount) return task_list def on_start(self): self.is_vip random.random() 0.8 # 20%的用户是VIP task def view_item(self): self.client.get(/item) def vip_discount(self): self.client.get(/vip-only)4. 执行、监控与结果分析实战4.1 单机与分布式执行命令详解单机无Web UI模式适合CI/CDlocust -f locustfile.py --headless -u 100 -r 10 --run-time 1h30m --hosthttps://your-target.com--headless: 无头模式不启动Web UI。-u 100: 设置总用户数为100。-r 10: 设置孵化速率Hatch rate为每秒10个用户直到达到总用户数。--run-time: 设置测试运行时长格式如1h30m20s。--host: 覆盖脚本中定义的host。输出会在控制台测试结束后会生成摘要。单机启动Web UIlocust -f locustfile.py --hosthttps://your-target.com访问http://localhost:8089即可打开控制界面。在UI中你可以动态设置用户数、孵化速率并点击开始。分布式执行启动Master节点不产生负载locust -f locustfile.py --master --hosthttps://your-target.com在每台Worker机器上启动Worker节点假设Master IP是192.168.1.100locust -f locustfile.py --worker --master-host192.168.1.100打开Master节点的Web UIhttp://192.168.1.100:8089你会看到连接的Worker数量。然后像单机一样启动测试即可。实操心得在分布式执行时确保所有Worker节点上的locustfile.py脚本和Python环境特别是第三方库版本完全一致。否则可能会出现不可预知的行为。建议使用Docker容器或配置管理工具来保证环境一致性。4.2 Web UI界面深度解读Locust的Web UI虽然简洁但信息密度很高。状态标签页显示当前用户数、孵化速率、RPS、失败率以及实时图表。图表包括总RPS、响应时间平均、中位数、P95、P99和用户数。重点关注P95/P99响应时间它们比平均响应时间更能反映尾部延迟对用户体验影响更大。图表响应时间和RPS的趋势图是定位性能问题的关键。如果随着用户数增加响应时间曲线陡然上升而RPS曲线持平甚至下降说明系统已经达到瓶颈。失败率任何非零的失败率都需要警惕。点击“失败”标签页可以查看具体的失败请求和异常信息。下载数据测试结束后务必从“Download Data”选项卡下载CSV报告。包含请求统计、响应时间分位数、异常统计等用于后续详细分析。4.3 关键性能指标KPI与结果分析运行完测试Locust会在控制台输出一个摘要并可以生成HTML报告。你需要学会解读这些数据。Requests/s (RPS)每秒请求数。这是吞吐量的核心指标。在负载增加时观察RPS是否线性增长。当系统达到瓶颈时RPS会达到峰值并开始波动或下降。Response Times (ms)Average平均响应时间参考价值一般。Median (50%)中位数响应时间有一半的请求比它快。95th / 99th percentile (P95/P99)最重要的指标之一。95%的请求响应时间低于此值。它反映了绝大多数用户的体验。P99则反映了最慢的那1%请求的情况对于高要求系统尤其关键。Failure %失败请求的百分比。在性能测试中即使响应慢只要没超时或返回错误码就不算失败。你需要根据业务定义合理的超时时间和断言条件。Number of Users并发用户数。注意Locust中的“用户”是活跃的、正在执行任务的协程。它不等于在线用户数更接近同时发起请求的用户数。如何分析寻找拐点逐步增加负载用户数绘制RPS和P95响应时间随用户数变化的曲线。当P95响应时间开始非线性急剧上升而RPS增长放缓或停滞时这个点就是系统的性能拐点即最大有效吞吐量点。对比基线每次代码发布或配置变更后用相同的负载模型进行测试对比关键指标如P95响应时间、RPS。如果出现显著退化就需要排查。关注错误分析失败请求的日志看是超时、4xx/5xx错误还是自定义断言失败。这往往是系统bug或配置问题的直接体现。5. 高级技巧与生产环境最佳实践5.1 自定义客户端测试非HTTP协议Locust的魔力在于其User类不限于HTTP。你可以继承基类User使用任何Python客户端库。示例使用WebSocket客户端from locust import User, task, between, events import websocket import json import time class WebSocketUser(User): wait_time between(1, 5) def on_start(self): # 建立WebSocket连接 ws_url ws://your-websocket-server/ws self.ws websocket.create_connection(ws_url) # 可以监听连接事件 events.user_started.fire(user_instanceself) def on_stop(self): # 关闭连接 if self.ws: self.ws.close() events.user_stopped.fire(user_instanceself) task def send_message(self): message json.dumps({type: ping, data: int(time.time())}) start_time time.time() try: self.ws.send(message) result self.ws.recv() # 接收响应 response_time int((time.time() - start_time) * 1000) # 毫秒 # 手动触发请求成功事件用于统计 events.request.fire( request_typeWS, nameping, response_timeresponse_time, response_lengthlen(result), exceptionNone, context{} ) # 可以在这里对result进行断言 except Exception as e: response_time int((time.time() - start_time) * 1000) events.request.fire( request_typeWS, nameping, response_timeresponse_time, response_length0, exceptione, context{} )通过手动触发events.request事件你可以将任何协议的请求纳入Locust的统计系统中。同理你可以测试gRPC、TCP、MQTT等任何协议。5.2 利用Events钩子实现复杂逻辑Events系统是Locust的瑞士军刀。除了之前提到的test_start和request还有几个常用钩子init在Locust环境初始化时触发可用于解析命令行参数、注册自定义参数。quitting在Locust进程退出前触发用于清理资源。user_error当用户代码如on_start或task方法发生未处理异常时触发。实战案例在每个Worker上独立准备测试数据在分布式模式下test_start事件只在Master上触发一次。如果每个Worker需要独立准备数据比如从数据库读取一批独立的测试账号可以使用init事件。from locust import events, User import pymysql import random test_accounts [] # 全局变量但每个Worker进程独立 events.init.add_listener def on_locust_init(environment, **kwargs): 在每个进程Master和每个Worker初始化时执行 global test_accounts # 连接数据库读取测试账号 db pymysql.connect(host..., user..., password..., database...) cursor db.cursor() cursor.execute(SELECT username, password FROM test_users WHERE statusactive) test_accounts cursor.fetchall() cursor.close() db.close() print(f[初始化] 进程 {os.getpid()} 加载了 {len(test_accounts)} 个测试账号) class ApiUser(User): def on_start(self): if test_accounts: self.username, self.password random.choice(test_accounts) # 使用这个账号登录...5.3 集成到CI/CD流水线将性能测试左移集成到CI/CD中是保障线上稳定的重要手段。基本集成步骤编写稳定的测试脚本脚本要能独立运行不依赖外部不确定状态。使用--headless模式。定义性能基准确定核心接口的P95响应时间、RPS等合格线。创建CI任务在Jenkins、GitLab CI、GitHub Actions中增加一个性能测试阶段。执行与断言运行Locust命令并解析其输出或生成的报告进行断言。示例使用pytest和locust的插件可以使用pytest来组织和管理Locust测试并利用locust-plugins等第三方库增强功能。# .github/workflows/performance.yml 示例 (GitHub Actions) name: Performance Test on: [push] jobs: load-test: runs-on: ubuntu-latest steps: - uses: actions/checkoutv2 - name: Set up Python uses: actions/setup-pythonv2 with: { python-version: 3.9 } - name: Install dependencies run: pip install locust - name: Run load test run: | locust -f locustfile.py \ --headless \ -u 50 -r 5 \ --run-time 2m \ --host ${{ secrets.TEST_HOST }} \ --csvresults \ --htmlreport.html # 注意这里需要先启动被测试服务或确保服务已就绪 - name: Check performance thresholds run: | # 使用一个Python脚本解析results_stats.csv检查P95是否小于200ms python check_perf.py - name: Upload report uses: actions/upload-artifactv2 with: { name: locust-report, path: report.html }check_perf.py是一个自定义脚本用于解析Locust生成的CSV结果文件并判断是否通过。# check_perf.py import csv import sys def check_performance(): with open(results_stats.csv, r) as f: reader csv.DictReader(f) for row in reader: if row[Name] Aggregated: # 查看汇总行 p95 float(row[95%]) failure_rate float(row[Failure Rate]) print(fAggregated P95: {p95}ms, Failure Rate: {failure_rate*100}%) if p95 200: # 阈值设为200ms print(fERROR: P95响应时间 {p95}ms 超过阈值200ms) sys.exit(1) if failure_rate 0.01: # 失败率阈值1% print(fERROR: 失败率 {failure_rate*100}% 超过阈值1%) sys.exit(1) print(性能测试通过) sys.exit(0) print(未找到聚合数据) sys.exit(1) if __name__ __main__: check_performance()6. 常见问题排查与性能调优指南6.1 Locust自身性能瓶颈与优化有时候压测工具本身会成为瓶颈。以下迹象可能表明是Locust Worker资源不足Worker节点的CPU或内存使用率接近100%。增加Worker节点总RPS却没有线性增长。Master的Web UI出现卡顿或数据更新延迟。优化策略增加Worker节点这是最直接的方法。确保Worker机器有足够的CPU和网络资源。优化测试脚本避免在任务中执行繁重的计算或同步I/O比如频繁读写大文件、进行复杂的字符串处理。这些操作会阻塞协程影响并发能力。将数据预加载到内存中。谨慎使用catch_responseTrue和手动触发事件这些操作有一定开销。对于简单的成功/失败判断直接依赖HTTP状态码更高效。使用更高效的HTTP客户端Locust默认的HttpSession基于requests。对于极高并发可以尝试使用aiohttp作为客户端需要自定义User类。调整Locust配置--expect-workers在Master启动时指定期望的Worker数量Master会等待所有Worker连接后才允许启动测试避免数据不一致。监控Master和Worker的日志查看是否有错误或警告。6.2 被测系统监控与瓶颈定位当Locust报告性能不佳时下一步是定位被测系统的瓶颈。你需要结合系统监控来看。应用服务器指标CPU使用率、内存使用量、线程池状态、GC频率。如果CPU持续高位可能是计算瓶颈或低效代码如果内存持续增长可能有内存泄漏。数据库指标查询QPS、慢查询数量、连接数、锁等待。性能问题常常出在数据库。关注Locust运行期间出现的慢查询日志。中间件指标Redis/Memcached的命中率、Kafka的堆积情况、Nginx的活跃连接数。网络与操作系统指标网络带宽、TCP重传率、磁盘IO等待。一个典型的排查流程现象随着用户数增加响应时间变长RPS上不去。检查Locust Worker资源使用正常排除工具瓶颈。检查应用服务器发现CPU使用率很低但线程池活跃线程数很高很多线程处于BLOCKED或WAITING状态。推断线程在等待某个外部资源可能是数据库连接。检查数据库发现数据库连接池已满大量活跃连接执行缓慢的SQL。定位分析慢查询日志找到没有使用索引的查询或锁竞争激烈的表。解决优化SQL增加索引或调整业务逻辑。6.3 脚本编写中的典型“坑”与避坑指南坑数据竞争导致断言失败或逻辑错误场景多个虚拟用户共享一个全局列表来获取测试数据导致同一数据被多个用户使用。解决为每个虚拟用户实例独立准备数据或在on_start中分配。对于需要全局唯一且消耗性的数据如订单号使用线程安全的队列queue.Queue或分布式锁如果Worker跨机器。坑连接未关闭导致资源泄漏场景在自定义客户端如直接使用requests或数据库连接时没有正确关闭连接。解决在User类的on_stop方法中确保清理资源。对于HTTPLocust的client会自动管理连接池。坑wait_time设置不当导致负载不真实场景使用constant(0)即无等待连续发送请求。这会产生“脉冲式”压力可能瞬间打满系统队列掩盖了系统在稳定流下的真实表现。解决始终设置合理的wait_time如between(1, 3)以模拟真实用户的思考时间。对于稳定性测试使用constant_pacing来控制稳定的RPS。坑断言过于宽松或严格导致失败率失真场景只检查HTTP状态码为200但业务上响应体可能表示失败如{code: 500, msg: error}。解决结合catch_responseTrue对响应体进行充分校验。但同时也要注意不要将一些非核心的字段校验失败定义为请求完全失败这可能会高估系统的功能故障率。可以区分“性能失败”和“业务失败”。坑分布式运行时测试启动时间不一致场景Master点击启动后各个Worker加载脚本、连接Master、接收指令有微小延迟导致初始时刻负载不均衡。解决对于短时测试如1分钟影响较大。可以适当增加--step-load步进加载的步长时间或忽略测试开始前几秒的数据。对于长时间稳定性测试此影响可忽略。我个人在实际使用Locust进行全链路压测时最深的一点体会是脚本的可靠性和可维护性比追求极致的并发数更重要。一个结构清晰、数据隔离良好、断言完备的脚本虽然可能在单机并发上稍微损失一点性能但它能让你在分析结果时充满信心快速定位问题是出在脚本还是被测系统。尤其是在CI/CD中稳定、可重复的测试才是王道。花时间设计好测试数据模型和用户行为流往往比盲目调高并发用户数能发现更多有价值的性能问题。
