从零到一掌握Locust：Python分布式性能测试实战指南

1. 项目概述：为什么是Locust？

如果你正在寻找一个能模拟成千上万用户、用代码定义用户行为、并且能让你完全掌控测试逻辑的性能测试工具，那么Locust大概率会进入你的视野。它不是JMeter那样的“点击式”工具，而是一个用Python代码编写测试脚本的开源框架。这意味着，如果你或你的团队熟悉Python，Locust的学习曲线会异常平滑，你获得的灵活性和扩展性也是传统工具难以比拟的。

我最初接触性能测试时，也用过一些录制回放的工具，但很快就遇到了瓶颈：复杂的业务逻辑、动态参数、条件判断、数据驱动，这些在图形界面里配置起来要么极其繁琐，要么根本无法实现。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或PowerShell，macOS/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的支持非常出色。

1. 安装Python扩展：在VS Code的扩展商店搜索并安装“Python”扩展（由Microsoft发布）。
2. 选择解释器：用VS Code打开你的locust_project文件夹。按Ctrl+Shift+P（Windows/Linux）或Cmd+Shift+P（macOS）打开命令面板，输入“Python: Select Interpreter”，然后选择路径中包含locust_project\venv或locust_project/.venv的Python解释器。这确保了VS Code使用我们刚创建的虚拟环境。
3. 安装代码格式化工具（推荐）：在VS Code的终端（确保已激活venv）里运行pip install black。然后在VS Code设置中，将“Format On Save”打开，并设置默认格式化工具为Black。这能让你的代码保持整洁统一的风格。

至此，你的编码环境已经就绪。我们即将开始编写第一个性能测试脚本。

3. 核心概念与第一个测试脚本

在动手写代码前，必须理解Locust里的三个核心概念，这构成了所有测试脚本的骨架。

1. HttpUser（或User类）： 这是虚拟用户的蓝图。你定义的每个用户类，都代表一类用户行为。HttpUser是一个最常用的内置类，它提供了方便的client属性（一个HttpSession实例）来发送HTTP请求。
2. TaskSet： 任务集。用于将多个任务（Task）组织在一起，可以嵌套，模拟复杂的用户操作流程。在Locust 2.x版本中，更推荐使用@task装饰器直接在User类中定义任务，但对于有顺序或权重的任务组，TaskSet依然有用。
3. @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_response=True) as response: # 你可以对响应进行断言，判断请求是否“真正”成功 if response.status_code == 200 and "文章列表" in response.text: response.success() else: response.failure(f"Failed! Status: {response.status_code}") @task(1) def search_article(self): # 模拟一个带查询参数的搜索请求 self.client.get("/search?keyword=locust") # 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(f"User {self.id} logged in successfully.") # 通常可以将token保存在self.client.headers中供后续请求使用 # self.client.headers["Authorization"] = f"Bearer {login_response.json()['token']}" else: print(f"User {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_response=True上下文管理器，并不仅依赖HTTP状态码，还检查了响应体内容，以此更精确地判断业务成功与否。这是生产级脚本的必备操作。

实操心得：catch_response=True和手动调用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 --host=https://your-test-api.com --web-host=0.0.0.0 --web-port=8089

• --host：指定被测试系统的基地址。我们脚本中的self.client.get("/articles")会拼接成https://your-test-api.com/articles。
• --web-host=0.0.0.0：允许其他机器通过IP访问Web UI，方便远程监控。
• --web-port：指定Web UI端口。

启动后，打开浏览器访问http://localhost:8089，你会看到Locust的启动界面。

4.2 Web UI配置与启动

Web UI界面非常直观：

1. Number of users：要模拟的总用户数（峰值）。
2. Spawn rate：每秒启动多少个用户，直到达到总用户数。例如，用户数1000，生成速率10，意味着Locust会用100秒逐步启动所有用户，而不是瞬间启动，这有助于观察系统在压力逐步增大时的表现。
3. 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_response=True) as response: # 注意：这里使用了`name`参数。Locust的统计是按`name`聚合的。 # 如果不加`name`，每个不同的keyword都会产生一条独立的统计记录，导致图表杂乱。 # 使用`name`将其归一化为“/search?q=[keyword]”这一项。 if response.status_code == 200: response.success() else: response.failure(f"Search 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.client（HttpSession）会自动处理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": "test@locust.io", "password": "test123"}) if login_resp.status_code == 200: self.token = login_resp.json()["access_token"] # 2. 将token设置到后续所有请求的Header中 self.client.headers.update({"Authorization": f"Bearer {self.token}"}) print(f"User {self.id} got token.") else: # 如果登录失败，可以标记此用户为停止状态，或者记录失败 self.stop(force=True) # 强制停止这个用户实例 @task def get_private_data(self): # 此时的请求会自动带上Authorization Header self.client.get("/user/private-data") @task(2) # 这个任务权重更高 def update_profile(self): new_name = f"LocustUser_{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", json=payload, catch_response=True) 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 --host=https://your-test-api.com --web-host=0.0.0.0

--master参数指定此实例为Master。--web-host=0.0.0.0让Web UI能被其他机器访问。

步骤2：启动Worker节点在worker1_host和worker2_host机器上，需要拥有相同的locustfile.py脚本和Python环境。然后分别运行：

