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与UserLocust的脚本模型非常直观它模拟的是用户行为而不是单纯的“发请求”。这是它和很多其他工具哲学上的不同。在Locust中核心是两个类HttpUser或其基类User和TaskSet。User类代表一类用户。你可以把它想象成一个“用户模板”。在这个类里你定义这类用户的属性比如他们等待每个任务执行完的间隔时间wait_time以及他们需要执行哪些任务tasks。TaskSet类代表一组任务的集合。一个用户可以在不同场景下执行不同的任务集。例如一个“浏览型用户”的任务集可能只包含查看商品列表和详情的任务而一个“购买型用户”的任务集则包含登录、加购、下单等一系列任务。Task任务用户具体做的动作通常就是一个HTTP接口请求。任务通过Python的task装饰器来定义你可以给任务赋予权重权重越高被随机执行的概率就越大。这种设计让编写贴近真实场景的压测脚本变得非常自然。你不再是在配置一个个孤立的HTTP请求而是在描述“用户会怎么做”。例如一个用户进入APP他可能先浏览权重3然后搜索权重1偶尔才会下单权重0.5。这种权重比例可以直接映射到task装饰器的参数上。2.3 分布式与无状态横向扩展的艺术当需要模拟的用户数超过单台机器的能力通常受限于网络带宽或CPULocust可以轻松地以主从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_responseTrue) as response: # 可以自定义成功/失败的判断逻辑 if response.status_code 200: response.success() else: response.failure(fUnexpected status code: {response.status_code}) task(1) # 权重为1 def view_user_profile(self): # 假设需要传递token在header中 headers {Authorization: fBearer {self.token}} if hasattr(self, token) else {} with self.client.get(/api/user/1, headersheaders, catch_responseTrue) as response: if response.elapsed.total_seconds() 0.5: # 如果响应时间超过500ms标记为失败 response.failure(fResponse too slow: {response.elapsed}) elif response.status_code ! 200: response.failure(fBad status code: {response.status_code}) else: response.success() # 每个User实例停止时执行一次 def on_stop(self): # 可以做清理工作如登出 pass代码关键点解析HttpUser因为我们主要做HTTP测试所以继承这个类最方便。wait_time定义了用户思考时间。between(1,5)是最常用的表示均匀分布。还有constant(3)固定3秒和constant_pacing(2)固定节奏确保任务间隔至少2秒。task核心装饰器。权重比是3:1意味着在长时间运行中view_homepage被调用的次数大约是view_user_profile的3倍。self.client这是发起请求的核心对象。它支持getpostputdelete等方法接口设计和requests库高度一致学习成本为零。catch_responseTrue这个上下文管理器允许你更精细地控制请求的成功与失败。默认情况下HTTP状态码非2xx/3xx会被标记为失败。但有时业务上200返回错误码也算失败或者5xx但你希望忽略都可以在这里处理。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(forceTrue) return with self.client.get(f/api/search?q{keyword}, name/api/search, catch_responseTrue) 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_responseTrue) 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, jsonlogin_payload, name01_Login) as resp: if resp.status_code 200: self.auth_token resp.json()[data][token] self.headers {Authorization: fBearer {self.auth_token}} else: # 如果登录失败记录异常并停止这个用户 resp.failure(Login failed) self.stop(forceTrue) # 强制停止该用户实例 task(5) def get_profile(self): # 使用保存在实例中的headers self.client.get(/api/profile, headersself.headers, name02_GetProfile) task(1) def create_order(self): order_data {product_id: 123, quantity: 1} with self.client.post(/api/order, jsonorder_data, headersself.headers, name03_CreateOrder) as resp: if resp.status_code 201: self.order_id resp.json()[order_id] else: resp.failure(fCreate order failed: {resp.text}) # 可以基于上一个请求的结果发起新请求 task def view_my_order(self): if hasattr(self, order_id): self.client.get(f/api/order/{self.order_id}, headersself.headers, name04_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%概率进入BrowseTasks50%进入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 --hosthttp://your-server.com # 更常用的指定运行时间并导出结果文件 locust -f locustfile.py \ --headless \ --users 500 \ --spawn-rate 20 \ --run-time 5m \ --csvresults/my_test \ # 生成CSV报告的前缀 --csv-full-history \ # 记录整个运行期间的历史数据 --hosthttp://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 --hosthttp://your-server.com步骤2启动一个或多个Worker节点Worker节点需要能访问到locustfile.py和Master节点默认端口8089。# 在另一台机器或另一个终端 locust -f locustfile.py --worker --master-hostMASTER_IP_ADDRESS你可以启动任意多个Worker它们会自动向Master注册。步骤3在Web UI或命令行控制压测此时访问Master节点的Web UIhttp://MASTER_IP:8089启动压测。你会发现你可以设置的总用户数上限变大了因为资源由所有Worker共同提供。踩坑记录分布式运行时确保所有Worker节点上的locustfile.py是完全一致的特别是共享数据源如队列的初始化逻辑。否则可能出现不可预知的行为。建议将压测脚本放在版本库中在所有节点统一拉取。5.3 Locust脚本性能调优即使有gevent加持写得不好的Locust脚本也可能成为瓶颈。以下是一些优化点避免在任务函数中进行阻塞操作或繁重计算如前所述这会阻塞整个事件循环。如果必须做使用gevent.spawn或gevent.pool将其放到单独的greenlet中执行。谨慎使用time.sleep虽然gevent将其替换为非阻塞的但大量虚拟用户频繁sleep也会消耗资源。对于精确的节奏控制考虑使用constant_pacing。优化请求验证逻辑catch_responseTrue和复杂的响应体解析如解析大的JSON会有开销。在高压下如果只关心请求是否成功发出可以简化验证逻辑甚至对某些接口关闭catch_response。使用连接池requests.Sessionself.client底层使用会自动保持HTTP连接复用。确保你没有在每个请求中创建新的Session。监控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脚本调试技巧打印日志在脚本中使用Python标准的logging模块或直接print。注意在分布式模式下日志会输出在各自Worker的终端。可以通过--logfile参数将日志重定向到文件。import logging logger logging.getLogger(__name__) task def my_task(self): logger.info(Starting a task) # ... 请求 logger.debug(fResponse time: {response.elapsed})运行Locust时可以通过--loglevel DEBUG来输出更详细的日志。使用catch_response进行断言调试这是最强大的调试手段之一。你可以在with块内对响应内容做任何检查并灵活地标记成功或失败。with self.client.post(/api, jsondata, catch_responseTrue) as resp: # 检查状态码 if resp.status_code ! 200: resp.failure(fStatus code error: {resp.status_code}) return # 检查JSON结构 try: json_data resp.json() if json_data.get(code) ! 0: resp.failure(fBusiness logic error: {json_data.get(msg)}) else: resp.success() except json.JSONDecodeError: resp.failure(Response is not valid JSON)先以单个用户、慢速模式运行使用--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 \ --csvperf_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(fTotal orders simulated: {len(order_amounts)}) print(fTotal transaction volume: {sum(order_amounts)})集成到监控系统你可以将Locust的实时统计数据通过events.report_to_master或events.worker_report发送到你的监控系统如Prometheus实现压测数据的可视化大盘。经过这些年的实战Locust对我来说已经不仅仅是一个压测工具更像是一个用代码定义“用户行为模型”的仿真平台。它的学习曲线平缓但上限极高。从简单的API压测到复杂的全链路业务场景模拟从单机运行到千级节点的分布式集群它都能很好地胜任。最关键的是它让性能测试脚本化、工程化成为了开发流程中自然的一环。下次当你需要评估系统容量或寻找性能瓶颈时不妨从写一个简单的locustfile.py开始。
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与UserLocust的脚本模型非常直观它模拟的是用户行为而不是单纯的“发请求”。这是它和很多其他工具哲学上的不同。在Locust中核心是两个类HttpUser或其基类User和TaskSet。User类代表一类用户。你可以把它想象成一个“用户模板”。在这个类里你定义这类用户的属性比如他们等待每个任务执行完的间隔时间wait_time以及他们需要执行哪些任务tasks。TaskSet类代表一组任务的集合。一个用户可以在不同场景下执行不同的任务集。例如一个“浏览型用户”的任务集可能只包含查看商品列表和详情的任务而一个“购买型用户”的任务集则包含登录、加购、下单等一系列任务。Task任务用户具体做的动作通常就是一个HTTP接口请求。任务通过Python的task装饰器来定义你可以给任务赋予权重权重越高被随机执行的概率就越大。这种设计让编写贴近真实场景的压测脚本变得非常自然。你不再是在配置一个个孤立的HTTP请求而是在描述“用户会怎么做”。例如一个用户进入APP他可能先浏览权重3然后搜索权重1偶尔才会下单权重0.5。这种权重比例可以直接映射到task装饰器的参数上。2.3 分布式与无状态横向扩展的艺术当需要模拟的用户数超过单台机器的能力通常受限于网络带宽或CPULocust可以轻松地以主从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_responseTrue) as response: # 可以自定义成功/失败的判断逻辑 if response.status_code 200: response.success() else: response.failure(fUnexpected status code: {response.status_code}) task(1) # 权重为1 def view_user_profile(self): # 假设需要传递token在header中 headers {Authorization: fBearer {self.token}} if hasattr(self, token) else {} with self.client.get(/api/user/1, headersheaders, catch_responseTrue) as response: if response.elapsed.total_seconds() 0.5: # 如果响应时间超过500ms标记为失败 response.failure(fResponse too slow: {response.elapsed}) elif response.status_code ! 200: response.failure(fBad status code: {response.status_code}) else: response.success() # 每个User实例停止时执行一次 def on_stop(self): # 可以做清理工作如登出 pass代码关键点解析HttpUser因为我们主要做HTTP测试所以继承这个类最方便。wait_time定义了用户思考时间。between(1,5)是最常用的表示均匀分布。还有constant(3)固定3秒和constant_pacing(2)固定节奏确保任务间隔至少2秒。task核心装饰器。权重比是3:1意味着在长时间运行中view_homepage被调用的次数大约是view_user_profile的3倍。self.client这是发起请求的核心对象。它支持getpostputdelete等方法接口设计和requests库高度一致学习成本为零。catch_responseTrue这个上下文管理器允许你更精细地控制请求的成功与失败。默认情况下HTTP状态码非2xx/3xx会被标记为失败。但有时业务上200返回错误码也算失败或者5xx但你希望忽略都可以在这里处理。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(forceTrue) return with self.client.get(f/api/search?q{keyword}, name/api/search, catch_responseTrue) 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_responseTrue) 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, jsonlogin_payload, name01_Login) as resp: if resp.status_code 200: self.auth_token resp.json()[data][token] self.headers {Authorization: fBearer {self.auth_token}} else: # 如果登录失败记录异常并停止这个用户 resp.failure(Login failed) self.stop(forceTrue) # 强制停止该用户实例 task(5) def get_profile(self): # 使用保存在实例中的headers self.client.get(/api/profile, headersself.headers, name02_GetProfile) task(1) def create_order(self): order_data {product_id: 123, quantity: 1} with self.client.post(/api/order, jsonorder_data, headersself.headers, name03_CreateOrder) as resp: if resp.status_code 201: self.order_id resp.json()[order_id] else: resp.failure(fCreate order failed: {resp.text}) # 可以基于上一个请求的结果发起新请求 task def view_my_order(self): if hasattr(self, order_id): self.client.get(f/api/order/{self.order_id}, headersself.headers, name04_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%概率进入BrowseTasks50%进入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 --hosthttp://your-server.com # 更常用的指定运行时间并导出结果文件 locust -f locustfile.py \ --headless \ --users 500 \ --spawn-rate 20 \ --run-time 5m \ --csvresults/my_test \ # 生成CSV报告的前缀 --csv-full-history \ # 记录整个运行期间的历史数据 --hosthttp://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 --hosthttp://your-server.com步骤2启动一个或多个Worker节点Worker节点需要能访问到locustfile.py和Master节点默认端口8089。# 在另一台机器或另一个终端 locust -f locustfile.py --worker --master-hostMASTER_IP_ADDRESS你可以启动任意多个Worker它们会自动向Master注册。步骤3在Web UI或命令行控制压测此时访问Master节点的Web UIhttp://MASTER_IP:8089启动压测。你会发现你可以设置的总用户数上限变大了因为资源由所有Worker共同提供。踩坑记录分布式运行时确保所有Worker节点上的locustfile.py是完全一致的特别是共享数据源如队列的初始化逻辑。否则可能出现不可预知的行为。建议将压测脚本放在版本库中在所有节点统一拉取。5.3 Locust脚本性能调优即使有gevent加持写得不好的Locust脚本也可能成为瓶颈。以下是一些优化点避免在任务函数中进行阻塞操作或繁重计算如前所述这会阻塞整个事件循环。如果必须做使用gevent.spawn或gevent.pool将其放到单独的greenlet中执行。谨慎使用time.sleep虽然gevent将其替换为非阻塞的但大量虚拟用户频繁sleep也会消耗资源。对于精确的节奏控制考虑使用constant_pacing。优化请求验证逻辑catch_responseTrue和复杂的响应体解析如解析大的JSON会有开销。在高压下如果只关心请求是否成功发出可以简化验证逻辑甚至对某些接口关闭catch_response。使用连接池requests.Sessionself.client底层使用会自动保持HTTP连接复用。确保你没有在每个请求中创建新的Session。监控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脚本调试技巧打印日志在脚本中使用Python标准的logging模块或直接print。注意在分布式模式下日志会输出在各自Worker的终端。可以通过--logfile参数将日志重定向到文件。import logging logger logging.getLogger(__name__) task def my_task(self): logger.info(Starting a task) # ... 请求 logger.debug(fResponse time: {response.elapsed})运行Locust时可以通过--loglevel DEBUG来输出更详细的日志。使用catch_response进行断言调试这是最强大的调试手段之一。你可以在with块内对响应内容做任何检查并灵活地标记成功或失败。with self.client.post(/api, jsondata, catch_responseTrue) as resp: # 检查状态码 if resp.status_code ! 200: resp.failure(fStatus code error: {resp.status_code}) return # 检查JSON结构 try: json_data resp.json() if json_data.get(code) ! 0: resp.failure(fBusiness logic error: {json_data.get(msg)}) else: resp.success() except json.JSONDecodeError: resp.failure(Response is not valid JSON)先以单个用户、慢速模式运行使用--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 \ --csvperf_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(fTotal orders simulated: {len(order_amounts)}) print(fTotal transaction volume: {sum(order_amounts)})集成到监控系统你可以将Locust的实时统计数据通过events.report_to_master或events.worker_report发送到你的监控系统如Prometheus实现压测数据的可视化大盘。经过这些年的实战Locust对我来说已经不仅仅是一个压测工具更像是一个用代码定义“用户行为模型”的仿真平台。它的学习曲线平缓但上限极高。从简单的API压测到复杂的全链路业务场景模拟从单机运行到千级节点的分布式集群它都能很好地胜任。最关键的是它让性能测试脚本化、工程化成为了开发流程中自然的一环。下次当你需要评估系统容量或寻找性能瓶颈时不妨从写一个简单的locustfile.py开始。
