Locust 配置
基础 locustfile.py
from locust import HttpUser, task, between
class WebsiteUser(HttpUser):
wait_time = between(1, 3)
host = "https://api.example.com"
def on_start(self):
response = self.client.post("/auth/login", json={
"username": "testuser",
"password": "testpass",
})
token = response.json()["token"]
self.client.headers.update({"Authorization": f"Bearer {token}"})
@task(3) # 权重 3
def browse_products(self):
self.client.get("/products")
@task(1)
def view_product(self):
self.client.get("/products/1", name="/products/[id]")
@task(2)
def search(self):
self.client.get("/search?q=widget")
TaskSet — 分组行为
from locust import HttpUser, TaskSet, task, between
class UserBrowsingBehavior(TaskSet):
@task(5)
def view_listing(self):
self.client.get("/listings")
@task(1)
def create_listing(self):
self.client.post("/listings", json={"title": "测试"})
class BrowsingUser(HttpUser):
tasks = [UserBrowsingBehavior]
wait_time = between(0.5, 2)
运行 Locust
# Web UI 模式（打开 http://localhost:8089）
locust -f locustfile.py
# 无头模式
locust -f locustfile.py \
--headless \
--users 100 \
--spawn-rate 10 \
--run-time 2m \
--host https://api.example.com
# 导出 CSV 结果
locust -f locustfile.py \
--headless -u 50 -r 5 -t 1m \
--csv results/run1 \
--html results/report.html
自定义负载形状
from locust import LoadTestShape
class StepLoadShape(LoadTestShape):
step_time = 30
step_load = 20
spawn_rate = 10
time_limit = 300
def tick(self):
run_time = self.get_run_time()
if run_time > self.time_limit:
return None
current_step = run_time // self.step_time
return (current_step * self.step_load, self.spawn_rate)
分布式测试
# 主节点
locust -f locustfile.py --master --host https://api.example.com
# 工作节点（在独立机器上运行）
locust -f locustfile.py --worker --master-host=192.168.1.100
# 同机多 worker
locust -f locustfile.py --worker --master-host=localhost &
locust -f locustfile.py --worker --master-host=localhost &
事件与自定义断言
from locust import events
@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
stats = environment.stats.total
print(f"总请求数: {stats.num_requests}")
print(f"失败率: {stats.fail_ratio:.1%}")
if stats.fail_ratio > 0.01:
environment.process_exit_code = 1