locust --worker --master-host=192.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 --host=https://your-test-api.com worker: image: locustio/locust volumes: - ./:/mnt/locust command: -f /mnt/locust/locustfile.py --worker --master-host=master deploy: replicas: 4 # 启动4个worker容器实例

然后在本机运行docker-compose up -d，就会启动1个Master容器和4个Worker容器，并自动连接。你只需访问本机的8089端口即可管理测试。

避坑技巧：

1. 网络与防火墙：确保Master节点的5557（Worker通信端口）和8089（Web UI端口）在防火墙中是开放的，并且Worker节点能访问到Master节点的IP。
2. 时间同步：分布式环境下，确保所有机器（或容器）的时间基本同步，否则日志时间戳会混乱。可以使用NTP服务。
3. 数据一致性：确保所有Worker节点上的测试数据文件（如CSV）内容一致，或者通过共享存储（如NFS）访问同一份文件。
4. 资源监控：压测时，不仅要监控被测系统，也要监控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 --host=https://your-test-api.com

• --headless：启用无头模式。
• --users：总用户数。
• --spawn-rate：生成速率。
• --run-time：测试运行时间（例如30s,5m,1h30m）。时间到后测试自动停止。
• --csv=prefix：以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/checkout@v3 - name: Set up Python uses: actions/setup-python@v4 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 }} \ --csv=results/locust_results \ --html=results/locust_report.html \ --loglevel INFO continue-on-error: true # 即使测试失败，也继续执行后续步骤以保存结果 - name: Upload test results uses: actions/upload-artifact@v3 with: name: locust-reports path: results/ retention-days: 7

在这个工作流中：

1. 我们安装了Python和Locust。
2. 以无头模式运行Locust，模拟50个用户，在1分钟内完成测试。
3. 将结果输出为CSV和HTML报告。
4. 最后，将生成的结果文件上传到GitHub Actions的Artifacts中，供后续下载分析。

注意事项：在CI中运行性能测试，通常是对一个独立的测试环境（如Staging环境）进行。要确保该环境的稳定性和数据隔离。另外，CI环境可能资源有限，模拟的用户数不宜过大，重点在于回归验证核心接口的性能是否出现显著退化，而非进行极限压测。

8. 常见问题排查与性能调优心得

在实际使用Locust的过程中，你肯定会遇到各种问题。这里记录了一些典型问题和我的排查思路。

8.1 Locust自身问题排查

8.2 测试脚本与性能调优心得

1. 关于wait_time：wait_time = between(1, 3)和wait_time = constant(1)产生的压力模型天差地别。前者更贴近真实用户（有思考时间），后者是持续不断的请求。选择哪种取决于你的测试目标：是考察系统在真实用户模型下的表现，还是考察系统的纯接口吞吐量极限。大多数场景下，使用随机等待时间更合理。

2. 关于网络连接：Locust默认对每个请求使用短连接（HTTP/1.1）。对于高并发场景，建立TCP连接的开销很大。启用HTTP Keep-Alive可以大幅提升效率。幸运的是，Locust底层使用的geventhttpclient默认就支持Keep-Alive。你只需要确保你的locustfile.py中使用的self.client是同一个会话对象（默认就是），它就会自动复用连接。你可以通过打印self.client的id来验证。

3. 减少不必要的日志：默认的INFO日志会打印每个请求，在高压下会产生巨量IO，影响性能。在生产压测时，可以使用--loglevel WARNING或--loglevel ERROR来减少日志输出。

4. 小心Python的GIL：Locust基于gevent（一个协程库），通过异步IO来实现高并发。但如果你在任务中执行了大量的CPU密集型计算（如复杂的图像处理、数据加密），这会阻塞整个线程，严重影响并发能力。对于CPU密集型操作，应考虑将其移到被测服务端，或者使用Locust的fast_http客户端（已废弃）或探索其他客户端模式。通常，性能测试脚本的逻辑应尽量轻量，只负责发请求和做简单断言。

5. 断言要快，但要准：catch_response=True和响应内容解析（如response.json()）会有性能开销。如果只是为了检查状态码，直接用response.status_code == 200判断即可。如果需要检查响应体，尽量使用简单的字符串查找（如in操作）而非完整的JSON解析，除非必要。在超高压测试下，这些开销累积起来也不容忽视。

6. 分布式数据共享难题：如果测试需要全局唯一ID或共享计数器（例如，模拟用户注册，用户名不能重复），简单的cycle迭代文件在分布式下会出问题（多个Worker会读取相同的数据段）。解决方案有：a) 使用中央存储如Redis生成ID；b) 为每个Worker预分配不同的数据段；c) 使用更复杂的分布式任务队列。这通常是编写高级Locust脚本时最复杂的部分。

最后，性能测试的本质不是“跑工具”，而是“做实验”和“观察分析”。Locust给了你强大的实验工具，但如何设计实验场景（用户模型）、如何解读数据（区分系统瓶颈和测试工具瓶颈）、如何定位问题，这些更需要你的业务知识、系统架构知识和分析能力。多跑，多看，多思考，把每次压测当成一次对系统深入理解的机会，你的收获会远大于一份简单的测试报告。