Python脚本压力测试Locust怎么用

wen 实用脚本 2026-07-19 4

本文目录导读：

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

我来详细介绍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

注意事项

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

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