环境声明

• Python 版本：Python 3.12+（建议 3.10 以上版本）
• Locust 版本：Locust 2.32+
• 开发工具：PyCharm 或 VS Code
• 操作系统：Windows / macOS / Linux（通用）

---

学习目标

1. 理解性能测试的核心概念与分类
2. 掌握 Locust 框架的安装与基本架构
3. 学会编写 User 类和定义测试任务
4. 掌握 HTTP 请求的编写与认证处理
5. 理解负载模式与并发控制策略
6. 学会收集和分析性能指标
7. 掌握事件钩子与自定义指标的方法
8. 了解分布式负载测试的配置与执行
9. 通过电商网站实战案例巩固所学知识

---

1. 性能测试基础概念

性能测试是验证系统在不同负载条件下的响应能力、稳定性和资源消耗情况的重要手段。根据测试目的的不同，性能测试可分为以下几类：

1.1 负载测试（Load Testing）

负载测试是在预期负载条件下验证系统性能的行为。目的是确定系统在正常工作负载下的表现，确保系统能够处理预期的用户数量。

典型场景：

• 电商网站在促销活动期间的用户访问量
• 在线教育平台在直播课开始时的并发登录

1.2 压力测试（Stress Testing）

压力测试是将系统暴露在极端负载条件下，以确定系统的性能瓶颈和崩溃点。目的是了解系统在超出正常负载时的行为。

典型场景：

• 系统能够承受的最大并发用户数
• 系统在资源耗尽时的降级策略

1.3 稳定性测试（Soak Testing）

稳定性测试是在长时间运行的情况下验证系统的稳定性和资源泄漏情况。目的是发现内存泄漏、连接池耗尽等渐进性问题。

典型场景：

• 系统连续运行 24 小时或更长时间
• 验证数据库连接池是否会出现耗尽

1.4 容量测试（Capacity Testing）

容量测试是确定系统能够支持的最大用户数量或数据量。目的是为系统扩容提供数据支持。

典型场景：

• 确定数据库在性能下降前能存储多少数据
• 评估服务器集群需要多少节点

---

2. 性能指标详解

性能测试需要关注以下核心指标：

2.1 响应时间（Response Time）

响应时间是指从发送请求到接收到响应所经历的时间。通常关注以下统计值：

指标	说明	建议阈值
平均响应时间	所有请求响应时间的平均值	< 500ms
中位数响应时间	50% 请求的响应时间	< 300ms
90% 响应时间	90% 请求的响应时间	< 1000ms
95% 响应时间	95% 请求的响应时间	< 2000ms
最大响应时间	最慢请求的响应时间	视业务而定

2.2 吞吐量（Throughput）

吞吐量表示系统在单位时间内处理的请求数量，通常以 RPS（Requests Per Second） 或 TPS（Transactions Per Second） 衡量。

2.3 并发数（Concurrency）

并发数是指同时向系统发送请求的用户数量。注意区分：

• 并发用户数：同时在线的用户数量
• 并发请求数：同时处理的请求数量

2.4 错误率（Error Rate）

错误率是指失败请求占总请求的百分比。一般要求：

• 错误率 < 0.1%（核心接口）
• 错误率 < 1%（普通接口）

2.5 资源利用率（Resource Utilization）

包括 CPU 使用率、内存使用率、磁盘 I/O、网络带宽等。通常建议：

• CPU 使用率 < 70%
• 内存使用率 < 80%
• 磁盘 I/O 等待时间 < 20ms

---

3. Locust 安装与架构

3.1 安装 Locust

# 使用 pip 安装
pip install locust

# 验证安装
locust --version

3.2 Locust 架构特点

Locust 是一个基于 Python 的可扩展负载测试工具，具有以下核心特点：

基于协程（Coroutine）

Locust 使用 Python 的 asyncio 和 gevent 实现协程并发，可以在单台机器上模拟数千甚至数万个并发用户，而不会像线程那样消耗大量内存。

