1. 项目概述：为什么我们需要Locust这样的性能测试工具？

在软件开发和运维的日常里，性能测试是个绕不开的话题。无论是上线一个新接口，还是对现有系统进行容量评估，我们都需要回答一个问题：这个系统到底能扛住多少并发用户？传统的工具，比如JMeter，功能强大但配置繁琐，写个复杂的逻辑脚本得跟它的GUI界面斗智斗勇半天。对于习惯用代码解决一切问题的开发者来说，总感觉隔了一层。

这就是Locust出现并迅速在开发者社区流行开来的原因。它本质上是一个用Python写的开源负载测试框架。最大的特点就是“一切皆代码”。你用Python来定义用户行为，这意味你可以利用Python生态里几乎所有的库（比如requests, httpx, websocket-client）来模拟任何你能想到的复杂场景：从简单的HTTP API调用，到需要处理登录态、依赖前序接口结果、甚至模拟WebSocket长连接或自定义协议的压力，Locust都能优雅地胜任。

我最初接触Locust是在一个微服务项目的性能摸底阶段。当时我们需要模拟一种“用户登录后，浏览商品列表，随机选择商品加入购物车，最后部分用户执行结算”的混合场景。用JMeter实现这个流程，参数化和逻辑控制让我头疼不已。换成Locust后，我用几十行Python代码就清晰地描述了整个业务流程，还能方便地引入Faker库生成随机用户数据，测试脚本的可读性和可维护性直接上了一个台阶。从那时起，Locust就成了我性能测试工具箱里的首选。

简单来说，如果你或你的团队符合以下情况，Locust会是一个极佳的选择：

  1. 测试团队或开发人员具备Python基础。
  2. 测试场景复杂，涉及多步骤、有状态（Session）的流程。
  3. 希望测试脚本能像普通代码一样进行版本管理、模块化复用。
  4. 需要分布式压测来产生巨大的并发压力。
  5. 偏爱简洁的Web UI来实时观察测试结果，而非复杂的桌面客户端。

2. Locust核心设计哲学与架构拆解

Locust的设计非常“Pythonic”，也极其简洁。理解它的核心概念，是写出高效、准确测试脚本的关键。

2.1 核心概念：User、TaskSet与Events

Locust的世界由几个核心对象构成：

  • User类 ：这是你需要编写的最主要的类。它代表了一类虚拟用户。在这个类里，你通过 wait_time 属性定义用户在执行任务之间的等待时间（例如，模拟用户思考时间），更重要的是，通过 tasks 属性来定义这个用户会执行哪些任务。
  • TaskSet类 ：可以理解为“任务集”。一个User可以执行一个TaskSet。TaskSet内部可以定义更细粒度的任务，并且支持嵌套，让你能够构建出层次化的用户行为模型。例如，你可以定义一个 BrowseProducts 的TaskSet，里面包含“查看列表”、“搜索商品”、“查看详情”等任务；再定义一个 PurchaseFlow 的TaskSet。然后在User类里，按权重分配这两个TaskSet。
  • Task ：任务，就是用户具体执行的操作。在Locust里，一个任务就是一个Python方法（用 @task 装饰器标记）。这个方法里通常包含了发起HTTP请求、处理响应、解析数据等逻辑。
  • Events ：事件钩子。Locust提供了丰富的生命周期事件，如 init （测试初始化）、 request （每次请求发送前后）、 quitting （测试停止）等。你可以通过监听这些事件，来注入自定义逻辑，比如在测试开始时准备测试数据，在每次请求后对响应进行额外的校验，或者在测试结束后清理环境。

这种设计带来的最大好处是 极强的表现力 。你可以用面向对象的思想来建模你的虚拟用户。比如，模拟“普通浏览用户”和“抢购用户”两种不同类型的行为，你可以创建两个不同的User类，赋予它们不同的任务权重和等待时间，让测试场景更加贴近真实。

2.2 运行模式：单机与分布式

