第三方软件测评机构:【Locust的性能测试和负载测试】
Locust是一个用Python编写的开源负载测试工具，它允许使用Python代码定义用户行为，能够模拟数百万用户并发来测试您的系统。
Locust的优势：使用纯 Python 编写测试场景、分布式支持轻松扩展到多台机器、实时Web UI更直观监控测试进度和结果、高并发能力是基于gevent的协程实现
1.安装与配置
基础安装
# 使用 pip 安装
pip install locust# 验证安装
locust --version选择依赖项
# 安装带额外功能的版本
pip install locust[extra]# 或单独安装所需组件
pip install pyzmq # 分布式模式支持
pip install geventhttpclient # 更快的 HTTP 客户端2. 脚本编写
创建基础测试脚本-locustfile.py
from locust import HttpUser, TaskSet, task, between
import randomclass UserBehavior(TaskSet):@task(3) # 权重为3，执行频率更高def view_products(self):# 查看产品列表self.client.get("/api/products")@task(2)def view_product_detail(self):# 查看产品详情，使用随机IDproduct_id = random.randint(1, 100)self.client.get(f"/api/products/{product_id}")@task(1)def create_order(self):# 创建订单headers = {'Content-Type': 'application/json'}data = {"product_id": random.randint(1, 100),"quantity": random.randint(1, 5)}self.client.post("/api/orders", json=data, headers=headers)class WebsiteUser(HttpUser):tasks = [UserBehavior]wait_time = between(1, 5) # 用户等待时间1-5秒def on_start(self):# 用户启动时执行（如登录）self.client.post("/api/login", json={"username": "test_user","password": "test_pass"})3. 功能配置
自定义客户端配置
from locust import HttpUser, task
import jsonclass ApiUser(HttpUser):host = "https://api.zmtests.com"def __init__(self, *args, **kwargs):super().__init__(*args, **kwargs)self.client.verify = False # 禁用SSL验证self.client.timeout = 30 # 设置超时时间@taskdef complex_request(self):# 自定义请求头headers = {"Authorization": "Bearer token123","User-Agent": "LocustLoadTest/1.0"}# 带参数的GET请求params = {"page": 1, "limit": 20}response = self.client.get("/api/data", params=params, headers=headers)# 处理响应if response.status_code == 200:data = response.json()print(f"Received {len(data)} items")事件钩子编写
from locust import events
from locust.runners import MasterRunner, WorkerRunner
import time@events.test_start.add_listener
def on_test_start(environment, **kwargs):print("测试开始执行")if isinstance(environment.runner, MasterRunner):print("这是Master节点")elif isinstance(environment.runner, WorkerRunner):print("这是Worker节点")@events.request.add_listener
def on_request(request_type, name, response_time, response_length, response, context, exception, **kwargs):if exception:print(f"请求失败: {name}, 错误: {exception}")else:print(f"请求成功: {name}, 响应时间: {response_time}ms")@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):print("测试结束")4. 运行
启动方式
Web UI 模式
# 基础启动
locust -f locustfile.py# 指定主机和端口
locust -f locustfile.py --host=https://api.zmtests.com --web-port=8080# 无Web界面模式（用于CI/CD）
locust -f locustfile.py --headless -u 100 -r 10 -t 10m命令行参数
locust -f locustfile.py \--host=https://api.zmtests.com \--users=1000 \ # 最大用户数--spawn-rate=10 \ # 每秒启动用户数--run-time=30m \ # 运行时间--web-host=0.0.0.0 \ # Web界面绑定地址--web-port=8080 \ # Web界面端口--headless \ # 无头模式--csv=results \ # 导出CSV结果--html=report.html # 生成HTML报告分布式运行
# 启动Master节点
locust -f locustfile.py --master --expect-workers=4# 启动Worker节点（在其他机器上）
locust -f locustfile.py --worker --master-host=192.168.1.100
5. 性能优化
优化测试脚本
from locust import HttpUser, task, events
import gevent
from gevent.pool import Groupclass OptimizedUser(HttpUser):@taskdef parallel_requests(self):# 并行执行多个请求urls = ["/api/products","/api/categories", "/api/users/me"]group = Group()for url in urls:group.spawn(lambda u=url: self.client.get(u))group.join()@taskdef batch_operations(self):# 批量操作减少请求次数orders = [{"product_id": i, "quantity": 1} for i in range(1, 6)]self.client.post("/api/orders/batch", json=orders)自定义客户端
from locust import HttpUser
from locust.clients import HttpSession
import urllib3class CustomHttpUser(HttpUser):def __init__(self, *args, **kwargs):super().__init__(*args, **kwargs)# 禁用不安全请求警告urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)@taskdef custom_request(self):# 使用连接池和会话复用with self.client.get("/api/data", catch_response=True) as response:if response.elapsed.total_seconds() > 2:response.failure("响应时间过长")elif response.status_code != 200:response.failure(f"HTTP错误: {response.status_code}")else:response.success()6. 数据驱动测试
import csv
import json
from locust import HttpUser, taskclass DataDrivenUser(HttpUser):def __init__(self, *args, **kwargs):super().__init__(*args, **kwargs)self.test_data = self.load_test_data()def load_test_data(self):# 从CSV文件加载测试数据data = []with open('test_data.csv', 'r') as f:reader = csv.DictReader(f)for row in reader:data.append(row)return data@taskdef create_user(self):if self.test_data:user_data = random.choice(self.test_data)self.client.post("/api/users", json=user_data)7. 验证
from locust import HttpUser, task
import jsonschema
from jsonschema import validateclass ValidatingUser(HttpUser):user_schema = {"type": "object","properties": {"id": {"type": "integer"},"name": {"type": "string"},"email": {"type": "string", "format": "email"}},"required": ["id", "name", "email"]}@taskdef validate_response(self):with self.client.get("/api/user/1", catch_response=True) as response:try:data = response.json()validate(instance=data, schema=self.user_schema)# 自定义业务逻辑验证if data["name"] and len(data["name"]) > 0:response.success()else:response.failure("用户名无效")except jsonschema.ValidationError as e:response.failure(f"响应格式错误: {e}")except ValueError:response.failure("响应不是有效的JSON")8. 集成与自动化
FROM python:3.9
RUN pip install locust
COPY locustfile.py /mnt/locustfile.py
WORKDIR /mnt
EXPOSE 8089
CMD ["locust", "-f", "locustfile.py", "--host", "http://target-service"]CI/CD 集成
#yaml
name: Load Test
on: [push]
jobs:load-test:runs-on: ubuntu-lateststeps:- uses: actions/checkout@v2- name: Setup Pythonuses: actions/setup-python@v2with:python-version: '3.9'- name: Install dependenciesrun: pip install locust- name: Run load testrun: |locust -f locustfile.py \--headless \--users 100 \--spawn-rate 10 \--run-time 5m \--host ${{ secrets.TARGET_HOST }} \--html report.html \--only-summary