python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python Locust性能负载测试

Python Locust搭建高性能负载测试工具

作者：AIGC创想家

Locust 是一个开源的、基于 Python 的负载测试工具,它允许开发者使用 Python 代码来定义用户行为,从而模拟真实用户对系统进行压力测试,所以本文就来使用Locust搭建一个高性能负载测试工具吧

简介

Locust 是一个开源的、基于 Python 的负载测试工具，它允许开发者使用 Python 代码来定义用户行为，从而模拟真实用户对系统进行压力测试。Locust 以其简单易用、可扩展性强和实时监控等特点，成为了性能测试领域的首选工具之一。

为什么选择 Locust？

Python 代码驱动：使用 Python 编写测试脚本，灵活且易于维护

分布式支持：可以轻松扩展到多台机器进行大规模测试

实时 Web UI：提供直观的实时监控界面

可扩展性：支持自定义事件和指标

开源免费：完全开源，社区活跃

跨平台支持：支持 Windows、Linux、macOS 等主流操作系统

丰富的插件生态：支持多种扩展和插件

安装与配置

基本安装

pip install locust

使用虚拟环境（推荐）

# 创建虚拟环境
python -m venv locust_env

# 激活虚拟环境
# Windows
locust_env\Scripts\activate
# Linux/macOS
source locust_env/bin/activate

# 安装 Locust
pip install locust

依赖管理

创建 requirements.txt 文件：

locust>=2.15.1
requests>=2.31.0
gevent>=23.7.0

安装依赖：

pip install -r requirements.txt

验证安装

locust --version

基本概念

1. User 类

User 类定义了模拟用户的行为。每个用户实例代表一个并发用户。

2. Task

Task 是用户执行的具体操作，比如访问网页、提交表单等。

3. Wait Time

定义用户执行任务之间的等待时间，模拟真实用户行为。

4. 测试场景

测试场景定义了用户的行为序列和测试流程。

示例代码

基础示例

from locust import HttpUser, task, between

class WebsiteUser(HttpUser):
    wait_time = between(1, 5)  # 用户执行任务之间等待1-5秒
    
    @task
    def index_page(self):
        self.client.get("/")
    
    @task(3)  # 权重为3，表示这个任务被执行的概率是其他任务的3倍
    def view_items(self):
        self.client.get("/items")

高级示例：API 测试

from locust import HttpUser, task, between
import json

class APIUser(HttpUser):
    wait_time = between(1, 3)
    
    def on_start(self):
        self.token = self.login()
    
    def login(self):
        response = self.client.post("/api/login", 
            json={"username": "test", "password": "test123"})
        return response.json()["token"]
    
    @task
    def get_user_profile(self):
        headers = {"Authorization": f"Bearer {self.token}"}
        self.client.get("/api/profile", headers=headers)
    
    @task
    def update_user_settings(self):
        headers = {"Authorization": f"Bearer {self.token}"}
        data = {"theme": "dark", "notifications": True}
        self.client.put("/api/settings", 
            headers=headers, 
            json=data)

复杂场景示例：电商网站测试

from locust import HttpUser, task, between
import random

class EcommerceUser(HttpUser):
    wait_time = between(2, 5)
    
    def on_start(self):
        self.cart_items = []
    
    @task(2)
    def browse_products(self):
        category = random.choice(["electronics", "clothing", "books"])
        self.client.get(f"/products/{category}")
    
    @task
    def add_to_cart(self):
        product_id = random.randint(1, 100)
        response = self.client.post(
            "/cart/add",
            json={"product_id": product_id, "quantity": 1}
        )
        if response.status_code == 200:
            self.cart_items.append(product_id)
    
    @task
    def checkout(self):
        if self.cart_items:
            self.client.post(
                "/checkout",
                json={
                    "items": self.cart_items,
                    "payment_method": "credit_card",
                    "shipping_address": {
                        "street": "123 Test St",
                        "city": "Test City",
                        "country": "Test Country"
                    }
                }
            )

高级特性

1. 自定义事件

from locust import events
import time