Locust的运行模式清晰且强大：

  1. 单机模式 ：这是最常用的开发调试模式。你在一台机器上启动Locust，指定虚拟用户数、增长速率和目标主机。所有用户都在单个进程中运行（通过gevent实现协程并发）。这对于快速验证脚本逻辑和进行小规模压测足够了。
  2. 分布式模式 ：当需要模拟成千上万的并发用户时，单机资源（CPU、网络、端口）会成为瓶颈。Locust的分布式模式采用一主多从（Master-Worker）架构。
    • Master节点 ：负责协调，不模拟任何用户。它启动Web UI，收集所有Worker节点的统计数据并汇总展示。
    • Worker节点 ：可以部署在多台机器上。它们从Master节点接收指令，真正负责生成和运行虚拟用户，并将测试数据实时上报给Master。

这种架构使得横向扩展变得非常容易。你只需要在多台机器上启动Worker进程并指向同一个Master，就能轻松汇聚多台机器的压力产生能力。在实际项目中，我们经常使用Docker来快速部署一整套分布式的Locust集群。

2.3 Web UI与控制台：两种管理视角

Locust提供了两种方式来控制和观察测试：

  • Web UI ：通过浏览器访问（默认 http://localhost:8089 ），提供了一个图形化界面。你可以在这里启动/停止测试，设置用户数、增长速率，并实时查看RPS（每秒请求数）、响应时间、失败率等关键指标图表。这对于演示、实时监控和快速调整参数非常友好。
  • 无头模式 ：通过命令行参数（ --headless ）启动，不启动Web UI。通常用于将测试集成到CI/CD流水线中。你可以直接指定总用户数、运行时长等参数，测试结束后会在控制台输出摘要报告，并可以配合 --csv 参数将结果导出为CSV文件，便于后续分析。

3. 从零开始：编写你的第一个Locust性能测试脚本

理论说得再多，不如动手写一个。我们以一个最常见的场景为例：测试一个简单的用户登录和查询信息的API。

3.1 环境准备与安装

首先，确保你有一个Python环境（3.6及以上）。使用pip安装Locust是最简单的方式：

pip install locust

安装完成后，可以在命令行输入 locust --version 验证是否成功。

注意：如果你的测试涉及 HTTPS 或需要更复杂的HTTP客户端行为，建议也安装 requests 库，虽然Locust内置了基于 geventhttpclient 的客户端，但 requests 的API对大多数人来说更熟悉。 pip install requests

3.2 基础脚本结构剖析

创建一个名为 locustfile.py 的文件（这是Locust默认寻找的入口文件），并输入以下内容：

from locust import HttpUser, task, between

class QuickstartUser(HttpUser):
    # 模拟用户在每个任务执行后，等待1到5秒之间的一个随机时间
    wait_time = between(1, 5)

    # 标记为一个任务，权重为3，意味着在任务列表中，它被选中的概率是3/(3+1)=75%
    @task(3)
    def view_items(self):
        # 使用self.client发起请求，它是HttpUser内置的HttpSession实例
        # 会自动记录请求耗时、状态码等信息
        with self.client.get("/api/items", catch_response=True) as response:
            # 你可以对响应进行自定义校验
            if response.status_code == 200:
                # 假设我们期望返回的JSON中包含`items`数组
                if "items" not in response.json():
                    response.failure("Response does not contain 'items' key")
            else:
                response.failure(f"Bad status code: {response.status_code}")

    @task(1) # 权重为1，选中概率25%
    def login(self):
        # 模拟登录请求，发送JSON数据
        login_data = {"username": "test_user", "password": "secret"}
        with self.client.post("/api/login", json=login_data, catch_response=True) as response:
            if response.status_code == 200:
                token = response.json().get("token")
                if token:
                    # 将token存储到用户的session中，供后续请求使用
                    self.client.headers["Authorization"] = f"Bearer {token}"
                    # 登录成功后，可以继续执行其他操作，比如查询用户信息
                    self.client.get("/api/profile")
                else:
                    response.failure("Login succeeded but no token received")
            else:
                response.failure(f"Login failed with status code: {response.status_code}")

    # 每个虚拟用户开始运行时，会执行一次on_start方法
    def on_start(self):
        # 这里可以放一些初始化操作，比如先访问一下首页
        self.client.get("/")