分布式设计

Locust 支持 Master-Worker 分布式架构，可以通过多台机器共同产生负载，轻松实现大规模并发测试。

Web UI 界面

Locust 提供直观的 Web 界面，可以实时监控测试进度、查看性能指标、调整并发用户数等。

3.3 Locust 核心组件

组件	说明
User	模拟真实用户行为的类
Task	用户执行的具体操作
HttpUser	专门用于 HTTP 测试的 User 子类
events	事件钩子，用于自定义指标和报告
stats	内置的统计信息收集器

---

4. User 类与任务定义

4.1 创建基础 User 类

from locust import HttpUser, task, between


class WebsiteUser(HttpUser):
    """网站用户模拟类"""
    
    # 用户执行任务之间的等待时间（1-5秒）
    wait_time = between(1, 5)
    
    # 基础 URL（可选，也可以在请求中指定完整 URL）
    host = "https://example.com"
    
    @task
    def index_page(self):
        """访问首页"""
        self.client.get("/")
    
    @task
    def about_page(self):
        """访问关于页面"""
        self.client.get("/about")

4.2 Task 装饰器详解

task 装饰器用于标记 User 类中的方法为测试任务。

from locust import HttpUser, task


class TaskDemoUser(HttpUser):
    
    # 权重为 3，执行概率是 view_item 的 3 倍
    @task(3)
    def view_index(self):
        self.client.get("/")
    
    # 权重为 1
    @task(1)
    def view_item(self):
        self.client.get("/item/123")
    
    # 权重为 2
    @task(2)
    def view_cart(self):
        self.client.get("/cart")

权重分配说明：

• view_index 执行概率 = 3 / (3+1+2) = 50%
• view_item 执行概率 = 1 / (3+1+2) = 16.7%
• view_cart 执行概率 = 2 / (3+1+2) = 33.3%

4.3 任务序列与任务集

from locust import HttpUser, task, SequentialTaskSet


class UserBehavior(SequentialTaskSet):
    """顺序执行的任务集"""
    
    @task
    def login(self):
        self.client.post("/login", json={
            "username": "test_user",
            "password": "test_pass"
        })
    
    @task
    def view_profile(self):
        # SequentialTaskSet 会按顺序执行
        self.client.get("/profile")
    
    @task
    def logout(self):
        self.client.post("/logout")


class WebsiteUser(HttpUser):
    tasks = [UserBehavior]
    wait_time = between(1, 3)

---

5. 请求编写

5.1 GET 与 POST 请求

from locust import HttpUser, task


class ApiUser(HttpUser):
    
    @task
    def get_request(self):
        """GET 请求示例"""
        response = self.client.get(
            "/api/users",
            params={"page": 1, "size": 10},  # 查询参数
            headers={"Accept": "application/json"}
        )
        # 可以添加断言
        if response.status_code != 200:
            print(f"请求失败: {response.text}")
    
    @task
    def post_request(self):
        """POST 请求示例"""
        response = self.client.post(
            "/api/users",
            json={
                "name": "张三",
                "email": "zhangsan@example.com"
            },
            headers={"Content-Type": "application/json"}
        )

5.2 请求头与认证

from locust import HttpUser, task, between


class AuthenticatedUser(HttpUser):
    wait_time = between(1, 3)
    
    def on_start(self):
        """每个用户启动时执行，用于登录获取 Token"""
        response = self.client.post("/api/login", json={
            "username": "test_user",
            "password": "test_password"
        })
        
        if response.status_code == 200:
            data = response.json()
            self.token = data.get("token")
            # 设置默认请求头
            self.client.headers.update({
                "Authorization": f"Bearer {self.token}",
                "Content-Type": "application/json"
            })
    
    @task
    def access_protected_resource(self):
        """访问受保护资源"""
        self.client.get("/api/protected/data")
    
    def on_stop(self):
        """每个用户停止时执行，用于清理"""
        self.client.post("/api/logout")

