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 random
class UserBehavior(TaskSet):
@task(3) # 权重为3，执行频率更高
def view_products(self):
# 查看产品列表
self.client.get("/api/products")
@task(2)
def view_product_detail(self):
# 查看产品详情，使用随机ID
product_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 json
class 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 # 设置超时时间
@task
def 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 Group
class OptimizedUser(HttpUser):
@task
def 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()
@task
def 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 urllib3
class CustomHttpUser(HttpUser):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# 禁用不安全请求警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
@task
def 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, task
class 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
@task
def 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 validate
class ValidatingUser(HttpUser):
user_schema = {
"type": "object",
"properties": {
"id": {"type": "integer"},
"name": {"type": "string"},
"email": {"type": "string", "format": "email"}
},
"required": ["id", "name", "email"]
}
@task
def 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-latest
steps:
- uses: actions/checkout@v2
- name: Setup Python
uses: actions/setup-python@v2
with:
python-version: '3.9'
- name: Install dependencies
run: pip install locust
- name: Run load test
run: |
locust -f locustfile.py \
--headless \
--users 100 \
--spawn-rate 10 \
--run-time 5m \
--host ${{ secrets.TARGET_HOST }} \
--html report.html \
--only-summary