这个脚本定义了一个名为 QuickstartUser 的用户类。它继承自 HttpUser ，这意味着它自带了一个用于发送HTTP请求的 client 属性。

  • wait_time = between(1, 5) ：定义了思考时间，让虚拟用户行为更真实。
  • @task 装饰器将方法标记为任务。权重参数( weight )决定了任务被选中的相对概率。这里 view_items 任务比 login 任务被调用的频率更高。
  • self.client.get/post ：发起请求。 catch_response=True 允许我们手动控制请求的成功/失败判定。
  • on_start ：每个用户实例在开始执行任务循环前会调用一次，适合做登录等初始化操作。

3.3 运行与观察

在终端中，进入 locustfile.py 所在的目录，运行：

locust

默认会启动Web UI在 http://localhost:8089 。打开浏览器，你会看到Locust的启动页面。需要填写：

  • Number of users ：要模拟的总用户数。
  • Spawn rate ：每秒启动多少个用户（用户增长速率）。
  • Host ：被测试系统的根URL（例如 http://your-api-server.com ）。

填写后点击 “Start swarming”，测试就开始了。切换到 “Charts” 标签页，你可以看到实时变化的RPS、响应时间（平均、中位数、P95、P99）和失败率图表。“Statistics”标签页提供了详细的表格数据。

4. 进阶实战：构建复杂、真实的测试场景

基础脚本只能应对简单接口。在实际项目中，测试场景往往复杂得多。下面分享几个进阶技巧。

4.1 参数化与测试数据管理

压测不能所有用户都用同一组数据，这不符合真实情况，也容易触发系统的缓存机制导致测试失真。

1. 使用CSV文件管理测试数据： 假设我们需要用不同的用户名和密码进行登录测试。可以创建一个 users.csv 文件：

username,password
user1,pass123
user2,pass456
user3,pass789

然后在Locust脚本中读取并使用：

import csv
from locust import HttpUser, task, between

class ParameterizedUser(HttpUser):
    wait_time = between(2, 5)

    def on_start(self):
        # 在用户启动时，从数据池中取出一组数据
        self.user_data = self.get_user_data()
        if not self.user_data:
            self.stop() # 如果没有数据了，就停止这个用户

    @task
    def login(self):
        data = {
            "username": self.user_data['username'],
            "password": self.user_data['password']
        }
        with self.client.post("/login", json=data, catch_response=True) as resp:
            if resp.status_code == 200:
                # 登录成功，保存token等
                pass
            else:
                resp.failure(f"Login failed for {data['username']}")

    def get_user_data(self):
        # 一个简单的从CSV循环读取数据的示例（实际生产环境需要考虑并发读取和数据唯一性）
        # 更佳实践是将数据预加载到队列中，每个用户从队列中获取
        with open('users.csv', 'r') as f:
            reader = csv.DictReader(f)
            users = list(reader)
        # 这里简单返回第一个用户，实际应用中应实现一个线程安全的队列或轮询机制
        # 例如使用 itertools.cycle 或 queue.Queue
        import random
        return random.choice(users) if users else None

实操心得 ：对于大规模参数化，不建议在 on_start 或每个任务中频繁读取文件。最佳实践是在测试初始化阶段（利用 @events.init.add_listener ）将所有测试数据加载到内存中的一个共享数据结构（如 queue.Queue ）中，各个虚拟用户协程从中安全地获取数据。这能极大提升性能并避免I/O瓶颈。

2. 使用Faker库动态生成数据： 对于不需要持久化关联的数据（如搜索关键词、地址信息），使用Faker动态生成更加灵活。

from faker import Faker

