Locust Python 负载测试教程

Locust 是一款开源的 Python 负载测试工具，允许你用 Python 代码定义用户行为，模拟真实用户访问场景。它提供直观的 Web 界面实时展示测试进度，支持分布式测试横向扩展负载。本文将在搬瓦工 VPS 上从零开始搭建 Locust 测试环境并编写各类测试场景。

一、安装 Locust

# 确保 Python 3.7+ 已安装
python3 --version

# 使用 pip 安装
pip3 install locust

# 验证安装
locust --version

# 如果需要在虚拟环境中安装
python3 -m venv locust-env
source locust-env/bin/activate
pip install locust

二、编写第一个测试脚本

创建 locustfile.py：

cat > locustfile.py <<'EOF'
from locust import HttpUser, task, between

class WebsiteUser(HttpUser):
    # 模拟用户在每次请求之间等待 1-3 秒
    wait_time = between(1, 3)

    @task(3)  # 权重为 3，被选中的概率更高
    def view_homepage(self):
        self.client.get("/")

    @task(2)
    def view_about(self):
        self.client.get("/about")

    @task(1)
    def view_api(self):
        with self.client.get("/api/status", catch_response=True) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"Status code: {response.status_code}")
EOF

三、运行测试

3.1 Web 界面模式

# 启动 Locust（默认 Web 界面在 8089 端口）
locust -f locustfile.py --host=http://localhost

# 指定端口
locust -f locustfile.py --host=http://localhost --web-port=8089

# 允许外部访问 Web 界面
locust -f locustfile.py --host=http://localhost --web-host=0.0.0.0

启动后通过浏览器访问 http://VPS_IP:8089，设置并发用户数和每秒增长速率后开始测试。

3.2 命令行模式（无界面）

# 无界面运行：100 用户，每秒增加 10 个，运行 5 分钟
locust -f locustfile.py --host=http://localhost \
  --headless -u 100 -r 10 --run-time 5m

# 输出 CSV 报告
locust -f locustfile.py --host=http://localhost \
  --headless -u 100 -r 10 --run-time 5m \
  --csv=results/test

四、高级测试场景

4.1 POST 请求与 JSON 数据

cat > api_test.py <<'EOF'
from locust import HttpUser, task, between
import json

class APIUser(HttpUser):
    wait_time = between(1, 2)

    def on_start(self):
        """用户启动时执行（如登录）"""
        response = self.client.post("/api/login", json={
            "username": "testuser",
            "password": "testpass"
        })
        if response.status_code == 200:
            self.token = response.json().get("token")

    @task
    def get_profile(self):
        self.client.get("/api/profile", headers={
            "Authorization": f"Bearer {self.token}"
        })

    @task
    def create_post(self):
        self.client.post("/api/posts", json={
            "title": "Test Post",
            "content": "This is a load test post"
        }, headers={
            "Authorization": f"Bearer {self.token}"
        })

    def on_stop(self):
        """用户退出时执行"""
        self.client.post("/api/logout")
EOF

4.2 多用户角色模拟

cat > multi_role.py <<'EOF'
from locust import HttpUser, task, between

class AdminUser(HttpUser):
    weight = 1  # 10% 的用户是管理员
    wait_time = between(2, 5)

    @task
    def view_dashboard(self):
        self.client.get("/admin/dashboard")

    @task
    def view_reports(self):
        self.client.get("/admin/reports")

class NormalUser(HttpUser):
    weight = 9  # 90% 的用户是普通用户
    wait_time = between(1, 3)

    @task(5)
    def browse_products(self):
        self.client.get("/products")

    @task(2)
    def search(self):
        self.client.get("/search?q=test")

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

4.3 顺序任务流

cat > sequential.py <<'EOF'
from locust import HttpUser, task, between, SequentialTaskSet

class ShoppingFlow(SequentialTaskSet):
    """模拟完整的购物流程"""

    @task
    def browse(self):
        self.client.get("/products")

    @task
    def view_product(self):
        self.client.get("/products/1")

    @task
    def add_to_cart(self):
        self.client.post("/cart/add", json={"product_id": 1, "quantity": 1})

    @task
    def checkout(self):
        self.client.post("/checkout", json={"payment_method": "card"})
        self.interrupt()  # 完成后中断，重新开始

class ShoppingUser(HttpUser):
    wait_time = between(1, 3)
    tasks = [ShoppingFlow]
EOF

五、分布式测试

当单台机器无法产生足够负载时，可以使用分布式模式：

# 主节点（Master）
locust -f locustfile.py --master --host=http://target-server

# 工作节点（Worker）- 在其他机器上运行
locust -f locustfile.py --worker --master-host=MASTER_IP

# 指定工作节点数量的无界面模式
locust -f locustfile.py --master --headless \
  -u 1000 -r 50 --run-time 10m \
  --expect-workers=3

六、测试结果分析

Locust 的 Web 界面提供实时图表，命令行模式可以输出 CSV 报告：

# 生成 HTML 报告
locust -f locustfile.py --host=http://localhost \
  --headless -u 100 -r 10 --run-time 5m \
  --html=report.html

# CSV 报告会生成以下文件：
# results_stats.csv       - 请求统计
# results_stats_history.csv - 时间序列数据
# results_failures.csv    - 失败详情
# results_exceptions.csv  - 异常信息

关注的核心指标：

• RPS（Requests Per Second）：系统每秒处理的请求数。
• Response Time（P50/P95/P99）：不同百分位的响应时间。
• Failure Rate：请求失败率，理想值为 0%。
• Current Users：当前并发用户数。

七、性能调优建议

# 在运行 Locust 的机器上调整系统限制
ulimit -n 65535

# 调整内核参数以支持大量连接
sysctl -w net.ipv4.tcp_tw_reuse=1
sysctl -w net.core.somaxconn=65535
sysctl -w net.ipv4.ip_local_port_range="1024 65535"

总结

Locust 的 Python 编程方式让测试脚本非常灵活，特别适合模拟复杂的用户行为场景。对于简单的基准测试可以使用 wrk，对于更现代化的测试框架可以使用 K6。性能瓶颈定位可以借助 Perf 和 火焰图 工具。选购搬瓦工 VPS 请查看 全部方案，购买时使用优惠码 NODESEEK2026 可享受 6.77% 的折扣，通过 bwh81.net 访问官网。