5.3 文件上传与表单提交

from locust import HttpUser, task


class FileUploadUser(HttpUser):
    
    @task
    def upload_file(self):
        """文件上传示例"""
        with open("test_file.txt", "rb") as f:
            self.client.post(
                "/api/upload",
                files={"file": ("test_file.txt", f, "text/plain")},
                data={"description": "测试文件上传"}
            )
    
    @task
    def submit_form(self):
        """表单提交示例"""
        self.client.post(
            "/api/form",
            data={
                "username": "testuser",
                "password": "testpass",
                "remember": "on"
            }
        )

---

6. 负载模式与并发控制

6.1 等待时间策略

from locust import HttpUser, task, between, constant, constant_pacing


class WaitTimeDemoUser(HttpUser):
    
    # 在 1-5 秒之间随机等待（最常用）
    # wait_time = between(1, 5)
    
    # 固定等待 3 秒
    # wait_time = constant(3)
    
    # 每 5 秒执行一次任务（任务执行时间 + 等待时间 = 5 秒）
    wait_time = constant_pacing(5)
    
    @task
    def my_task(self):
        self.client.get("/")

6.2 自定义负载模式

from locust import HttpUser, task
import random


class CustomWaitUser(HttpUser):
    
    def wait_time(self):
        """自定义等待时间函数"""
        # 90% 的概率等待 1-3 秒，10% 的概率等待 10-15 秒
        if random.random() < 0.9:
            return random.uniform(1, 3)
        else:
            return random.uniform(10, 15)
    
    @task
    def my_task(self):
        self.client.get("/")

6.3 阶梯负载测试

使用 LoadTestShape 实现复杂的负载模式：

from locust import HttpUser, task, LoadTestShape


class StepLoadShape(LoadTestShape):
    """
    阶梯负载模式：
    每 30 秒增加 100 个用户，直到达到 1000 用户
    """
    
    step_time = 30  # 每步持续时间（秒）
    step_load = 100  # 每步增加的用户数
    spawn_rate = 50  # 每秒启动用户数
    max_users = 1000  # 最大用户数
    
    def tick(self):
        run_time = self.get_run_time()
        
        # 计算当前应该有多少用户
        current_step = run_time // self.step_time
        user_count = min(current_step * self.step_load, self.max_users)
        
        return (user_count, self.spawn_rate)


class WebsiteUser(HttpUser):
    @task
    def index(self):
        self.client.get("/")

6.4 峰值负载测试

from locust import LoadTestShape


class SpikeLoadShape(LoadTestShape):
    """
    峰值负载模式：
    模拟突发流量，快速增加到高峰，然后快速下降
    """
    
    def tick(self):
        run_time = self.get_run_time()
        
        if run_time < 60:
            # 第一阶段：快速 ramp up 到 5000 用户
            return (int(run_time * 83), 100)
        elif run_time < 120:
            # 第二阶段：保持 5000 用户 60 秒
            return (5000, 100)
        elif run_time < 180:
            # 第三阶段：快速下降到 500 用户
            return (500, 100)
        else:
            # 测试结束
            return None

---

7. 性能指标收集与分析

7.1 内置统计信息

Locust 自动收集以下统计信息：

指标	说明
Requests/s	每秒请求数
Fails/s	每秒失败数
Median	中位数响应时间
90%ile	90% 响应时间
95%ile	95% 响应时间
99%ile	99% 响应时间
Average	平均响应时间
Min	最小响应时间
Max	最大响应时间
Average size	平均响应大小

7.2 失败分析

from locust import HttpUser, task, events


class ErrorHandlingUser(HttpUser):
    
    @task
    def test_with_error_handling(self):
        with self.client.get("/api/data", catch_response=True) as response:
            if response.status_code == 200:
                data = response.json()
                if data.get("status") != "success":
                    # 标记为失败
                    response.failure(f"业务逻辑失败: {data}")
            else:
                response.failure(f"HTTP 错误: {response.status_code}")