class FakeDataUser(HttpUser):
    wait_time = between(1, 3)
    fake = Faker() # 每个用户实例有自己的Faker生成器

    @task
    def update_profile(self):
        profile_data = {
            "name": self.fake.name(),
            "email": self.fake.email(),
            "address": self.fake.address()
        }
        self.client.put("/api/profile", json=profile_data)

4.2 处理关联请求与状态保持

很多API调用是有状态的，比如先登录获取token，后续请求都要带上这个token。

from locust import HttpUser, task, between

class StatefulUser(HttpUser):
    wait_time = between(1, 3)

    def on_start(self):
        # 登录并保存token
        resp = self.client.post("/api/auth/login", json={"user": "test", "pwd": "test"})
        if resp.status_code == 200:
            self.token = resp.json()["access_token"]
            # 将token设置到客户端默认头中
            self.client.headers = {"Authorization": f"Bearer {self.token}"}
        else:
            # 登录失败，标记此用户为失败并停止
            resp.failure("Login failed")
            self.stop()

    @task
    def get_protected_resource(self):
        # 此时的请求会自动带上Authorization头
        self.client.get("/api/protected/data")

    @task
    def logout(self):
        # 登出，并清除token
        self.client.post("/api/auth/logout")
        self.client.headers.pop("Authorization", None)
        # 登出后，可以停止该用户或重新登录
        self.stop()

4.3 使用TaskSet组织复杂业务流程

当用户行为流程非常复杂时，使用TaskSet可以让你更好地组织代码。

from locust import HttpUser, task, TaskSet, between

class BrowseBehavior(TaskSet):
    # 这个TaskSet内的任务权重是独立的
    @task(5)
    def view_index(self):
        self.client.get("/shop")

    @task(3)
    def view_category(self):
        categories = ["electronics", "books", "clothing"]
        self.client.get(f"/shop/category/{random.choice(categories)}")

    @task(1)
    def search(self):
        self.client.get("/shop/search", params={"q": "laptop"})

    # 可以中断当前TaskSet，跳回父级（User或上一级TaskSet）的任务选择
    @task(1)
    def leave(self):
        self.interrupt()

class CartBehavior(TaskSet):
    @task
    def add_to_cart(self):
        item_id = random.randint(1, 100)
        self.client.post(f"/cart/add/{item_id}")

    @task
    def view_cart(self):
        self.client.get("/cart")

    @task(1)
    def checkout(self):
        # 结算是一个出口点，执行后中断
        self.client.post("/cart/checkout")
        self.interrupt()

class WebsiteUser(HttpUser):
    wait_time = between(2, 6)
    # 在User级别定义任务，可以是普通方法，也可以是TaskSet类
    # 这里按权重分配了三种行为：浏览行为、购物车行为、直接去首页
    tasks = {
        BrowseBehavior: 4,   # 权重4
        CartBehavior: 2,     # 权重2
        task(1): lambda self: self.client.get("/") # 直接访问首页的任务，权重1
    }

在这个例子中， WebsiteUser 虚拟用户有70%的概率（4/(4+2+1)）进入 BrowseBehavior （浏览行为），在里面随机执行查看首页、分类或搜索任务；有约28.6%的概率进入 CartBehavior （购物车行为）；有约14.3%的概率直接访问首页。 interrupt() 方法允许从嵌套的TaskSet中跳出，回到上一级的任务选择逻辑。

5. 性能测试实战：分布式压测与结果分析

单机Locust可能受限于机器性能，无法产生足够压力。分布式压测是应对高并发场景的标准做法。

5.1 搭建分布式Locust集群

假设我们有三台机器： master_node (192.168.1.100), worker1 (192.168.1.101), worker2 (192.168.1.102)。

  1. 在Master节点启动： 

    # 在 master_node 上
locust -f locustfile.py --master --host=http://your-target-system.com

    --master 参数指定当前实例为Master节点。它会启动Web UI（默认8089端口）。

  2. 在Worker节点启动： 

    # 在 worker1 和 worker2 上
