1. 项目概述为什么是Locust如果你正在寻找一个能模拟成千上万用户、用代码定义用户行为、并且能让你完全掌控测试逻辑的性能测试工具那么Locust大概率会进入你的视野。它不是JMeter那样的“点击式”工具而是一个用Python代码编写测试脚本的开源框架。这意味着如果你或你的团队熟悉PythonLocust的学习曲线会异常平滑你获得的灵活性和扩展性也是传统工具难以比拟的。我最初接触性能测试时也用过一些录制回放的工具但很快就遇到了瓶颈复杂的业务逻辑、动态参数、条件判断、数据驱动这些在图形界面里配置起来要么极其繁琐要么根本无法实现。Locust的出现解决了这个痛点。它把测试场景的定义权完全交给了代码。你想让用户先登录、再搜索商品、然后随机将其中几件加入购物车最后只有30%的用户去结算用Python的if-else、for循环和随机函数你可以像写业务代码一样自然地描述出这个场景。这种“代码即场景”的理念是Locust最核心的吸引力。另一个关键优势在于它的架构。Locust采用主从Master-Worker模式一个Master节点负责协调测试、收集数据多个Worker节点可以分布在不同的机器上负责真正执行虚拟用户Locust并施加负载。这种分布式设计让你能轻松地发起远超单机能力的并发压力。而且它自带一个简洁的Web UI可以实时查看RPS每秒请求数、响应时间、错误率等关键指标并动态调整并发用户数这比盯着命令行输出要直观太多了。简单来说Locust适合这样的你有一定Python基础不满足于黑盒工具希望测试脚本能像产品代码一样可维护、可版本控制并且需要应对复杂、贴近真实用户行为的性能测试场景。接下来我们就从零开始把它用起来。2. 环境准备与Locust安装工欲善其事必先利其器。Locust的安装非常简单但一个清晰、独立的Python环境是后续一切顺利的基础。我强烈建议不要使用系统自带的Python而是使用虚拟环境。这能避免项目间的依赖冲突也是Python开发的最佳实践。2.1 创建独立的Python虚拟环境虚拟环境就像一个沙箱你在里面安装的所有包都不会影响到系统或其他项目。这里我推荐使用venv它是Python 3.3自带的模块无需额外安装。首先打开你的终端Windows用CMD或PowerShellmacOS/Linux用Terminal创建一个专门用于本项目的目录并进入mkdir locust_project cd locust_project然后在这个目录下创建虚拟环境。环境文件夹的名字通常叫venv或.venv# Windows python -m venv venv # macOS/Linux python3 -m venv venv创建完成后激活这个虚拟环境# Windows (CMD) venv\Scripts\activate.bat # Windows (PowerShell) venv\Scripts\Activate.ps1 # macOS/Linux source venv/bin/activate激活后你的命令行提示符前通常会显示(venv)这表明你已经进入了这个独立的Python环境。后续所有pip安装操作都只在这个环境内生效。注意如果你在Windows PowerShell上执行激活脚本时遇到“禁止运行脚本”的错误需要以管理员身份打开PowerShell先执行Set-ExecutionPolicy RemoteSigned更改执行策略。这是一个常见的坑。2.2 安装Locust及其依赖环境激活后安装Locust就一行命令pip install locust这条命令会安装Locust核心框架以及它依赖的库比如用于HTTP请求的geventhttpclient用于Web UI的Flask等。安装完成后可以通过以下命令验证是否成功locust --version如果正确输出版本号如locust 2.20.0说明安装成功。2.3 可选配置IDE以VS Code为例虽然用任何文本编辑器都能写Locust脚本但一个好的IDE能极大提升效率。VS Code对Python的支持非常出色。安装Python扩展在VS Code的扩展商店搜索并安装“Python”扩展由Microsoft发布。选择解释器用VS Code打开你的locust_project文件夹。按CtrlShiftPWindows/Linux或CmdShiftPmacOS打开命令面板输入“Python: Select Interpreter”然后选择路径中包含locust_project\venv或locust_project/.venv的Python解释器。这确保了VS Code使用我们刚创建的虚拟环境。安装代码格式化工具推荐在VS Code的终端确保已激活venv里运行pip install black。然后在VS Code设置中将“Format On Save”打开并设置默认格式化工具为Black。这能让你的代码保持整洁统一的风格。至此你的编码环境已经就绪。我们即将开始编写第一个性能测试脚本。3. 核心概念与第一个测试脚本在动手写代码前必须理解Locust里的三个核心概念这构成了所有测试脚本的骨架。HttpUser或User类 这是虚拟用户的蓝图。你定义的每个用户类都代表一类用户行为。HttpUser是一个最常用的内置类它提供了方便的client属性一个HttpSession实例来发送HTTP请求。TaskSet 任务集。用于将多个任务Task组织在一起可以嵌套模拟复杂的用户操作流程。在Locust 2.x版本中更推荐使用task装饰器直接在User类中定义任务但对于有顺序或权重的任务组TaskSet依然有用。task装饰器 用来标记一个方法是用户要执行的任务。你可以通过task(权重)来设置不同任务被执行的频率比例。下面我们用一个最经典的例子——“模拟用户浏览网站”——来把这些概念串起来。假设我们要测试一个博客网站用户行为是先访问首页然后有70%的概率去查看文章列表30%的概率直接搜索。创建一个名为locustfile.py的文件这是Locust默认寻找的入口脚本并写入以下内容from locust import HttpUser, task, between class WebsiteUser(HttpUser): # wait_time 定义了用户在每个任务执行后的等待时间 # between(5, 15) 表示等待5到15秒之间的一个随机数更贴近真实用户 wait_time between(5, 15) # 权重为3意味着在所有任务中执行此任务的相对概率更高 task(3) def view_article_list(self): # self.client 是HttpSession实例用法和requests库非常相似 # 这里模拟GET请求访问文章列表页 with self.client.get(/articles, catch_responseTrue) as response: # 你可以对响应进行断言判断请求是否“真正”成功 if response.status_code 200 and 文章列表 in response.text: response.success() else: response.failure(fFailed! Status: {response.status_code}) task(1) def search_article(self): # 模拟一个带查询参数的搜索请求 self.client.get(/search?keywordlocust) # on_start 方法会在每个虚拟用户开始运行时自动调用一次 # 常用于模拟登录等前置操作 def on_start(self): # 这里假设登录是一个POST请求返回一个认证token这里简化处理 login_response self.client.post(/login, json{username: test, password: test}) if login_response.status_code 200: print(fUser {self.id} logged in successfully.) # 通常可以将token保存在self.client.headers中供后续请求使用 # self.client.headers[Authorization] fBearer {login_response.json()[token]} else: print(fUser {self.id} login failed!)这个脚本虽然简单但包含了Locust脚本的所有关键要素用户类WebsiteUser继承自HttpUser。任务view_article_list和search_article两个方法被task装饰成为了用户可能执行的任务。权重3:1意味着在长时间运行中访问列表页的次数大约是搜索的3倍。等待时间wait_time between(5, 15)让虚拟用户在任务间“思考”避免产生不切实际的、毫秒级间隔的轰炸式请求。生命周期钩子on_start用于用户启动时的初始化如登录。响应验证在view_article_list任务中我们使用了catch_responseTrue上下文管理器并不仅依赖HTTP状态码还检查了响应体内容以此更精确地判断业务成功与否。这是生产级脚本的必备操作。实操心得catch_responseTrue和手动调用response.success()/failure()是Locust脚本健壮性的关键。很多接口可能返回200状态码但业务逻辑是失败的如“系统繁忙”。只有加入了业务断言你的性能测试结果失败率才具有真正的参考价值。4. 运行测试与Web UI解读脚本写好了让我们来运行它看看Locust的威力。4.1 启动Locust Master在终端中确保位于locustfile.py所在的目录并且虚拟环境已激活然后运行locust默认情况下Locust会使用当前目录下的locustfile.py作为测试脚本并启动Web UI在http://localhost:8089。如果你想指定其他主机或端口可以使用locust --hosthttps://your-test-api.com --web-host0.0.0.0 --web-port8089--host指定被测试系统的基地址。我们脚本中的self.client.get(/articles)会拼接成https://your-test-api.com/articles。--web-host0.0.0.0允许其他机器通过IP访问Web UI方便远程监控。--web-port指定Web UI端口。启动后打开浏览器访问http://localhost:8089你会看到Locust的启动界面。4.2 Web UI配置与启动Web UI界面非常直观Number of users要模拟的总用户数峰值。Spawn rate每秒启动多少个用户直到达到总用户数。例如用户数1000生成速率10意味着Locust会用100秒逐步启动所有用户而不是瞬间启动这有助于观察系统在压力逐步增大时的表现。Host这里可以再次覆盖被测试系统的地址。填写好这些参数点击“Start swarming”测试就正式开始了。界面会自动跳转到数据统计页。4.3 核心监控指标解读测试运行后Web UI会实时刷新数据。你需要重点关注以下几个标签页和指标Statistics统计这是最重要的标签页。它以表格形式展示了每个请求按URL和Method区分的性能数据。Type/Name: 请求的路径和方式。Requests: 总请求数。Fails: 失败请求数即你调用过response.failure()的请求。Median, 90%, 95%, 99%: 响应时间的百分位数。重点关注90%或95%分位值它表示有90%/95%的请求响应时间低于这个值。这比平均响应时间更能反映大多数用户的体验。Average: 平均响应时间。Min/Max: 最小/最大响应时间。Average size: 平均响应大小。Current RPS: 当前每秒请求数。这是系统吞吐量的直接体现。Current Failures/s: 当前每秒失败数。Charts图表以曲线图形式展示总RPS、响应时间、用户数随时间的变化。非常直观可以用来定位性能拐点。比如当用户数达到某个值时响应时间曲线突然陡增这里可能就是系统的瓶颈点。Failures失败列出所有失败的请求包括URL、错误信息和发生时间。这是调试脚本和排查系统问题的主要依据。Exceptions异常显示测试运行过程中Python代码抛出的异常。Download Data可以将统计数据以CSV格式下载下来用于后续更细致的分析或报告生成。注意事项Locust的Web UI在测试数据量极大如运行数小时、产生数亿请求时可能会占用较多浏览器内存。对于长时间的压力测试建议定期下载数据并考虑使用--headless模式运行或使用Locust提供的时序数据库如InfluxDB集成方案将数据导出到更专业的监控工具如Grafana中进行观察。5. 编写高级且健壮的测试脚本基础脚本只能应对简单场景。真实的业务往往涉及状态保持、参数化、关联和复杂的业务流程。下面我们来提升脚本的复杂度。5.1 参数化与测试数据管理让所有虚拟用户使用同一组数据如同一个用户名登录会导致缓存命中率畸高无法反映真实压力也容易触发系统的防重放机制。我们需要参数化。方法一从列表中随机选取from locust import HttpUser, task, between import random class SearchUser(HttpUser): wait_time between(2, 5) # 准备一个搜索关键词池 search_keywords [python, 性能测试, locust, docker, kubernetes, 大数据] task def search(self): keyword random.choice(self.search_keywords) with self.client.get(f/search?q{keyword}, name/search?q[keyword], catch_responseTrue) as response: # 注意这里使用了name参数。Locust的统计是按name聚合的。 # 如果不加name每个不同的keyword都会产生一条独立的统计记录导致图表杂乱。 # 使用name将其归一化为“/search?q[keyword]”这一项。 if response.status_code 200: response.success() else: response.failure(fSearch failed for keyword: {keyword})方法二从文件中循环读取更适用于大量数据假设我们有一个user_credentials.csv文件username,password user1,pass1 user2,pass2 ... ...import csv from locust import HttpUser, task, between from itertools import cycle class LoginUser(HttpUser): wait_time between(1, 3) # 在类加载时读取CSV文件并创建一个无限循环的迭代器 with open(user_credentials.csv, r) as f: reader csv.DictReader(f) credentials_pool cycle(list(reader)) # cycle使其可循环迭代 def on_start(self): # 每个用户启动时从池子里取一组凭证 self.credential next(self.credentials_pool) task def visit_profile(self): # 使用专属的凭证信息来访问个性化页面 self.client.get(f/profile?username{self.credential[username]}, name/profile)实操心得对于性能测试数据池的大小要足够。如果只有10组数据却模拟1000个用户很快就会出现数据重复压力模型就不准确了。通常建议数据池容量至少是并发用户数的5-10倍。5.2 处理关联与状态保持现代Web应用大量使用Session、Token或Cookie来保持用户状态。Locust的self.clientHttpSession会自动处理Cookie就像浏览器一样。对于Token我们需要手动管理。from locust import HttpUser, task, between class ApiUser(HttpUser): wait_time between(1, 5) host https://api.example.com def on_start(self): # 1. 登录获取token login_resp self.client.post(/auth/login, json{email: testlocust.io, password: test123}) if login_resp.status_code 200: self.token login_resp.json()[access_token] # 2. 将token设置到后续所有请求的Header中 self.client.headers.update({Authorization: fBearer {self.token}}) print(fUser {self.id} got token.) else: # 如果登录失败可以标记此用户为停止状态或者记录失败 self.stop(forceTrue) # 强制停止这个用户实例 task def get_private_data(self): # 此时的请求会自动带上Authorization Header self.client.get(/user/private-data) task(2) # 这个任务权重更高 def update_profile(self): new_name fLocustUser_{self.id} self.client.put(/user/profile, json{displayName: new_name})5.3 模拟复杂的用户行为序列有时用户的操作是有固定顺序的比如“添加商品到购物车 - 查看购物车 - 结算”。我们可以用SequentialTaskSet来实现。from locust import HttpUser, task, between from locust import SequentialTaskSet class ShoppingCartUser(HttpUser): wait_time between(5, 10) host https://shop.example.com # tasks 属性可以指向一个TaskSet类 tasks [ShoppingCartTaskSet] class ShoppingCartTaskSet(SequentialTaskSet): # SequentialTaskSet中的task会按定义顺序依次执行 # 每个task执行完后会等待User类中定义的wait_time task def add_item_to_cart(self): item_id prod_123 self.client.post(f/cart/add/{item_id}, name/cart/add/[item_id]) # 这里可以解析响应获取购物车ID等存入self.user即父类HttpUser实例中共享 # self.user.cart_id response.json()[cartId] task def view_cart(self): self.client.get(/cart) task def checkout(self): # 假设结算需要购物车ID可以从self.user中获取 # payload {cartId: self.user.cart_id} payload {cartId: temp_cart_id} with self.client.post(/order/checkout, jsonpayload, catch_responseTrue) as resp: if resp.status_code 200: resp.success() self.interrupt() # 关键执行完结算后跳出这个SequentialTaskSet # 然后会回到User类的task选择逻辑可能再次进入这个TaskSet开始新一轮“加购-结算” else: resp.failure(Checkout failed)关键点SequentialTaskSet会循环执行其内部任务。在上例中如果不加self.interrupt()用户会永远在add_item_to_cart - view_cart - checkout这个序列里循环。self.interrupt()的作用是强制当前用户跳出这个顺序任务集回到父类ShoppingCartUser的tasks列表中重新选择任务执行这里只有一个ShoppingCartTaskSet所以又会重新进入。这样就模拟了用户完成一次购物流程后可能开始下一次流程的行为。6. 分布式压测与提升负载能力单机运行的Locust能模拟的用户数受限于本机的CPU、内存和网络端口数。要产生更大的压力必须使用分布式模式。6.1 分布式架构原理Locust采用主从Master-Worker架构Master节点负责管理测试、分发任务、收集并汇总所有Worker节点的数据、提供Web UI。它本身不模拟用户。Worker节点负责执行locustfile.py脚本模拟虚拟用户生成真正的请求负载。可以启动多个Worker并且可以分布在不同的物理机或虚拟机上。6.2 启动分布式集群假设你有三台机器master_host(192.168.1.100),worker1_host(192.168.1.101),worker2_host(192.168.1.102)。步骤1启动Master节点在master_host机器上进入项目目录运行locust --master --hosthttps://your-test-api.com --web-host0.0.0.0--master参数指定此实例为Master。--web-host0.0.0.0让Web UI能被其他机器访问。步骤2启动Worker节点在worker1_host和worker2_host机器上需要拥有相同的locustfile.py脚本和Python环境。然后分别运行locust --worker --master-host192.168.1.100--worker指定为Worker节点--master-host指向Master节点的IP地址。步骤3在Web UI中控制测试现在访问http://192.168.1.100:8089你会看到Web UI。在界面底部会显示“2 workers connected”表示有两个Worker已就绪。此时你配置的用户数如10000会被自动分配到两个Worker上执行。6.3 容器化部署使用Docker手动在多台机器上配置环境很麻烦。使用Docker可以极大简化部署。Locust官方提供了Docker镜像。使用Docker Compose一键部署集群创建一个docker-compose.yml文件version: 3 services: master: image: locustio/locust ports: - 8089:8089 - 5557:5557 volumes: - ./:/mnt/locust # 将本地的locustfile.py挂载到容器内 command: -f /mnt/locust/locustfile.py --master --hosthttps://your-test-api.com worker: image: locustio/locust volumes: - ./:/mnt/locust command: -f /mnt/locust/locustfile.py --worker --master-hostmaster deploy: replicas: 4 # 启动4个worker容器实例然后在本机运行docker-compose up -d就会启动1个Master容器和4个Worker容器并自动连接。你只需访问本机的8089端口即可管理测试。避坑技巧网络与防火墙确保Master节点的5557Worker通信端口和8089Web UI端口在防火墙中是开放的并且Worker节点能访问到Master节点的IP。时间同步分布式环境下确保所有机器或容器的时间基本同步否则日志时间戳会混乱。可以使用NTP服务。数据一致性确保所有Worker节点上的测试数据文件如CSV内容一致或者通过共享存储如NFS访问同一份文件。资源监控压测时不仅要监控被测系统也要监控Master和Worker节点本身的CPU、内存、网络IO。如果Worker节点资源耗尽会成为压测瓶颈产生“压不上去”的假象。可以使用htop,nload等工具。7. 集成CI/CD与自动化测试性能测试不应该只是手动执行的“活动”而应该成为持续交付流水线中的一个自动化环节。Locust可以轻松集成到Jenkins、GitLab CI、GitHub Actions等CI/CD平台中。7.1 无头模式Headless运行与结果导出Locust支持--headless模式即不启动Web UI在命令行中直接运行测试并结束后退出非常适合自动化。locust --headless --users 100 --spawn-rate 10 --run-time 30s --hosthttps://your-test-api.com--headless启用无头模式。--users总用户数。--spawn-rate生成速率。--run-time测试运行时间例如30s,5m,1h30m。时间到后测试自动停止。--csvprefix以CSV格式导出结果会生成多个文件如prefix_stats.csv,prefix_failures.csv等。7.2 在GitHub Actions中集成示例下面是一个简单的GitHub Actions工作流示例它在每次推送到main分支时自动运行Locust性能测试并将结果归档。# .github/workflows/locust.yml name: Performance Test with Locust on: push: branches: [ main ] jobs: performance-test: runs-on: ubuntu-latest steps: - name: Checkout code uses: actions/checkoutv3 - name: Set up Python uses: actions/setup-pythonv4 with: python-version: 3.9 - name: Install dependencies run: | python -m pip install --upgrade pip pip install locust - name: Run Locust performance test run: | # 假设你的测试需要先启动一个本地测试服务如用docker-compose up启动被测系统 # 这里我们直接对某个外部服务进行测试仅为示例实际需替换HOST locust --headless \ --users 50 \ --spawn-rate 5 \ --run-time 1m \ --host${{ secrets.TEST_HOST }} \ --csvresults/locust_results \ --htmlresults/locust_report.html \ --loglevel INFO continue-on-error: true # 即使测试失败也继续执行后续步骤以保存结果 - name: Upload test results uses: actions/upload-artifactv3 with: name: locust-reports path: results/ retention-days: 7在这个工作流中我们安装了Python和Locust。以无头模式运行Locust模拟50个用户在1分钟内完成测试。将结果输出为CSV和HTML报告。最后将生成的结果文件上传到GitHub Actions的Artifacts中供后续下载分析。注意事项在CI中运行性能测试通常是对一个独立的测试环境如Staging环境进行。要确保该环境的稳定性和数据隔离。另外CI环境可能资源有限模拟的用户数不宜过大重点在于回归验证核心接口的性能是否出现显著退化而非进行极限压测。8. 常见问题排查与性能调优心得在实际使用Locust的过程中你肯定会遇到各种问题。这里记录了一些典型问题和我的排查思路。8.1 Locust自身问题排查现象可能原因排查与解决“Address already in use”8089或5557端口被占用。使用locust --web-port8090指定其他端口。或找出占用进程lsof -i:8089(macOS/Linux) /netstat -ano | findstr :8089(Windows)。Worker无法连接Master防火墙阻止、Master主机名/IP错误、网络不通。1. 检查Master节点防火墙规则。2. 确保Worker命令中的--master-host正确。3. 从Worker节点ping/telnet Master节点的5557端口。Web UI无数据或数据不动Master与Worker网络通信问题测试脚本有错误导致Worker崩溃。1. 检查Master日志和Worker日志启动时加--logfilelocust.log。2. 在Web UI的“Workers”标签页查看Worker状态是否正常。3. 简化脚本逐步排查。模拟用户数达不到设定值单机资源CPU、内存、可用端口耗尽。1. 使用top或htop查看Locust进程资源使用情况。2. 增加--worker数量分布式。3. 优化脚本减少单个用户的内存占用如避免在任务中加载大文件到内存。响应时间异常增长RPS上不去被测系统达到瓶颈或Locust Worker成为瓶颈。1. 监控被测系统的资源CPU、内存、磁盘IO、网络带宽、数据库连接数。2. 同时监控Locust Worker机器的资源。如果Worker CPU已满则需要增加Worker节点。8.2 测试脚本与性能调优心得关于wait_timewait_time between(1, 3)和wait_time constant(1)产生的压力模型天差地别。前者更贴近真实用户有思考时间后者是持续不断的请求。选择哪种取决于你的测试目标是考察系统在真实用户模型下的表现还是考察系统的纯接口吞吐量极限。大多数场景下使用随机等待时间更合理。关于网络连接Locust默认对每个请求使用短连接HTTP/1.1。对于高并发场景建立TCP连接的开销很大。启用HTTP Keep-Alive可以大幅提升效率。幸运的是Locust底层使用的geventhttpclient默认就支持Keep-Alive。你只需要确保你的locustfile.py中使用的self.client是同一个会话对象默认就是它就会自动复用连接。你可以通过打印self.client的id来验证。减少不必要的日志默认的INFO日志会打印每个请求在高压下会产生巨量IO影响性能。在生产压测时可以使用--loglevel WARNING或--loglevel ERROR来减少日志输出。小心Python的GILLocust基于gevent一个协程库通过异步IO来实现高并发。但如果你在任务中执行了大量的CPU密集型计算如复杂的图像处理、数据加密这会阻塞整个线程严重影响并发能力。对于CPU密集型操作应考虑将其移到被测服务端或者使用Locust的fast_http客户端已废弃或探索其他客户端模式。通常性能测试脚本的逻辑应尽量轻量只负责发请求和做简单断言。断言要快但要准catch_responseTrue和响应内容解析如response.json()会有性能开销。如果只是为了检查状态码直接用response.status_code 200判断即可。如果需要检查响应体尽量使用简单的字符串查找如in操作而非完整的JSON解析除非必要。在超高压测试下这些开销累积起来也不容忽视。分布式数据共享难题如果测试需要全局唯一ID或共享计数器例如模拟用户注册用户名不能重复简单的cycle迭代文件在分布式下会出问题多个Worker会读取相同的数据段。解决方案有a) 使用中央存储如Redis生成IDb) 为每个Worker预分配不同的数据段c) 使用更复杂的分布式任务队列。这通常是编写高级Locust脚本时最复杂的部分。最后性能测试的本质不是“跑工具”而是“做实验”和“观察分析”。Locust给了你强大的实验工具但如何设计实验场景用户模型、如何解读数据区分系统瓶颈和测试工具瓶颈、如何定位问题这些更需要你的业务知识、系统架构知识和分析能力。多跑多看多思考把每次压测当成一次对系统深入理解的机会你的收获会远大于一份简单的测试报告。