7.3 自定义成功/失败标记

from locust import HttpUser, task


class CustomResponseUser(HttpUser):
    
    @task
    def check_response_time(self):
        with self.client.get("/api/slow", catch_response=True) as response:
            if response.elapsed.total_seconds() > 2.0:
                response.failure("响应时间超过 2 秒")
            elif response.status_code == 200:
                response.success()
            else:
                response.failure(f"状态码错误: {response.status_code}")

---

8. 事件钩子与自定义指标

8.1 事件钩子详解

from locust import events
from locust.runners import MasterRunner


@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    """测试开始时执行"""
    print("测试开始！")
    if isinstance(environment.runner, MasterRunner):
        print("主节点启动测试")


@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    """测试结束时执行"""
    print("测试结束！")


@events.request.add_listener
def on_request(request_type, name, response_time, response_length, 
               response, context, exception, **kwargs):
    """每次请求完成时执行"""
    if exception:
        print(f"请求失败: {name}, 异常: {exception}")


@events.user_error.add_listener
def on_user_error(user_instance, exception, tb, **kwargs):
    """用户执行出错时执行"""
    print(f"用户错误: {exception}")

8.2 自定义指标收集

from locust import HttpUser, task, events
from locust.stats import stats_logger
import time


# 自定义指标存储
custom_metrics = {
    "login_success": 0,
    "login_failure": 0,
    "total_response_size": 0
}


@events.request.add_listener
def on_request(request_type, name, response_time, response_length, 
               response, context, exception, **kwargs):
    """收集自定义指标"""
    if response_length:
        custom_metrics["total_response_size"] += response_length
    
    if name == "/api/login":
        if exception:
            custom_metrics["login_failure"] += 1
        else:
            custom_metrics["login_success"] += 1


@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    """测试结束时输出自定义指标"""
    print("\n===== 自定义指标报告 =====")
    print(f"登录成功次数: {custom_metrics['login_success']}")
    print(f"登录失败次数: {custom_metrics['login_failure']}")
    print(f"总响应大小: {custom_metrics['total_response_size']} bytes")
    print("========================\n")


class MetricsUser(HttpUser):
    
    @task
    def login_and_browse(self):
        # 登录
        self.client.post("/api/login", json={
            "username": "test",
            "password": "test"
        })
        
        # 浏览
        self.client.get("/api/products")

8.3 自定义客户端

from locust import User, task, events
from locust.exception import StopUser
import time
import random


class DatabaseClient:
    """自定义数据库客户端"""
    
    def __init__(self, host):
        self.host = host
        self.connection = None
    
    def connect(self):
        # 模拟数据库连接
        time.sleep(0.1)
        self.connection = f"connection_to_{self.host}"
        return self.connection
    
    def query(self, sql):
        start_time = time.time()
        try:
            # 模拟查询
            time.sleep(random.uniform(0.01, 0.1))
            result = f"Result of: {sql}"
            
            # 触发请求事件
            events.request.fire(
                request_type="DATABASE",
                name="query",
                response_time=(time.time() - start_time) * 1000,
                response_length=len(str(result)),
                response=None,
                context={},
                exception=None
            )
            return result
        except Exception as e:
            events.request.fire(
                request_type="DATABASE",
                name="query",
                response_time=(time.time() - start_time) * 1000,
                response_length=0,
                response=None,
                context={},
                exception=e
            )
            raise
    
    def close(self):
        self.connection = None


class DatabaseUser(User):
    """使用自定义客户端的用户"""
    
    abstract = True
    
    def __init__(self, environment):
        super().__init__(environment)
        self.client = DatabaseClient("localhost:5432")
    
    def on_start(self):
        self.client.connect()
    
    def on_stop(self):
        self.client.close()


class QueryUser(DatabaseUser):
    
    @task
    def run_query(self):
        self.client.query("SELECT * FROM users")