locust -f locustfile.py --worker --master-host=192.168.1.100

    --worker 指定为Worker节点， --master-host 指向Master节点的IP地址。

启动后，在Master的Web UI上，你会在“Workers”标签页看到两个已连接的Worker。现在，你在UI上设置的用户数和速率，将会由这两个Worker共同承担。

注意事项 ：

  • 防火墙 ：确保Master节点（默认端口8089用于Web UI，5557用于与Worker通信）和Worker节点（默认端口5558用于接收任务）的相关端口在机器间是开放的。
  • 代码一致性 ：所有Master和Worker节点上的 locustfile.py 以及其引用的任何自定义模块、数据文件必须完全一致。通常建议使用版本控制系统（如Git）同步，或通过共享存储（如NFS）挂载。
  • 资源监控 ：压测时，务必监控Master和Worker节点的CPU、内存、网络带宽使用情况。Worker节点如果资源耗尽，会成为瓶颈，产生不了预期压力。

5.2 关键性能指标解读与瓶颈定位

Locust的Web UI和CSV报告提供了丰富的指标。看懂这些指标是分析性能瓶颈的基础。

指标 含义 分析要点
RPS (Requests/s) 每秒请求数 系统吞吐量的直接体现。随着并发用户增加，RPS增长到一定程度后趋于平缓或下降，说明系统达到瓶颈。
Response Time (ms) 响应时间 平均响应时间 ：整体趋势参考。
中位数 (Median) ：有一半的请求快于此值。
P95 / P99 ： 最关键指标 。表示95%/99%的请求响应时间低于此值。P99过高，说明有少量请求体验极差，可能遇到了慢查询、锁竞争等问题。
Failure Rate 失败率 任何非2xx/3xx的HTTP状态码或被标记为 failure 的请求都算失败。压测中失败率应接近于0。若失败率随压力上升而升高，可能是系统错误（如数据库连接池耗尽）、或测试脚本问题（如参数化数据冲突）。
Number of Users 并发用户数 当前活跃的虚拟用户数。

如何定位瓶颈？

  1. 观察曲线 ：在压测过程中，观察RPS和响应时间曲线。理想情况下，RPS随用户数线性增长，响应时间保持平稳。如果响应时间开始陡增，而RPS不再增长甚至下降，说明系统已经达到瓶颈。
  2. 对比P95/P99与中位数 ：如果P95/P99远高于中位数，说明系统处理存在“长尾”现象。可能的原因包括：数据库慢查询、外部依赖服务响应慢、垃圾回收（GC）停顿、线程锁竞争等。需要结合被压测系统的应用日志、数据库监控、JVM监控（如果是Java应用）等进一步排查。
  3. 分析失败请求 ：在Locust的“Failures”标签页查看具体的失败请求和原因。是超时？还是返回了5xx错误？这能直接指引你找到有问题的接口或服务。

5.3 将测试集成到CI/CD流水线

自动化性能测试是DevOps实践中的重要一环。你可以使用Locust的无头模式，在流水线中自动执行测试并判断结果是否通过。

# 一个基本的命令行示例，模拟100个用户，每秒增加5个，运行3分钟
locust -f locustfile.py \
  --headless \
  --host=http://staging-api.example.com \
  --users=100 \
  --spawn-rate=5 \
  --run-time=3m \
  --csv=report \
  --html=report.html
  • --headless : 无头模式，不启动Web UI。
  • --users 和 --spawn-rate : 定义负载模型。
  • --run-time : 测试运行时长。
  • --csv : 将统计数据导出为CSV文件（会生成多个，如 report_stats.csv ）。
  • --html : 生成一个HTML格式的报告摘要。

在CI脚本中（如Jenkinsfile、GitLab CI .gitlab-ci.yml ），你可以这样集成：

# .gitlab-ci.yml 示例
stages:
  - performance