@events.request.add_listener
def on_request(request_type, name, response_time, response_length, **kwargs):
    print(f"Request: {name}, Response Time: {response_time}ms")

@events.test_start.add_listener
def on_test_start(**kwargs):
    print("Test is starting")

@events.test_stop.add_listener
def on_test_stop(**kwargs):
    print("Test is stopping")

2. 分布式测试

启动主节点：

locust -f locustfile.py --master

启动工作节点：

locust -f locustfile.py --worker

3. 数据驱动测试

from locust import HttpUser, task, between
import csv
import random

class DataDrivenUser(HttpUser):
    wait_time = between(1, 3)
    
    def on_start(self):
        self.test_data = []
        with open('test_data.csv', 'r') as f:
            reader = csv.DictReader(f)
            self.test_data = list(reader)
    
    @task
    def test_with_data(self):
        data = random.choice(self.test_data)
        self.client.post("/api/endpoint", json=data)

4. 自定义指标收集

from locust import HttpUser, task, between
from locust.stats import RequestStats

class CustomMetricsUser(HttpUser):
    wait_time = between(1, 3)
    
    @task
    def custom_metric_task(self):
        with self.client.measure("custom_operation"):
            # 执行自定义操作
            time.sleep(0.1)

测试报告和数据分析

1. HTML 报告

locust -f locustfile.py --html=report.html

2. CSV 导出

locust -f locustfile.py --csv=results

3. 自定义报告

from locust import events
import json
import time

@events.test_stop.add_listener
def on_test_stop(**kwargs):
    stats = RequestStats.get_current()
    report = {
        "timestamp": time.time(),
        "total_requests": stats.total.num_requests,
        "failed_requests": stats.total.num_failures,
        "avg_response_time": stats.total.avg_response_time,
        "max_response_time": stats.total.max_response_time,
        "min_response_time": stats.total.min_response_time
    }
    
    with open("custom_report.json", "w") as f:
        json.dump(report, f)

与其他测试工具对比

特性LocustJMeterGatlingK6
编程语言PythonJavaScalaJavaScript
学习曲线低中中低
分布式支持是是是是
实时监控是是是是
报告生成基础丰富丰富丰富
社区活跃度高高中高

实际应用案例

1. 电商网站性能测试

from locust import HttpUser, task, between
import random

​​​​​​​class EcommerceLoadTest(HttpUser):
    wait_time = between(1, 3)
    
    @task(3)
    def browse_products(self):
        self.client.get("/products")
    
    @task(2)
    def search_products(self):
        query = random.choice(["laptop", "phone", "tablet"])
        self.client.get(f"/search?q={query}")
    
    @task(1)
    def checkout_process(self):
        self.client.post("/cart/add", json={"product_id": 1, "quantity": 1})
        self.client.get("/checkout")
        self.client.post("/payment", json={"method": "credit_card"})

2. API 性能测试

from locust import HttpUser, task, between
import json

class APILoadTest(HttpUser):
    wait_time = between(0.1, 0.5)
    
    @task
    def api_endpoints(self):
        # 测试多个 API 端点
        self.client.get("/api/v1/users")
        self.client.post("/api/v1/users", json={"name": "Test User"})
        self.client.get("/api/v1/users/1")
        self.client.put("/api/v1/users/1", json={"name": "Updated User"})

故障排除指南

1. 常见问题

内存使用过高

连接超时

测试数据问题

2. 性能优化建议

合理设置用户数

优化测试脚本

监控系统资源

错误处理

总结

Locust 是一个功能强大且灵活的负载测试工具，通过 Python 代码可以轻松实现复杂的测试场景。它的分布式特性和实时监控能力使其成为性能测试的理想选择。通过合理使用 Locust，可以帮助开发团队更好地评估系统性能，发现潜在问题，确保系统在高负载下的稳定性。

最佳实践总结

1.测试前准备

2.测试执行

3.结果分析

4.持续改进

到此这篇关于Python Locust搭建高性能负载测试工具的文章就介绍到这了,更多相关Python Locust性能负载测试内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:
阅读全文