---

9. 分布式负载测试

9.1 Master-Worker 架构

Locust 支持分布式负载测试，架构如下：

+----------------+     +----------------+     +----------------+
|    Master      |---->|    Worker 1    |     |    Worker 2    |
|  (Web UI)      |     |  (产生负载)     |     |  (产生负载)     |
+----------------+     +----------------+     +----------------+
       |
       |              +----------------+
       +------------->|    Worker N    |
                      |  (产生负载)     |
                      +----------------+

Master 节点：负责协调、收集统计信息、提供 Web UI
Worker 节点：负责实际产生负载

9.2 启动分布式测试

启动 Master 节点：

# 启动 master，Web UI 默认在 8089 端口
locust -f locustfile.py --master

# 指定 master 绑定的端口
locust -f locustfile.py --master --master-bind-port=5557

启动 Worker 节点：

# 在同一台机器上启动 worker
locust -f locustfile.py --worker --master-host=localhost

# 在另一台机器上启动 worker
locust -f locustfile.py --worker --master-host=192.168.1.100

# 启动多个 worker 进程（利用多核 CPU）
locust -f locustfile.py --worker --master-host=localhost &
locust -f locustfile.py --worker --master-host=localhost &
locust -f locustfile.py --worker --master-host=localhost &

9.3 分布式测试最佳实践

from locust import HttpUser, task


class DistributedUser(HttpUser):
    """
    分布式测试注意事项：
    1. 确保所有 worker 都能访问被测系统
    2. 使用共享数据源时考虑并发问题
    3. 避免在 User 类中存储大量状态
    """
    
    @task
    def distributed_task(self):
        # 使用相对路径，确保在所有 worker 上都能正常工作
        self.client.get("/api/data")

---

10. 性能测试最佳实践

10.1 测试环境准备

1. 隔离环境：性能测试应在独立的测试环境中进行，避免影响生产环境
2. 数据准备：准备足够数量和真实分布的测试数据
3. 监控配置：部署完整的监控体系（应用监控、服务器监控、数据库监控）
4. 网络环境：确保测试机与被测系统之间的网络带宽不会成为瓶颈

10.2 数据准备策略

from locust import HttpUser, task
import random


# 预生成的测试数据
TEST_USERS = [
    {"username": f"user_{i}", "password": f"pass_{i}"}
    for i in range(1000)
]

PRODUCT_IDS = list(range(1, 10000))


class DataDrivenUser(HttpUser):
    
    def on_start(self):
        # 每个用户随机选择一个账号
        self.user_info = random.choice(TEST_USERS)
    
    @task
    def login_with_random_user(self):
        self.client.post("/api/login", json=self.user_info)
    
    @task
    def view_random_product(self):
        product_id = random.choice(PRODUCT_IDS)
        self.client.get(f"/api/products/{product_id}")

10.3 结果分析与报告

性能测试报告应包含以下内容：

1. 测试目标：本次测试的目的和预期指标
2. 测试环境：硬件配置、软件版本、网络拓扑
3. 测试场景：并发用户数、请求分布、持续时间
4. 性能指标：响应时间、吞吐量、错误率、资源利用率
5. 瓶颈分析：系统瓶颈点和优化建议
6. 结论与建议：是否通过测试，后续优化方向

---

11. 实战案例：电商网站性能测试

11.1 场景设计

模拟电商网站的典型用户行为：

用户行为流程：
1. 登录（100%）
2. 浏览商品列表（80%）
3. 查看商品详情（60%）
4. 加入购物车（30%）
5. 提交订单（20%）
6. 查看订单（40%）
7. 退出登录（100%）

11.2 完整测试代码

"""
电商网站性能测试脚本
测试场景：用户登录 -> 浏览 -> 下单 -> 退出
"""
from locust import HttpUser, task, between, SequentialTaskSet
import random