locust_test:
  stage: performance
  image: python:3.9
  script:
    - pip install locust
    - locust -f locustfile.py --headless --host=$TARGET_HOST --users=200 --spawn-rate=10 --run-time=2m --csv=ci_report --html=ci_report.html --check-rps=50 --check-fail-rate=0.1
  artifacts:
    paths:
      - ci_report*.csv
      - ci_report.html
  only:
    - main

这里使用了 --check-rps 和 --check-fail-rate 参数。 --check-rps=50 表示要求平均RPS不低于50， --check-fail-rate=0.1 表示要求失败率不高于0.1（10%）。如果测试结果不满足这些条件，Locust会以非零状态码退出，从而使CI任务失败，达到质量门禁的目的。

6. 避坑指南与高级技巧

在实际使用Locust的过程中，我踩过不少坑，也总结了一些提升测试效率和准确性的技巧。

6.1 常见问题与排查

问题1：压测时RPS上不去，但CPU/内存占用很低。

  • 可能原因 ： wait_time 设置过长。虚拟用户大部分时间在“思考”，没有发出足够请求。
  • 排查 ：检查脚本中的 wait_time 。对于纯压力测试（非模拟真实用户节奏），可以设置为 constant(0) 或很小的值。同时，检查被压测服务的网络延迟和连接数限制。

问题2：出现大量 “ConnectionResetError” 或 “RemoteDisconnected” 错误。

  • 可能原因 ：被压测服务器或中间件（如Nginx）的连接数已满，或Keep-Alive设置有问题。也可能是Locust客户端端口耗尽。
  • 解决方案 ：
    1. 调整Locust的HTTP客户端配置（在 HttpUser 类中设置）： 
      class MyUser(HttpUser):
    # 禁用连接池，每个请求新建连接（不推荐，开销大）
    # client = requests.Session() ... 然后配置
    # 或者使用自定义的HttpSession，调整连接适配器参数
    pass
    2. 更常见的是需要调整系统级的TCP参数（在压测机上）： 
      # 临时增加本地端口范围
sysctl -w net.ipv4.ip_local_port_range="1024 65535"
# 减少TIME_WAIT状态的等待时间，加速端口回收
sysctl -w net.ipv4.tcp_tw_reuse=1
sysctl -w net.ipv4.tcp_fin_timeout=30
    3. 检查并调整被压测服务端的最大连接数、线程池等配置。

问题3：测试数据（如用户Token）在Worker间冲突或重复使用。

  • 原因 ：在分布式模式下，如果所有Worker从同一个全局列表里顺序取数据，会导致不同Worker上的用户拿到相同数据。
  • 解决方案 ：使用独立的数据源或更智能的分片策略。例如，为每个Worker预分配一个数据段，或者使用支持原子操作的分布式队列（如Redis）来分发测试数据。在 @events.test_start.add_listener 中初始化一个全局的 queue.Queue ，并让每个用户在 on_start 时从中获取，是单机多进程的安全做法，但在分布式下需要更复杂的同步机制。

6.2 性能优化技巧

  1. 减少客户端开销 ：

    • 使用FastHttpUser ：Locust默认的 HttpUser 基于 geventhttpclient ，性能已经不错。但社区还提供了一个 FastHttpUser ，它基于 httptools 和 python-socks ，性能更高，尤其是在需要模拟极高并发（上万）时。安装 locust[fast] 即可使用。
    • 谨慎使用 catch_response ： catch_response=True 会捕获响应内容，如果响应体很大（如下载文件），会消耗大量内存。对于只关心状态码的请求，不要使用它。
    • 避免在任务中执行繁重的计算或I/O ：Locust的协程是单线程的，如果在任务中执行了阻塞性操作（如 sleep , 同步的文件读写，密集CPU计算），会阻塞整个事件循环，严重影响压测能力。应将此类操作替换为非阻塞版本或移到外部处理。

  2. 编写可维护的测试代码 ：

    • 模块化 ：将公共方法（如登录、获取配置）提取到单独的Python模块中。
    • 使用配置管理 ：不要将主机名、用户数等硬编码在脚本里。使用环境变量或配置文件（如 config.yaml ）来管理。
    • 善用事件钩子 ：利用 test_start 和 test_stop 事件来初始化和清理测试环境（如创建测试用户、清理测试数据）。

