1. Locust是什么
Locust是一种开源性能测试工具,它使用Python语言编写,支持使用简单的代码创建并发用户,并模拟这些用户对Web应用程序的负载进行压力测试。使用Locust,您可以轻松模拟数千个并发用户的操作,并在测试过程中测量Web应用程序的性能。
2. Locust特点
- 使用Python语言编写,易于学习和使用。
- 使用Master/Worker结构,支持分布式压力测试。
- 支持使用脚本编写测试用例,并支持HTTP、WebSockets、TCP等多种协议。
- 提供可视化的Web界面,可以实时监控测试性能和执行结果。
- 支持使用Python代码编写测试用例,灵活可控。
- 支持使用命令行或通过API接口对测试进行控制和管理。
3. Locust实现原理
3.1 核心类
HttpUser
:HttpUser是Locust中用于编写HTTP请求的用户类。通过继承HttpUser,您可以定义测试用户的行为和操作,例如发送HTTP GET/POST请求、等待时间、测试负载等。User
:User是Locust中的用户类,用于模拟网络协议之间的通信,例如 TCP 协议等,支持同时模拟多个用户的并发行为和操作。您可以使用User类的子类来定义多个用户,并在测试执行时模拟并发并发访问。SequentialTaskSet
/TaskSequence
/TaskSet
:TaskSequence是Locust中用于描述测试任务序列的基础类。通过继承TaskSequence,您可以定义一系列测试任务,以模拟用户操作的流程和顺序。TaskSet是TaskSequence的高级实现,支持更复杂的任务解决方案。SequentialTaskSet继承于 TaskSet,多个任务之间是串行执行的,即完成一个任务才会去执行下一个任务。具体区别参照以下样例。Event
:有时需要在运行的特定时间执行某些操作。针对这种需求,Locust 提供了事件钩子 (event hooks)。
3.2 常用方法
client.get(url, **kwargs)
:发送HTTP GET请求。client.post(url, data=None, json=None, **kwargs)
:发送HTTP POST请求。client.put(url, data=None, **kwargs)
:发送HTTP PUT请求。client.delete(url, **kwargs)
:发送HTTP DELETE请求。client.head(url, **kwargs)
:发送HTTP HEAD请求。client.patch(url, data=None, **kwargs)
:发送HTTP PATCH请求。between(min_wait, max_wait)
:设置任务之间的等待时间范围,单位S。task(weight=1)
:装饰测试函数,指定测试函数的权重、优先级,默认:1。
4. Locust常用库
Flask
:Web框架,用于构建Web界面。gevent
:Python协程库,提供高效的并发I/O支持。psutil
:跨平台库,用于查询操作系统和进程信息。pygal
:Python SVG图表生成库。pyzmq
:Python ZeroMQ机制,用于实现分布式测试和消息传递。requests
:Python HTTP库,用于发送HTTP请求。
5. Locust安装
5.1 安装Python
在使用Locust之前,您需要先安装Python。Locust需要使用Python3.6或更高版本。
Linux和macOS用户可以通过以下命令安装Python:
$ sudo apt-get install python3
Windows用户可以通过以下链接下载Python安装程序:https://www.python.org/downloads/windows/
5.2 安装Locust
在安装完Python之后,您可以通过使用pip
命令来安装Locust。在命令行终端中,运行以下命令即可完成Locust的安装:
$ pip install locust
在安装完成之后,您可以通过运行以下命令来检查Locust的版本:
$ locust -V
5.3 安装pyzmq
如果打算运行Locust 分布在多个进程/机器,需要安装pyzmq:
$ pip install pyzmq
6. Locust基础使用模板
假设我们需要测试一个社交网站,其中,用户可以发布帖子、查看帖子列表等 API,下面是一个实现这些操作的 Locust 测试:
from locust import HttpUser, SequentialTaskSet, task, between, constant
# 定义一个任务集合,包含一组序列任务
class UserBehavior(SequentialTaskSet):
# 用户登录
def login(self):
# prepare data
username = 'testuser'
password = 'testpass'
# 发送登录请求
response = self.client.post('/login', json={
'username': username,
'password': password
},catch_response=True)
# 校验登录结果
if response.status_code == 200:
self.auth_token = response.json().get('auth_token')
print("Login successfully, got auth_token:", self.auth_token)
else:
print("Failed to login:", response.status_code, response.json())
# 用户退出登录
def logout(self):
# 发送退出登录请求
response = self.client.post('/logout', headers={
'Authorization': 'Bearer ' + self.auth_token
},catch_response=True)
# 校验退出登录结果
if response.status_code == 200:
print("Logout successfully")
self.interrupt()
else:
print("Failed to logout:", response.status_code, response.content)
# 前置:登录
def on_start(self):
self.login()
# 后置:退出
def on_start(self):
self.logout()
# 任务1: 用户发帖
@task
def post_content(self):
# prepare data
title = 'title_' + str(randint(1000, 9999))
content = 'content_' + str(randint(1000, 9999))
# 发送发帖请求
response = self.client.post('/posts', json={
'title': title,
'content': content
}, headers={
'Authorization': 'Bearer ' + self.auth_token
},catch_response=True)
# 校验发帖结果
if response.status_code == 201:
print("Post successfully:", title)
else:
print("Failed to post:", response.status_code, response.content)
# 任务2: 用户查看帖子列表
@task
def view_posts(self):
# 发送查帖子列表请求
response = self.client.get('/posts', headers={
'Authorization': 'Bearer ' + self.auth_token
},catch_response=True)
# 校验查帖子列表结果
if response.status_code == 200:
post_list = response.json()
if len(post_list) > 0:
print("View posts successfully, got post number:", len(post_list))
else:
print("No posts found")
else:
print("Failed to view posts:", response.status_code, response.content)
# 定义 Locust 测试用户
class WebsiteUser(HttpUser):
tasks = [UserBehavior]
wait_time = between(0.5, 1)
在上面的代码中,我们在 SequentialTaskSet
中定义了一个由四个任务组成的序列,分别是:发帖、查看帖子列表。在运行测试时,每个用户将执行这个序列中的任务,依次模拟用户的行为。
同时注意,我们在每个请求中都在 header 中加入了 Authorization
,用于传递用户身份认证信息。另外,在每个任务执行完以后,我们需要校验响应的状态码和内容是否满足预期,例如在发帖任务中,只有在收到 HTTP 状态码为 201 时,才定义这个请求已经成功。
样例说明SequentialTaskSet
/TaskSequence
/TaskSet
主要区别:
SequentialTaskSet
继承于 TaskSet
,多个任务之间是串行执行的,即完成一个任务才会去执行下一个任务。
应用场景:
- 建立会话
- 踢出其他同名登录
TaskSet
多个任务之间并行执行。
应用场景:
- 并行执行多个接口
- 测试多个逻辑代码
TaskSequence
是一个新的概念,它在 Locust 1.4 版本中引入。在使用上,TaskSequence
类似于组合模式,将多个TaskSet
按照顺序组合成一个序列,从而实现一个业务流程的测试。
如果需要在任务集合之间加上一些固定的等待时间,可以使用 wait_time
参数指定;如果需要在任务之间加上随机时间间隔,可以使用 between
函数自定义起始和结束等待时间。
下面是一个使用 TaskSet
的样例:
from locust import HttpUser, TaskSet, between, task
class UserBehavior(TaskSet):
@task
def task1(self):
self.client.post("/task1")
@task
def task2(self):
self.client.post("/task2")
@task
def task3(self):
self.client.post("/task3")
class MyUser(HttpUser):
tasks = [UserBehavior]
wait_time = between(1, 2)
下面是一个使用 SequentialTaskSet
的样例:
from locust import HttpUser, TaskSet, SequentialTaskSet, between, task
class UserBehavior(SequentialTaskSet):
@task
def task1(self):
self.client.post("/login")
@task
def task2(self):
self.client.post("/post")
@task
def task3(self):
self.client.post("/logout")
class MyUser(HttpUser):
tasks = [UserBehavior]
wait_time = between(1, 2)
在可以预料到的情况下,如果您的任务是顺序执行的,推荐使用 SequentialTaskSet
;如果您的任务相互独立且不需要按特定顺序执行,则考虑使用 TaskSet
。
下面是一个使用 TaskSequence
的样例:
假设我们要对某个社交网站进行压力测试,测试场景如下:
- 用户登录
- 用户发帖
- 用户评论
- 用户点赞
- 用户查看帖子列表
我们可以将以上活动分别封装到 TaskSet
中作为独立的任务:
class Login(TaskSet):
def on_start(self):
self.client.post('/login', json={'username': 'test', 'password': 'test'})
class Post(TaskSet):
@task
def post_new(self):
self.client.post('/post', json={'title': 'newpost', 'content': 'newcontent'})
class Comment(TaskSet):
@task
def comment(self):
self.client.post('/comment', json={'post_id': 1, 'content': 'good post!'})
class Like(TaskSet):
@task
def like(self):
self.client.post('/like', json={'post_id': 1})
class ViewPosts(TaskSet):
@task
def view_list(self):
self.client.get('/posts')
接下来,我们可以在测试用户中,通过组合以上活动来模拟用户的真实流程:
class SocialNetworkUser(TaskSequence):
tasks = [Login, Post, Comment, Like, ViewPosts]
wait_time = constant(2)
上面的测试用户实现了一个标准的业务流,从用户登录开始,依次完成发帖、评论、点赞和查看帖子等操作,每个任务间隔2秒(wait_time
= 2)。
TaskSequence
通过将多个 TaskSet
按顺序连接,来实现业务流的测试,提高测试编写和维护的效率。而 TaskSet
则一般用来定义一组并行的任务,例如同时并发发送多个请求。
7. Locust高级用法:
数据关联
在一些复杂的测试场景中,需要依赖上一请求返回的结果作为下一次请求的参数,这就需要用到数据关联的功能。
示例代码:
from locust import HttpUser, task, between
class MyUser(HttpUser):
@task
def get_session(self):
response = self.client.post("/sessions")
session_id = response.json()["id"]
self.user_data = {"session_id": session_id}
@task
def get_user_profile(self):
self.client.get(f"/user/{self.user_data['session_id']}/profile")
在上面的例子中,get_user_profile
函数需要依赖 get_session
函数返回的结果,因此,我们在 get_session
函数中用 User
类的属性(self.user_data
)保存了所需要的数据,并在下一次请求中引用该数据。
参数化
在测试场景中,我们可能需要使用不同的参数进行多次请求,这就需要用到参数化的功能。
示例代码:
from locust import HttpUser, task, between
class MyUser(HttpUser):
WAIT_TIME = between(1, 2)
@task
def search(self):
self.client.get(f"/search", params={"keyword": self.random_keyword()})
def random_keyword(self):
keywords = ["apple", "banana", "orange"]
return random.choice(keywords)
在上面的例子中,关键字 keyword
仅仅是为了演示参数化的例子。我们在 search
函数中,使用 self.random_keyword()
获取随机的一个关键字,并通过 params
参数对搜索接口进行调用。
思考时间
在测试场景中,如果一直让模拟用户间隔同样的时间发起请求,可能会对系统产生不同于实际的负载。因此,可以使用思考时间的功能,让模拟用户在一定时间后再发起请求。
示例代码:
from locust import HttpUser, task, between
class MyUser(HttpUser):
WAIT_TIME = between(1, 2)
@task
def search(self):
self.client.get(f"/search")
self.think_time()
def think_time(self):
time.sleep(random.uniform(0.3, 0.5))
在上面的例子中,我们定义了 think_time
函数,用于控制思考时间。在 search
函数中,调用 think_time
函数,让模拟用户在发起两次搜索请求之间休息一段时间。
权重
在某些场景中,有些请求比其他请求更加重要,可以使用权重的功能来控制负载。
示例代码:
from locust import HttpUser, task, between
class MyUser(HttpUser):
WAIT_TIME = between(1, 2)
@task(3)
def search(self):
self.client.get(f"/search")
@task(1)
def get_user_profile(self):
self.client.get(f"/profile")
在上面的例子中,我们使用 @task
装饰器来设置请求的权重。在 search
函数中,我们将其权重设置为 3
,在 get_user_profile
函数中将其权重设置为 1
。
集合
在测试场景中,有时候需要组合不止一个请求,可以使用 Locust 集合的功能来实现。
示例代码:
from locust import HttpUser, TaskSet, task, between
class UserBehavior(TaskSet):
def on_start(self):
response = self.client.post("/login", {"username":"testuser", "password":"testpass"})
self.token = response.json()["token"]
@task
def get_profile(self):
self.client.get("/user/profile", headers={"Authorization": f"Bearer {self.token}"})
@task
def get_orders(self):
self.client.get("/user/orders", headers={"Authorization": f"Bearer {self.token}"})
class MyUser(HttpUser):
WAIT_TIME = between(1, 2)
tasks = {UserBehavior: 1}
在上面的例子中,我们使用 Locust 集合的方式,定义了一个 UserBehavior
类来表示一组请求。在该类的 on_start
函数中,先进行用户登录操作,获取 token
,然后在后续的请求中,使用 token
做认证。
最后,在 MyUser
类中,使用 tasks
属性来关联 UserBehavior
类,并设置该集合的权重,这里设置为 1
。当启动 Locust 后,每次会随机选择一个集合内的请求进行发送。
基础组合样例
这个样例是一个 Web 应用程序的压力测试,其中包含了集合、权重、思考时间、前置和后置等常用知识点。
假设有一个在线音乐平台,我们要对其进行压力测试。这个音乐平台有以下几个功能:
- 注册新用户
- 用户登录
- 搜索音乐
- 播放音乐
- 查看播放历史
压力测试中,我们需要模拟具有以下特征的用户:
- 注册和登录的用户占比为 20% 和 80%,使用权重进行控制
- 用户进行搜索和播放音乐的比例为 4:1,即每个用户进行 4 度搜索后才进行一次音乐播放
- 用户每进行一个操作后,需要思考 0.5-1.5 秒的时间
- 在每次操作前,需要判断用户是否已经登录,如果没有,则先进行登录操作
- 在每次操作后,需要查看播放历史
同时,我们需要模拟 1000 个用户,持续 1800 秒的测试时间。其中,注册和登录的用户,用户名和密码为随机生成一个 6 位数。
下面是 Locustfile 中的代码实现:
from locust import HttpUser, task, between, constant_pacing
import random
class MusicUser(HttpUser):
min_wait = 500
max_wait = 1500
@task(1)
def register_user(self):
self.client.post("/register", json={"username": str(random.randint(100000, 999999)), "password": str(random.randint(100000, 999999))})
@task(4)
def search_music(self):
if not self.logged_in():
self.login()
self.client.get("/search?keyword=Let It Go")
self.think_time()
@task(1)
def play_music(self):
if not self.logged_in():
self.login()
self.client.get("/play?id=123456")
self.think_time()
def browse_history(self):
if not self.logged_in():
self.login()
self.client.get("/history")
self.think_time()
def logged_in(self):
response = self.client.get("/profile")
return response.status_code == 200
def login(self):
self.client.post("/login", json={"username": "testuser", "password": "testpassword"})
self.browse_history()
def think_time(self):
self.sleep(random.uniform(0.5, 1.5))
def on_start(self):
if random.random() < 0.2:
self.register_user()
else:
self.login()
def on_stop(self):
self.browse_history()
在这个样例中,我们定义了一个名为 MusicUser
的 Locust 用户,并设置了最小等待时间为 500 毫秒,最大等待时间为 1500 毫秒。其中 think_time
方法用于模拟用户操作之间的思考时间。logged_in
方法用于判断用户是否已经登录,login
方法用于模拟用户登录操作。 browse_history
方法用于查看播放历史。
在 on_start
方法中,我们用一个随机数来判断用户进行注册还是登录操作。register_user
方法用于模拟用户注册操作,search_music
方法用于模拟用户进行音乐搜索操作,play_music
方法用于模拟用户进行音乐播放操作。
在这个样例中,我们使用了 task
装饰器来定义不同任务的权重。@task(1)
表示这个任务的权重为 1,而 @task(4)
表示这个任务的权重为 4。这个权重的设置可以用来控制每个任务的负载情况。
同时,在 search_music
和 play_music
两个任务中,我们使用了 if not self.logged_in():
来判断用户是否已经登录。如果没有登录,则先进行登录操作,然后再进行搜索音乐或播放音乐。
此外,在 search在
search_music和
play_music中,我们还使用了
self.think_time()来模拟用户操作之间的思考时间。这个思考时间是一个随机数,将在
think_time` 方法中生成。
最后,我们使用了 on_start
和 on_stop
方法来模拟用户的起始和终止行为。在 on_start
中,我们用一个随机数来判断用户进行注册还是登录操作;在 on_stop
中,我们调用 browse_history
方法来查看播放历史。
通过这个样例,我们可以看到如何使用 Locust 中的集合、权重、思考时间、前置和后置等常用知识点来进行高并发压力测试。
在 Locust 1.0 版本之后,TaskSet 已经不再是必须的了。您可以直接在 User 类中定义 task 方法,来告诉 Locust 模拟用户行为。本文提供的高并发样例中就是通过定义 @task
装饰器来模拟用户行为的。
当然,如果您想在 User 类中封装不同的测试任务,或者需要使用 setUp
和 tearDown
等方法来执行测试前或测试后的准备工作,那么可以继续使用 TaskSet 的方式。但是,如果您只需要简单模拟用户行为,上述的代码已经足够了。
复杂组合样例
以下是一个结合 User 和 TaskSet 的 Locust 高并发测试样例:
from locust import HttpUser, TaskSet, task, between
class LoginPage(TaskSet):
@task
def loadLoginPage(self):
self.client.get("/login")
@task
def submitLoginForm(self):
self.client.post("/submit_login", {
"username": "testuser",
"password": "testpass"
})
class SearchPage(TaskSet):
@task
def loadSearchPage(self):
self.client.get("/search")
@task
def performSearch(self):
self.client.post("/search_query", {
"query": "test"
})
class UserBehaviour(HttpUser):
tasks = [LoginPage, SearchPage]
wait_time = between(0.5, 2.0)
def on_start(self):
self.client.post("/login", {
"username": "testuser",
"password": "testpass"
})
在上述样例中,我们定义了两个 TaskSet:LoginPage
和 SearchPage
。LoginPage
主要用于加载登录页面和提交登录表单,SearchPage
主要用于加载搜索页面和执行搜索操作。
在 UserBehaviour
类中,我们通过定义 tasks
属性来指定了要执行的测试任务,即 LoginPage
和 SearchPage
。我们还定义了一个 wait_time
属性,用于控制每个用户执行测试任务之间的等待时间。
在 UserBehaviour
类中还定义了一个 on_start
方法,用于在模拟用户开始测试之前执行一些准备工作,包括登录操作。
使用 TaskSet 的优势在于我们可以将测试任务的逻辑进行更好地组织和拆分,提高代码的可读性和可维护性。对于复杂的测试场景,使用 TaskSet 和 User 的组合可以更好地支持测试任务之间的依赖关系和流程。
查看帮助文档,执行: locust --help
参数说明:
-
-h, --help
:显示帮助消息并退出。 -
-f LOCUSTFILE, --locustfile LOCUSTFILE
:指定要导入的Python模块。可以是Python文件、多个以逗号分隔的Python文件或包目录。默认为'locustfile'。 -
--config CONFIG
:指定配置文件路径。 -
-H HOST, --host HOST
:指定要进行负载测试的主机,格式为:http://10.21.32.33。 -
-u NUM_USERS, --users NUM_USERS
:指定并发用户峰值数量。主要与--headless或--autostart一起使用。在测试过程中可以通过键盘输入w、W (生成1或10个用户)和s、S (停止1或10个用户)来进行更改。 -
-r SPAWN_RATE, --spawn-rate SPAWN_RATE
:指定每秒生成用户的速率(用户数/秒)。主要与--headless或--autostart一起使用。 -
-t RUN_TIME, --run-time RUN_TIME
:指定停止测试的时间,例如(300s,20m,3h,1h30m等)。仅与--headless或--autostart一起使用。默认为永久运行。 -
-l, --list
:显示可用的用户类并退出。 -
--web-host WEB_HOST
:指定要绑定Web界面的主机,默认为所有接口("*")。 -
--web-port WEB_PORT, -P WEB_PORT
:指定Web服务端口号。 -
--headless
:禁用Web界面,并立即开始测试。使用-u和-t来控制用户数量和运行时间。 -
--autostart
:立即开始测试(类似于--headless,但不会禁用Web UI)。 -
--autoquit AUTOQUIT
:在测试结束后X秒钟内完全退出Locust。仅与--autostart一起使用,默认情况下,Locust会一直运行,直到使用CTRL+C结束。 -
--web-auth WEB_AUTH
:为Web界面打开基本身份验证。应该使用以下格式提供:username:password。 -
--tls-cert TLS_CERT
:指定用于通过HTTPS服务的TLS证书的路径。 -
--tls-key TLS_KEY
:指定用于通过HTTPS服务的TLS私钥的路径。 -
--class-picker
:在Web界面中启用选择框,以选择所有可用的用户类和形状类。 -
--master
:设置Locust以分布式模式运行,该进程作为主节点。用于运行从节点。 -
--master-bind-host MASTER_BIND_HOST
:指定Locust主节点绑定到的接口(主机名、IP地址)。只有在使用--master运行时才会被使用。默认为*(所有可用接口)。 -
--master-bind-port MASTER_BIND_PORT
:指定Locust主节点绑定的端口号。只有在使用--master运行时才会被使用。默认为5557。 -
--expect-workers EXPECT_WORKERS
:设置Locust主节点在开始测试前需要连接的从节点数量(仅在使用--headless/autostart时使用)。 -
--expect-workers-max-wait EXPECT_WORKERS_MAX_WAIT
:设置Locust主节点等待从节点连接的最长时间。默认为无限等待。 -
-T [TAG [TAG ...]], --tags [TAG [TAG ...]]
:列出要在测试中包含的标签,因此只有具有任何匹配标签的任务才会被执行。 -
-E [TAG [TAG ...]], --exclude-tags [TAG [TAG ...]]
:列出要从测试中排除的标签,因此只有没有匹配标签的任务才会被执行。 -
--csv CSV_PREFIX
:将当前请求统计信息以CSV格式保存到文件中。设置此选项将生成三个文件:[CSV_PREFIX]_stats.csv,[CSV_PREFIX]_stats_history.csv和[CSV_PREFIX]_failures.csv。 -
--csv-full-history
:将每个统计项以CSV格式存储到_stats_history.csv文件中。您还必须指定“--csv”参数以启用此选项。 -
--print-stats
:在UI运行中启用请求统计信息的定期打印。 -
--only-summary
:在--headless运行期间禁用请求统计信息的定期打印,仅打印摘要信息。 -
--reset-stats
:当生成的测试在所有的Locust Worker节点上均已启动后重置统计信息。在分布式模式下运行时,主节点和worker节点均应启用此选项。 -
--html HTML_FILE
:将HTML报告存储到指定的文件路径中。 -
--json
:将最终的统计信息以JSON格式打印到stdout。对于在其他程序/脚本中解析测试结果非常有用。与--headless和--skip-log一起使用,仅输出json数据。 -
--skip-log-setup
:禁用Locust的日志设置。日志设置由Locust测试或Python默认值提供。 -
--loglevel LOGLEVEL, -L LOGLEVEL
:选择DEBUG/INFO/WARNING/ERROR/CRITICAL级别。默认为INFO。 -
--logfile LOGFILE
:设置日志文件路径。如果未设置,日志将会写入stderr。 -
--show-task-ratio
:打印用户类的任务执行比率表格,用于非零用户选项时,如果某些类定义了非零fixed_count属性。 -
--show-task-ratio-json
:打印用户类的任务执行比率的json数据,用于非零用户选项时,如果某些类定义了非零fixed_count属性。 -
--version, -V
:显示程序的版本号并退出 -
--exit-code-on-error EXIT_CODE_ON_ERROR
:设置当测试结果包含任何失败或错误时要使用的进程退出代码。 -
-s STOP_TIMEOUT, --stop-timeout STOP_TIMEOUT
:在退出之前等待模拟用户完成所有正在执行的任务的秒数。默认为立即终止。仅在运行Locust分布式时,主进程需要指定此参数。 -
--equal-weights
:使用平均分布任务权重,覆盖locustfile中指定的权重。 -
--enable-rebalancing
:允许在测试运行期间自动重新平衡用户,如果添加或删除了新的worker节点。 -
UserClass
:可选地指定应使用哪些用户类(可使用-l或--list列出可用的用户类)。