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

pip install locust

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

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

# 安装 Locust
pip install locust

pip install -r requirements.txt

locust --version

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")

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"
                    }
                }
            )

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")

locust -f locustfile.py --master

locust -f locustfile.py --worker

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)

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)

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

locust -f locustfile.py --csv=results

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)

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"})

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"})