6.3 超越HTTP：测试其他协议

虽然Locust以HTTP测试闻名，但其基于协程的架构使其可以测试任何协议。你需要自己实现客户端逻辑。

示例：测试WebSocket服务 

import gevent
from websocket import create_connection, WebSocketTimeoutException
from locust import User, task, between, events

class WebSocketClient:
    def __init__(self, host):
        self.host = host
        self.ws = None

    def connect(self):
        self.ws = create_connection(f"ws://{self.host}/ws")
        # 可以在这里监听连接事件
        events.request.fire(request_type="WS", name="Connect", response_time=0, response_length=0)

    def send(self, message, name="Message"):
        start_time = time.time()
        try:
            self.ws.send(message)
            response = self.ws.recv() # 假设需要接收回复
            total_time = int((time.time() - start_time) * 1000)
            events.request.fire(
                request_type="WS",
                name=name,
                response_time=total_time,
                response_length=len(response),
                exception=None,
            )
            return response
        except WebSocketTimeoutException as e:
            total_time = int((time.time() - start_time) * 1000)
            events.request.fire(
                request_type="WS",
                name=name,
                response_time=total_time,
                response_length=0,
                exception=e,
            )
            raise

    def close(self):
        if self.ws:
            self.ws.close()

class WebSocketUser(User):
    abstract = True # 这是一个抽象基类
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.client = WebSocketClient(self.host)

    def on_start(self):
        self.client.connect()

    def on_stop(self):
        self.client.close()

class MyWebSocketUser(WebSocketUser):
    wait_time = between(0.5, 3)

    @task
    def send_ping(self):
        # 使用自定义的client发送消息
        response = self.client.send("ping", "Ping")
        # 可以对response进行断言
        if response != "pong":
            events.request.fire(
                request_type="WS",
                name="Ping",
                response_time=0,
                response_length=0,
                exception=AssertionError("Expected 'pong'"),
            )

这个例子展示了如何通过自定义客户端和触发Locust的 events.request 事件，将非HTTP协议的请求也纳入Locust的统计系统中。你可以用类似的方法测试gRPC、Socket.IO、自定义TCP协议等。

Locust的魅力在于它的简洁和灵活。它没有试图封装一切，而是给了你足够的积木，让你能用Python代码搭建出任何你想要的负载模型。从简单的API压测到复杂的全链路业务场景模拟，它都能很好地胜任。掌握它，意味着你拥有了用代码定义和驾驭流量的能力。

Logo
亚马逊云科技技术品牌专区

更多推荐

Kiro Editor 开发实战：使用 Cargo 构建、测试与性能优化指南

欢迎来到这篇终极指南，我们将深入探索如何使用Rust构建高性能的终端文本编辑器Kiro Editor。无论你是Rust新手还是经验丰富的开发者，这篇完整教程将带你了解如何利用Cargo工具链进行高效的开发、测试和性能优化，打造一款快速、轻量且功能强大的UTF-8文本编辑器。## 什么是Kiro Editor？Kiro Editor是一款使用Rust编写的极简终端文本编辑器，它最初是著名编辑

avatar 亚马逊云科技技术品牌专区

近期零基础学量化，先按基础差异拆学习顺序

读者应明白，拆解学习顺序时要先看自己的基础缺口，再决定哪一段需要放慢。不同流程阶段都有不同检查任务，学习路线要能反映这些差异。

avatar 亚马逊云科技技术品牌专区

2026年量化工具选择，要跟着能力基础走

读者应理解，选择工具前要先判断自己的能力基础和当前任务。工具应该帮助自己补上当前最关键的缺口，而不是替代学习顺序本身。

avatar 亚马逊云科技技术品牌专区