头像 淇迹时刻
头像
淇迹时刻
-- : -- : --
切换主题色
-- : -- : --
性能测试 性能测试 +1 Locust 1125 字 5 分钟 -- 次

第二阶段：Locust压测框架

最后更新于:

第二阶段：Locust压测框架

一、Locust简介

1.1 什么是Locust？

Locust是一个开源的分布式性能测试工具，使用Python编写。它允许你通过编写简单的Python脚本来定义用户行为，然后模拟成千上万的并发用户来测试你的系统。

1.2 Locust特点

  • Python脚本：使用Python定义用户行为，灵活易用
  • 分布式压测：支持多台机器分布式压测
  • 实时监控：提供Web界面实时监控压测状态
  • 可扩展性：易于扩展和自定义
  • WebSocket支持：原生支持WebSocket协议

1.3 安装Locust

1# 安装Locust
2pip install locust
3
4# 验证安装
5locust --version

二、基础脚本编写

2.1 最简单的Locust脚本

 1from locust import HttpUser, task, between
 2
 3class WebsiteUser(HttpUser):
 4    wait_time = between(1, 5)
 5
 6    @task
 7    def index_page(self):
 8        self.client.get("/")
 9
10    @task(3)
11    def view_item(self):
12        for item_id in range(10):
13            self.client.get(f"/item?id={item_id}", name="/item")
14            self.wait()

2.2 脚本结构说明

组件说明
HttpUser用户类，继承自HttpUser
wait_time任务间的等待时间，使用between(min, max)
@task装饰器，标记任务方法
@task(n)任务权重，数值越大执行频率越高
self.clientHTTP客户端，用于发送请求

2.3 HTTP请求方法

 1# GET请求
 2response = self.client.get("/api/users")
 3
 4# POST请求
 5response = self.client.post("/api/users", json={"name": "test"})
 6
 7# PUT请求
 8response = self.client.put("/api/users/1", json={"name": "updated"})
 9
10# DELETE请求
11response = self.client.delete("/api/users/1")
12
13# 请求带参数
14response = self.client.get("/api/users", params={"page": 1, "size": 10})
15
16# 请求带headers
17response = self.client.get("/api/users", headers={"Authorization": "Bearer token"})

2.4 响应处理

 1# 检查响应状态码
 2response = self.client.get("/api/users")
 3if response.status_code == 200:
 4    print("请求成功")
 5
 6# 检查响应内容
 7response = self.client.get("/api/users/1")
 8data = response.json()
 9print(f"用户ID: {data['id']}")
10
11# 使用catch_response捕获失败
12with self.client.get("/api/users/999", catch_response=True) as response:
13    if response.status_code == 404:
14        response.success()
15    else:
16        response.failure("Expected 404")

三、高级场景

3.1 任务集（TaskSet）

 1from locust import HttpUser, TaskSet, task, between
 2
 3class UserTasks(TaskSet):
 4    @task
 5    def index(self):
 6        self.client.get("/")
 7
 8    @task
 9    def profile(self):
10        self.client.get("/profile")
11
12    @task
13    def logout(self):
14        self.client.post("/logout")
15        self.interrupt()
16
17class WebsiteUser(HttpUser):
18    tasks = [UserTasks]
19    wait_time = between(1, 3)

3.2 用户状态管理

 1class WebsiteUser(HttpUser):
 2    wait_time = between(1, 3)
 3
 4    def on_start(self):
 5        # 用户开始时执行
 6        response = self.client.post("/login", json={
 7            "username": "test",
 8            "password": "123456"
 9        })
10        self.token = response.json()["token"]
11
12    def on_stop(self):
13        # 用户结束时执行
14        self.client.post("/logout", headers={
15            "Authorization": f"Bearer {self.token}"
16        })
17
18    @task
19    def get_profile(self):
20        self.client.get("/profile", headers={
21            "Authorization": f"Bearer {self.token}"
22        })

3.3 WebSocket支持

 1from locust import User, task, between
 2from locust.contrib.fasthttp import FastHttpUser
 3import websocket
 4
 5class WebSocketUser(User):
 6    wait_time = between(1, 3)
 7
 8    def on_start(self):
 9        self.ws = websocket.create_connection("ws://localhost:8080/ws")
10
11    @task
12    def send_message(self):
13        self.ws.send('{"action": "ping"}')
14        response = self.ws.recv()
15        print(f"Received: {response}")
16
17    def on_stop(self):
18        self.ws.close()

四、分布式压测

4.1 启动Master节点

1# 在主节点启动
2locust -f locustfile.py --master --master-host=192.168.1.100

4.2 启动Worker节点

1# 在多个工作节点启动
2locust -f locustfile.py --worker --master-host=192.168.1.100

4.3 分布式架构

 1                    ┌─────────────────┐
 2                    │   Master Node   │
 3                    │  (Web Interface)│
 4                    └────────┬────────┘
 5                             │
 6          ┌──────────────────┼──────────────────┐
 7          │                  │                  │
 8    ┌─────▼─────┐    ┌──────▼──────┐    ┌──────▼──────┐
 9    │ Worker 1  │    │  Worker 2   │    │  Worker N   │
10    │ (模拟用户) │    │  (模拟用户)  │    │  (模拟用户)  │
11    └─────┬─────┘    └──────┬──────┘    └──────┬──────┘
12          │                  │                  │
13          └──────────────────┼──────────────────┘
14                             │
15                    ┌────────▼────────┐
16                    │  目标系统        │
17                    └─────────────────┘

五、命令行参数

5.1 常用参数

 1# 指定脚本文件
 2locust -f locustfile.py
 3
 4# 指定主机
 5locust -f locustfile.py --host=http://localhost:8080
 6
 7# 指定用户数和每秒启动数
 8locust -f locustfile.py --host=http://localhost:8080 -u 1000 -r 100
 9
10# 无Web界面模式
11locust -f locustfile.py --host=http://localhost:8080 -u 1000 -r 100 --headless
12
13# 指定运行时间
14locust -f locustfile.py --host=http://localhost:8080 -u 1000 -r 100 --headless -t 1h
15
16# 生成HTML报告
17locust -f locustfile.py --host=http://localhost:8080 -u 1000 -r 100 --headless -t 1h --html=report.html

5.2 参数说明

参数说明
-f指定Locust脚本文件
--host指定目标主机
-u指定并发用户数
-r指定每秒启动用户数
--headless无Web界面模式
-t指定运行时间
--html生成HTML报告
--master启动Master节点
--worker启动Worker节点

六、实战练习

6.1 练习1：编写基础脚本

编写一个Locust脚本，测试以下接口：

  • GET /api/users - 获取用户列表
  • POST /api/users - 创建用户
  • GET /api/users/{id} - 获取单个用户
  • PUT /api/users/{id} - 更新用户
  • DELETE /api/users/{id} - 删除用户

6.2 练习2：用户行为模拟

模拟一个电商网站的用户行为：

  1. 访问首页
  2. 搜索商品
  3. 查看商品详情
  4. 加入购物车
  5. 下单购买

6.3 练习3：分布式压测

搭建分布式压测环境：

  1. 启动1个Master节点
  2. 启动3个Worker节点
  3. 模拟1000并发用户

下一章：03-JMeter实战.md

📡
👤 作者: 阿海
🔗 链接: https://zruoqi.cn/tech/test/performance/locust/
🌐 版权: 本站文章除特别声明外，均采用 CC BY-NC-SA 4.0 协议，转载请注明来自 阿海 Blog！
最新文章