# 测试数据
TEST_USERS = [
    {"username": f"test_user_{i}", "password": "password123"}
    for i in range(100)
]

PRODUCT_IDS = list(range(1001, 1101))


class UserLoginTask(SequentialTaskSet):
    """用户登录任务集"""
    
    def on_start(self):
        """用户启动时执行登录"""
        self.user_info = random.choice(TEST_USERS)
        self.token = None
        self.login()
    
    def login(self):
        """登录获取 Token"""
        with self.client.post(
            "/api/auth/login",
            json=self.user_info,
            catch_response=True
        ) as response:
            if response.status_code == 200:
                data = response.json()
                if data.get("code") == 0:
                    self.token = data.get("data", {}).get("token")
                    self.client.headers.update({
                        "Authorization": f"Bearer {self.token}"
                    })
                    response.success()
                else:
                    response.failure(f"登录失败: {data}")
            else:
                response.failure(f"HTTP 错误: {response.status_code}")
    
    @task(4)
    def browse_products(self):
        """浏览商品列表"""
        page = random.randint(1, 10)
        self.client.get(
            "/api/products",
            params={"page": page, "size": 20}
        )
    
    @task(3)
    def view_product_detail(self):
        """查看商品详情"""
        product_id = random.choice(PRODUCT_IDS)
        self.client.get(f"/api/products/{product_id}")
    
    @task(2)
    def add_to_cart(self):
        """加入购物车"""
        product_id = random.choice(PRODUCT_IDS)
        self.client.post("/api/cart/items", json={
            "product_id": product_id,
            "quantity": random.randint(1, 5)
        })
    
    @task(1)
    def create_order(self):
        """创建订单"""
        with self.client.post(
            "/api/orders",
            json={
                "items": [
                    {
                        "product_id": random.choice(PRODUCT_IDS),
                        "quantity": random.randint(1, 3)
                    }
                ],
                "address_id": random.randint(1, 10)
            },
            catch_response=True
        ) as response:
            if response.status_code == 200:
                data = response.json()
                if data.get("code") == 0:
                    response.success()
                else:
                    response.failure(f"下单失败: {data.get('message')}")
            else:
                response.failure(f"HTTP 错误: {response.status_code}")
    
    @task(2)
    def view_orders(self):
        """查看订单列表"""
        self.client.get("/api/orders", params={"page": 1, "size": 10})
    
    def on_stop(self):
        """用户停止时执行退出"""
        if self.token:
            self.client.post("/api/auth/logout")


class EcommerceUser(HttpUser):
    """电商网站用户"""
    tasks = [UserLoginTask]
    wait_time = between(2, 5)
    host = "https://api.ecommerce.example.com"

11.3 负载模式配置

from locust import LoadTestShape


class EcommerceLoadShape(LoadTestShape):
    """
    电商网站负载模式：
    - 0-5 分钟：预热阶段，从 0 逐步增加到 1000 用户
    - 5-20 分钟：稳定运行阶段，保持 1000 用户
    - 20-25 分钟：峰值测试，增加到 3000 用户
    - 25-30 分钟：逐步下降，回到 1000 用户
    - 30-35 分钟：收尾阶段，逐步减少到 0
    """
    
    stages = [
        (300, 1000, 100),    # 5 分钟，1000 用户
        (1200, 1000, 100),   # 20 分钟，1000 用户
        (1500, 3000, 200),   # 25 分钟，3000 用户
        (1800, 1000, 100),   # 30 分钟，1000 用户
        (2100, 0, 50),       # 35 分钟，0 用户
    ]
    
    def tick(self):
        run_time = self.get_run_time()
        
        for end_time, user_count, spawn_rate in self.stages:
            if run_time < end_time:
                return (user_count, spawn_rate)
        
        return None

11.4 运行测试

# 单节点测试
locust -f ecommerce_test.py

# 分布式测试
# 启动 master
locust -f ecommerce_test.py --master

