1. Locust性能测试框架概述

第一次接触Locust时，我被它的简洁性惊艳到了。这个用Python编写的开源工具，不需要复杂的配置就能快速生成高并发压力。与传统的JMeter不同，Locust的并发机制基于gevent协程，单机就能轻松模拟上万用户。

Locust的核心优势在于：

• 脚本即代码：用纯Python编写测试场景，比XML配置直观得多
• 分布式扩展：支持多机协同压测，轻松模拟百万级并发
• 实时监控：内置Web UI直观展示RPS、响应时间等关键指标
• 协议无关：除了HTTP，还能测试WebSocket、gRPC等各类协议

我在电商项目中使用Locust做全链路压测时，发现它的资源消耗只有JMeter的1/3。比如模拟1万并发用户时，JMeter需要8核16G的机器，而Locust只需4核8G。

2. 环境搭建与快速入门

2.1 安装Locust

推荐使用Python 3.7+环境，通过pip一键安装：

pip install locust

验证安装是否成功：

locust -V
# 预期输出：locust 2.15.1

常见问题解决：

• 如果遇到gevent安装失败，先升级pip：python -m pip install --upgrade pip
• Windows用户可能需要安装Microsoft C++ Build Tools

2.2 第一个测试脚本

创建一个locustfile.py：

from locust import HttpUser, task, between

class QuickstartUser(HttpUser):
    wait_time = between(1, 2.5)
    
    @task
    def hello_world(self):
        self.client.get("/")
        
    @task(3)
    def view_items(self):
        for item_id in range(10):
            self.client.get(f"/item?id={item_id}", name="/item")

这个脚本模拟用户：

1. 随机等待1-2.5秒
2. 有25%概率访问首页
3. 有75%概率浏览10个商品页

启动测试：

locust -f locustfile.py --host=http://your-test-site.com

3. 高级脚本编写技巧

3.1 用户行为建模

实际项目中，我们需要更真实的用户行为模拟。比如电商场景：

class ShoppingUser(HttpUser):
    wait_time = between(1, 3)
    
    def on_start(self):
        """ 用户登录 """
        self.client.post("/login", {
            "username": "test_user",
            "password": "123456"
        })
    
    @task(10)
    def browse_products(self):
        categories = ["electronics", "clothing", "home"]
        self.client.get(f"/category/{random.choice(categories)}")
        
    @task(2)
    def add_to_cart(self):
        product_id = random.randint(1, 1000)
        self.client.post("/cart", json={"product_id": product_id})
        
    @task(1)
    def checkout(self):
        self.client.get("/checkout")

3.2 参数化与数据驱动

使用CSV实现数据参数化：

import csv

class DataDrivenUser(HttpUser):
    def setup(self):
        with open("products.csv") as f:
            self.products = list(csv.DictReader(f))
    
    @task
    def view_product(self):
        product = random.choice(self.products)
        self.client.get(f"/product/{product['id']}")

3.3 自定义指标采集

通过事件钩子扩展监控：

from locust import events

@events.request.add_listener
def track_response_time(request_type, name, response_time, response_length, **kw):
    if response_time > 1000:  # 记录慢请求
        print(f"慢请求告警：{name} 耗时 {response_time}ms")

4. 分布式压测实战

4.1 单机多进程模式

启动主节点：

locust -f locustfile.py --master --expect-workers=4

启动4个工作进程：

locust -f locustfile.py --worker

4.2 多机分布式部署

主节点：

locust -f locustfile.py --master --expect-workers=8

从节点（在不同机器执行）：

locust -f locustfile.py --worker --master-host=192.168.1.100

4.3 性能优化建议

1. 资源监控：使用--csv=metrics导出数据结合Grafana展示
2. 心跳调优：通过--heartbeat-liveness=60调整worker超时时间
3. 网络优化：在云环境部署时，确保节点间网络延迟<50ms

5. 结果分析与问题定位

5.1 关键指标解读

• RPS：真实吞吐量，要区分成功/失败请求
• 响应时间：关注95分位值而非平均值
• 并发用户：活跃用户数 ≠ 请求并发数

5.2 常见性能瓶颈

1. 服务端：

   • CPU瓶颈：监控%CPU超过80%
   • 内存泄漏：观察RES内存持续增长
   • 线程阻塞：检查线程池等待队列

2. Locust端：

   • Worker负载不均：通过--csv检查各worker请求分布
   • 网络带宽不足：监控tx_bytes/sec接近带宽上限

5.3 调试技巧

启用详细日志：

locust --loglevel DEBUG

使用Py-Spy进行性能分析：

py-spy top --pid <locust_pid>

6. 进阶应用场景

6.1 测试非HTTP协议

自定义WebSocket客户端示例：

from websocket import create_connection
from locust import User, task, events

class WebSocketClient:
    def __init__(self, host):
        self.ws = create_connection(f"ws://{host}/chat")
    
    def send(self, message):
        start_time = time.time()
        try:
            self.ws.send(message)
            events.request_success.fire(
                request_type="websocket",
                name="send",
                response_time=int((time.time()-start_time)*1000),
                response_length=len(message)
            )
        except Exception as e:
            events.request_failure.fire(...)

class WebSocketUser(User):
    tasks = [ChatBehavior]
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.client = WebSocketClient(self.host)

6.2 与CI/CD集成

Jenkins Pipeline示例：

stage('Load Test') {
    steps {
        sh 'locust -f performance_test.py --headless -u 1000 -r 100 --run-time 30m'
        perfReport sourceDataFiles: 'locust_stats.csv'
    }
}

6.3 混合场景测试

模拟多类型用户：

class MobileUser(HttpUser):
    weight = 3
    tasks = [MobileBehavior]

class DesktopUser(HttpUser):
    weight = 1
    tasks = [DesktopBehavior]

7. 性能测试最佳实践

1. 渐进式加压：从低并发开始，逐步增加负载
2. 基准测试：每次发版前执行相同测试用例
3. 异常注入：模拟网络抖动、服务宕机等场景
4. 数据预热：确保测试数据库有足够数据量

实际项目中，我建议将性能测试分为三个阶段：

1. 冒烟测试：100并发验证基本功能
2. 容量测试：逐步加压至系统极限
3. 稳定性测试：80%峰值负载持续运行24小时

遇到过一个典型问题：某次大促前压测时，发现当并发达到5000时，服务响应时间从200ms陡增到5s。通过火焰图定位到是Redis连接池配置过小导致。这个案例让我深刻体会到渐进式加压的重要性。