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