# 启动多个 worker
locust -f ecommerce_test.py --worker --master-host=localhost
locust -f ecommerce_test.py --worker --master-host=localhost

---

12. 避坑小贴士

12.1 常见错误与解决方案

问题 1：Locust 启动后没有请求发出

原因：User 类没有正确设置 tasks 属性或使用了 @task 装饰器但没有继承 HttpUser。

解决：

# 正确写法
from locust import HttpUser, task

class CorrectUser(HttpUser):  # 必须继承 HttpUser
    @task
    def my_task(self):
        self.client.get("/")

问题 2：并发数很高但 RPS 很低

原因：等待时间设置过长，或者服务器响应慢导致用户被阻塞。

解决：

• 检查 wait_time 设置是否合理
• 检查服务器响应时间
• 考虑增加 worker 节点

问题 3：分布式测试时 worker 无法连接 master

原因：防火墙阻止了端口通信，或 master-bind-host 设置不正确。

解决：

# 启动 master 时绑定到所有接口
locust -f locustfile.py --master --master-bind-host=0.0.0.0

# worker 使用正确的 master 地址
locust -f locustfile.py --worker --master-host=192.168.1.100

问题 4：内存使用过高

原因：在 User 类中存储了大量数据，或请求响应数据没有释放。

解决：

• 避免在 User 实例中缓存大量数据
• 使用 catch_response=True 后确保正确处理响应
• 考虑减少并发用户数或增加机器内存

12.2 性能测试注意事项

1. 不要对生产环境进行压力测试，除非经过充分评估和授权
2. 测试数据要足够分散，避免缓存导致结果失真
3. 关注 95% 响应时间而非平均值，平均值容易掩盖性能问题
4. 监控系统资源，确保测试机本身不会成为瓶颈
5. 多次测试取平均值，单次测试结果可能有偶然性

---

13. 本章小结

本章我们深入学习了性能测试的核心概念和实践方法：

核心概念：

• 理解了负载测试、压力测试、稳定性测试和容量测试的区别
• 掌握了响应时间、吞吐量、并发数、错误率等关键性能指标

Locust 框架：

• 学会了安装和配置 Locust 环境
• 掌握了 User 类和 Task 的定义方法
• 理解了 Locust 基于协程的高并发架构

实战技能：

• 能够编写 HTTP 请求，处理认证和自定义请求头
• 掌握了多种负载模式的配置方法
• 学会了收集和分析性能指标
• 了解了分布式测试的架构和配置

最佳实践：

• 通过电商网站实战案例，掌握了复杂场景的性能测试方法
• 了解了性能测试的常见陷阱和解决方案

---

14. 课后练习

1. 基础练习：编写一个 Locust 测试脚本，模拟用户访问博客网站的首页、文章详情页和搜索功能。

2. 进阶练习：为一个 REST API 设计性能测试场景，包含登录、CRUD 操作，要求使用 SequentialTaskSet 保证操作顺序。

3. 实战练习：使用 LoadTestShape 实现一个自定义负载模式，模拟一天内的流量变化（早高峰、午休、晚高峰）。

4. 思考题：

   • 如何确定性能测试的并发用户数？
   • 响应时间突然增加可能是什么原因？
   • 分布式测试时如何保证测试数据的一致性？

---

15. 下一篇预告

【Python 测试开发】第10讲 | 接口自动化测试框架封装 - 构建企业级测试平台

在下一讲中，我们将学习如何：

• 封装通用的 HTTP 请求工具类
• 设计数据驱动的测试框架
• 实现测试报告自动生成
• 集成 Allure 生成美观的测试报告
• 构建完整的接口自动化测试平台

敬请期待！

---

版权声明：本文为《Python 测试开发》系列教程的一部分，转载请注明出处。

关于作者：专注于 Python 测试开发领域，致力于帮助测试工程师提升技术能力。

---

如果本文对你有帮助，欢迎点赞、收藏、评论交流！