1. 初识Locust：这个压测工具为何被开发者追捧？
第一次听说Locust是在一个电商项目的性能优化会议上。当时我们的系统在促销活动时频繁崩溃，传统的JMeter压测脚本难以模拟真实用户行为，直到团队里一位资深架构师扔出一句："试试Locust吧，用Python写压测脚本那种"。三天后，我们不仅找到了系统瓶颈，还意外发现了数据库连接池的配置缺陷。
Locust是一款开源的负载测试工具，它最大的特点是允许你用纯Python代码定义用户行为。与传统的XML配置式压测工具不同，Locust的测试场景就是普通的Python文件，这意味着你可以：
- 使用任何Python库（比如requests、aiohttp）
- 自由组合复杂的业务流
- 直接复用生产环境的认证逻辑
- 甚至连接真实数据库做数据校验
2. 核心架构解析：为什么选择Locust而不是JMeter？
2.1 事件驱动架构
Locust基于gevent库实现协程并发，单个进程就能模拟数千用户。我做过对比测试：在8核16G的机器上：
- JMeter 5000线程时CPU占用率达90%
- Locust 5000用户时CPU仅占用35% 这是因为JMeter用线程模型，而Locust使用协程切换：
from locust import HttpUser, task class WebsiteUser(HttpUser): @task def login(self): self.client.post("/login", json={ "username": "test", "password": "123456" })2.2 分布式压测能力
上周对某政务系统做压力测试时，我们用10台4核机器模拟了10万并发用户。Locust的master-worker模式特别简单：
- 启动master：
locust -f stress_test.py --master - 在每个worker节点执行：
locust -f stress_test.py --worker --master-host=192.168.1.100
重要提示：确保所有worker节点的Python环境和依赖库完全一致，否则会出现不可预知的错误
3. 实战技巧：从零构建电商压测场景
3.1 用户行为建模
模拟用户浏览商品→加入购物车→下单的完整流程：
class ShoppingUser(HttpUser): wait_time = between(1, 3) # 用户操作间隔 @task(3) def browse(self): self.client.get("/products") @task(2) def add_to_cart(self): product_id = random.choice(PRODUCT_IDS) self.client.post(f"/cart/{product_id}") @task(1) def checkout(self): self.client.post("/orders", json={ "items": [{"id": 101, "qty": 2}], "payment": "credit_card" })3.2 参数化技巧
真实场景需要动态参数，比如不同用户登录不同账号：
USER_CREDENTIALS = [ ("user1@test.com", "pass123"), ("user2@test.com", "pass456") ] class AuthUser(HttpUser): def on_start(self): email, pwd = random.choice(USER_CREDENTIALS) self.client.headers = { "Authorization": f"Bearer {self.login(email, pwd)}" } def login(self, email, password): resp = self.client.post("/auth", json={ "email": email, "password": password }) return resp.json()["token"]4. 高级功能：让你的压测更专业
4.1 自定义统计数据
在金融行业项目中，我们需要监控90%响应时间而非平均值：
from locust import events @events.request.add_listener def track_percentile(request_type, name, response_time, response_length, **kw): if name == "/api/payment": store_response_time(response_time) # 自定义存储逻辑4.2 测试数据准备
压测前自动生成测试数据：
from locust import TaskSet class AdminTasks(TaskSet): @task def setup_test_data(self): for i in range(100): self.client.post("/admin/products", json={ "name": f"测试商品_{i}", "price": random.randint(10, 100) })5. 性能优化：如何榨干Locust的最后一丝性能
5.1 连接池配置
高并发下TCP连接复用至关重要：
from urllib3 import PoolManager class CustomUser(HttpUser): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.client._session = PoolManager( maxsize=100, block=True, timeout=30.0 )5.2 避免常见陷阱
- 禁用重试机制：
self.client.get("/api", allow_redirects=False) - 关闭SSL验证（仅测试环境）：
self.client.verify = False - 设置合理超时：
self.client.timeout = (3.0, 10.0)
6. 可视化监控：Web UI与Prometheus集成
启动Locust时带上--web-host参数后，访问http://localhost:8089 可以看到：
- 实时RPS曲线
- 响应时间分布
- 失败请求统计
对于需要长期监控的场景，可以启用Prometheus exporter：
locust -f stress.py --prometheus-port 9090然后在Grafana中导入Locust仪表盘模板，效果比原生UI更专业。
7. 企业级实践：我们的千亿级流量压测方案
去年双11前，我们使用Locust+K8s进行了全链路压测：
- 用Helm部署Locust集群
- 通过HPA自动扩展worker节点
- 压测结果自动写入Elasticsearch
- 异常请求录制回放功能
关键配置示例：
# values.yaml worker: replicas: 50 resources: limits: cpu: 2 memory: 4Gi env: - name: TEST_ENV value: staging这个方案最终帮助我们发现了支付网关的限流缺陷，避免了当天可能出现的重大故障。
