Python脚本压力测试Locust怎么用

本文目录导读：

1. 安装Locust
2. 基础用法
3. 运行压力测试
4. 高级功能
5. 数据监控与导出
6. 实战示例：API压力测试
7. 最佳实践
8. 注意事项

我来详细介绍Locust的使用方法,这是一个非常流行的Python性能测试工具。

安装Locust

pip install locust

基础用法

最简单的压力测试脚本

创建一个 locustfile.py 文件：

from locust import HttpUser, task, between
class WebsiteUser(HttpUser):
    # 设置思考时间（用户执行任务之间的等待时间）
    wait_time = between(1, 5)
    @task
    def index_page(self):
        self.client.get("/")
    @task(3)  # 权重：这个任务执行概率是其他任务的3倍
    def about_page(self):
        self.client.get("/about")

更完整的示例

from locust import HttpUser, task, between, tag
import json
class QuickstartUser(HttpUser):
    wait_time = between(0.5, 3)
    def on_start(self):
        """每个用户开始测试时执行"""
        # 登录操作
        response = self.client.post("/login", {
            "username": "test_user",
            "password": "test_pass"
        })
        if response.status_code == 200:
            print("Login successful")
    @task(2)
    def view_items(self):
        # 查看商品列表
        with self.client.get("/items", catch_response=True) as response:
            if response.status_code != 200:
                response.failure("Failed to load items")
            elif len(response.text) < 100:
                response.failure("Response too short")
    @task(1)
    @tag("create")  # 可以添加标签
    def create_item(self):
        # 创建新商品
        payload = {
            "name": "Test Item",
            "price": 99.99
        }
        headers = {"Content-Type": "application/json"}
        with self.client.post("/items", 
                             data=json.dumps(payload),
                             headers=headers, 
                             catch_response=True) as response:
            if response.status_code != 201:
                response.failure("Failed to create item")
    def on_stop(self):
        """每个用户结束时执行"""
        self.client.post("/logout")

运行压力测试

命令行方式

# 基础命令
locust -f locustfile.py
# 指定Web界面端口
locust -f locustfile.py --web-port 8089
# 无Web界面模式（Headless模式）
locust -f locustfile.py --headless -u 100 -r 10 --run-time 1m
# 参数说明：
# -u 100: 模拟100个虚拟用户
# -r 10: 每秒启动10个用户
# --run-time 1m: 运行1分钟
# --headless: 无界面模式

使用配置文件

创建 locust.conf 文件：

[locust]
locustfile = locustfile.py
host = http://example.com
users = 100
spawn-rate = 10
run-time = 5m
headless = true

运行：

locust --config locust.conf

高级功能

多用户行为模式

from locust import HttpUser, TaskSet, task, between
class UserBehavior(TaskSet):
    def on_start(self):
        """初始化操作"""
        self.login()
    def login(self):
        self.client.post("/login", {
            "username": "user",
            "password": "pass"
        })
    @task(2)
    def browse(self):
        self.client.get("/products")
    @task(1)
    def add_to_cart(self):
        self.client.post("/cart", {
            "product_id": 123,
            "quantity": 1
        })
class WebsiteUser(HttpUser):
    tasks = [UserBehavior]
    wait_time = between(1, 3)
    host = "http://example.com"  # 默认主机地址

自定义负载策略

from locust import HttpUser, task, constant, constant_pacing, constant_throughput
class CustomLoadUser(HttpUser):
    # 固定等待时间
    # wait_time = constant(1)
    # 固定吞吐量（每秒执行指定次数的任务）
    wait_time = constant_throughput(10)
    # 固定执行频率（每N秒执行一次）
    # wait_time = constant_pacing(5)
    @task
    def my_task(self):
        self.client.get("/api/endpoint")

数据监控与导出

启动Web界面

locust -f locustfile.py --web-host 0.0.0.0 --web-port 8089

访问 http://localhost:8089，可以：

• 实时查看请求数、响应时间、失败率
• 图表显示性能趋势
• 导出CSV报告

导出测试数据

# 导出CSV结果
locust -f locustfile.py --headless -u 100 -r 10 --run-time 1m \
       --csv=locust_report \
       --csv-full-history
# 这将生成：
# - locust_report_stats.csv
# - locust_report_stats_history.csv
# - locust_report_failures.csv
# - locust_report_exceptions.csv

实战示例：API压力测试

from locust import HttpUser, task, between
import random
import json
class APIUser(HttpUser):
    wait_time = between(1, 3)
    host = "http://api.example.com"
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.token = None
    def on_start(self):
        # 获取认证token
        response = self.client.post("/auth/token", {
            "api_key": "test_key"
        })
        if response.status_code == 200:
            self.token = response.json()["token"]
            self.headers = {
                "Authorization": f"Bearer {self.token}",
                "Content-Type": "application/json"
            }
    @task(3)
    def get_users(self):
        self.client.get("/api/v1/users", 
                       headers=self.headers,
                       name="获取用户列表")
    @task(2)
    def create_user(self):
        user_data = {
            "name": f"TestUser_{random.randint(1000, 9999)}",
            "email": f"test{random.randint(1000, 9999)}@example.com"
        }
        with self.client.post("/api/v1/users",
                             data=json.dumps(user_data),
                             headers=self.headers,
                             catch_response=True,
                             name="创建用户") as response:
            if response.status_code == 201:
                # 成功后立即删除
                user_id = response.json()["id"]
                self.client.delete(f"/api/v1/users/{user_id}",
                                 headers=self.headers,
                                 name="删除用户")
    @task(1)
    def update_user(self):
        self.client.put("/api/v1/users/1",
                       data=json.dumps({"status": "active"}),
                       headers=self.headers,
                       name="更新用户")

最佳实践

参数配置建议

from locust import HttpUser, task, between, constant_pacing
class OptimizedUser(HttpUser):
    # 设置合理的思考时间
    wait_time = constant_pacing(1)  # 每秒执行1次
    # 设置超时时间
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.client.timeout = 30  # 30秒超时
    @task
    def api_call(self):
        with self.client.get("/api/data", 
                           catch_response=True,
                           timeout=10) as response:
            if response.elapsed.total_seconds() > 2:
                response.failure(f"Response time too long: {response.elapsed.total_seconds()}s")

分布式测试

# 主节点
locust -f locustfile.py --master --expect-workers 2
# 工作节点（多个终端）
locust -f locustfile.py --worker --master-host=localhost

注意事项

1. 开始测试前：确认测试目标，设置合理的并发数和运行时间
2. 监控资源：注意被测试服务器的CPU、内存、网络等资源使用情况
3. 数据清理：测试中创建的数据需要及时清理
4. 渐进式测试：从低并发开始，逐步增加负载
5. 真实场景模拟：思考时间、用户行为要尽量接近真实情况

Locust的优势在于其Pythonic的使用方式、强大的分布式支持以及良好的实时监控界面，通过上述示例和配置，你可以快速上手并进行有效的压力测试。