三天入门Locust压力测试分享
什么是LocustLocust是一个基于Python的带有可视化图形界面的测试工具。
本文有什么
本人不是专业的测试人员，Python也是先学先用的，所以不会涉及到太多的相关专业知识。本文主要是分享自己学习使用Locust的收获，基于官方文档和他人的博客，结合自己的使用体验，所以不是一篇教学或者专业文。你在这篇文章中看不到Locust的嵌套task和多locustfile测试等。
名词解释
locust：本意是蝗虫，在测试过程中看成是一个用户即可（后面所谓的HttpLocust只是带有特殊性质的用户）。
hatch rate：孵化率，表示每秒钟产生的用户数量，可以模拟一个逐渐增加的过程，可以根据曲线来考察接口的处理峰值。
TaskSet：顾名思义，是任务的集合，代表着每个用户具有的行为，也就是你想要让用户对整个后台功能做什么。
max_wait/min_wait：最大等待时间/最小等待时间，这是用户的休息时间，用户在执行任务的过程中，在这两个值确定的区间内随机选择一个时间进行休息，模拟真实用户的缓冲，同时也可以保证某些带有评率限制的接口能够安全调用，否则可能会出现很多错误甚至被封IP等。
locustfile.py：文件名称当然可以随便定义，但是这个文件的功能，相当于建立了一个用户的模型，需要产生的大量用户都将以这个文件定义为原型。
locustfile
locust以这个脚本作为用户的原型孵化用户，这个脚本包括两个部分，TaskSet子类和Locust子类。TaskSet子类是用于描述用户行为的，提供on_start方法执行前置操作。Locust子类主要是使用HttpLocust子类，因为现在的测试主要是针对http的接口，当然也可以通过一些特别的模块实现websocket的测试。（虽然感觉websocket的测试也没有什么必要）
Locust/HttpLocust
这个类主要定义用户的目标（host），相当于一个baseURL，用户的行为（请求）都会相对于这个目标进行。定义用户的休息间隔（max_wait/min_wait）。定义用户的行为（TaskSet）。一个普通的HttpLocust可能是以下这样的：
class WebsiteUser(HttpLocust):
host = "https://juejin.cn"
max_wait = 3000
min_wait = 3000
task_set = UserAction
你还可以在这个类里面通过类的静态变量来提供一个所有用户都能使用的变量。比如一个queue，一个list等。由于类的静态变量的特性，每个用户修改这个值可以实现一个递增的index之类的需求，参数化的逻辑可以借助这一点。
TaskSet
这个类负责定义每个用户的实际动作，一个用户可能有很多动作，注册啦，买东西啦，撤销啦等等。on_start方法是给每个用户开始这些动作之前的一个前置准备，可以在这个方法中初始化一些东西，比如登录或者链接websocket。
TaskSet中可以通过self.client来使用HttpSession class，也就是用户的http请求能力的供应者，可以理解成用户的浏览器，用来发送请求和接受响应的。self.client是自带cookie和session的能力的，完全相当于一个浏览器环境。也就是说self.client发送了一个请求后，比如登录请求，成功后再用这个self.client发送别的请求，会带上登录请求成功后服务器返回的cookie。
TaskSet还提供了一种神奇的东西ResponseContextManager class，当self.client发送的请求带有catch_response=True的参数，像下面这样：
with self.client.get("/", catch_response=True) as response:
if response.content == "":
response.failure("No data")
# response.success()
一般的情况下使用这样的即可：
response = self.client.get("/",data=data)
复制代码这种普通的请求在响应状态码小于400的情况下都认为成功，然而现在大部分接口都是通过返回的数据来表达你的动作O不OK，这就需要ResponseContextManager通过success和failure来手动修正正确与错误的显示。
TaskSet中self.locust可以取到HttpLocust的实例，也就是说你在HttpLocust中的某些变量，你是可以使用的。当然如何使用还是取决于你的测试逻辑。
有了以上这些东西，基本就可以开始定义自己的动作了。首先你需要引入locust提供的task装饰器，TaskSet子类中的实例方法如果被task装饰，那么就会认为是一个需要执行的动作。@task(weight)是支持权重安排的，也就是说多项任务可以按照一定的频繁度来调用，更接近用户的某些实际体验，比如购物过程中翻页可能要比添加购物车频繁。
一个普通的TaskSet像是这样：
class ForumPage(TaskSet):
@task(100)
def read_thread(self):
pass
@task(7)
def create_thread(self):
pass
虽然两个任务什么都没干，但是执行的频率可以看出来谁更频繁。没有特殊的设置下，两个任务按照权重随机的顺序执行，也就是有可能在短期内无法按照实际的权重执行，也就是说每个任务之间最好不要有顺序以来（但是可以通过一些操作来保证顺序）。
一个普通的task像是这样：
@task
def test_login(self):
# 这里通过类的静态变量来控制一个全局的变量，保证唯一的index
# WebsiteUser这就是Locust子类
WebsiteUser.index += 1
data = {
# 这里组织你的发送数据
# 通过self.locust来访问所属的Locust子类
"uid": self.locust.index
}
with self.client.post("/loginl", data=data,
catch_response=True) as response:
# response提供了一些快捷的方法和属性，让你判断请求的情况
if response.ok:
# 如果响应的content是json格式的，可以帮你转成字典
result = response.json()
# 通过返回的数据来手动判断是否成功
if result['code'] == 200:
response.success()
else:
response.failure(result['msg'])
else:
response.failure(bytes.decode(response.content))
我们跑起来
有了locustfile.py就可以开始跑起来了，有可视化界面还是阔以的。命令行跑起来我就不多说了，浏览器打开localhost:8089，会让你输入总用户量和孵化率。加入你输入200和1，就代表每一秒产生一个用户，直到200个用户都产生。注意：到达总量时统计数据会重置。由于用户的行为是尽可能模拟实际情况，所以考察数据的时候应当清楚一些。200个用户，假设有两个task，比重1：1，min_wait和max_wait都是1000ms，那么理想情况下某一时刻，200个用户都发送了一个请求，且这个请求的响应时间可以忽略，那么我们可以认为一秒的并发是200，等待时间是1s，故并发应当稳定在200。所以你需要考虑响应时间和等待时间，选择合适的用户数有可能才到达所需要的并发量。
我们需要满足其他的需求
参数化
我们在测试一个网站的功能时，我们通过locust模拟的用户需要不同的账号来进行操作，因此我们需要在TaskSet中做一些改造。之前我们说过了TaskSet拥有on_start方法执行前置动作，且self.client可以保持cookie和session，这不刚好可以满足登陆后执行操作的需求吗。那么下面一个问题就是如果使用不同的账号，这使用Python也是比较容易实现的。我们在WebsiteUser中定义一个类静态变量，比如下面这样：
class WebsiteUser(HttpLocust):
host = "https://juejin.cn"
max_wait = 3000
min_wait = 3000
task_set = UserAction
# 这个属性在locust启动时被初始化好
# 使用WebsiteUser.user_index来非竞争的更新，实现一个不重复的递增序列
user_index = 0
使用递增序列的方式，可以快速注册一大批用户，只需要确定好用户名和密码如果使用这个index进行产生即可。当然，如果不想使用这样的方式，还有一种就是使用数据结构，比如queue：
class WebsiteUser(HttpLocust):
host = 'https://juejin.cn'
task_set = UserBehavior
user_data_queue = queue.Queue()
# 一共300w用户准备
for index in range(1500000, 4500000):
data = {
# 准备你的注册或者登陆数据
}
user_data_queue.put_nowait(data)
min_wait = 500
max_wait = 500
@task
def test_register(self):
try:
# 用queue来使每个用户取得不同的账户信息
data = self.locust.user_data_queue.get()
except queue.Empty:
# 当然了，只出队列会导致数据跑空
# 如果需要循环，用完的数据从队列的另一头塞回去就可以重复利用了
print('account data run out, test ended.')
exit(0)
with self.client.post('/register', data=data,
catch_response=True) as response:
result = response.json()
if result['code'] == 200:
response.success()
else:
response.failure(result['msg'])
参数化的一些高级用法可以参见深入浅出开源性能测试工具Locust（脚本增强）会给你不少启发。
非http测试
少数情况我们想考量websocket的性能，比如同时连接的客户端以及推送到接受的延迟情况。Python强大的模块库肯定是有websocket相关的，通过搜索和尝试，pip2安装的websocket模块和pip3的好像不一样，使用pip3安装的实践成功。
一般来说我们在on_start方法中初始化连接：
self.ws = websocket.WebSocket()
self.ws.connect("ws://xxx")
并且将ws实例作为实例属性方便后续的操作。注意：websocket连接端在测试过程中断掉也是极有可能的，所以提供一个重连的方法可能会有必要。有了ws的连接之后呢，一般需要发送订阅信息让服务器来主动推送数据，如果ws在接受了订阅后会先返回一条订阅的结果，如果在可视化界面上通过数据统计出来，这就需要locust提供的hook。
from locust import TaskSet, task, HttpLocust, events将events导入进来，然后触发一些事件，皆可以让locust统计相应的数据：
events.request_success.fire(
request_type="wss",
name="send_entrust",
response_time=100,
response_length=300
)
request_type相当于请求的类型，可以随便填。name相当于请求url，response_time相当于响应事件，response_length相当于响应体大小。实际上self.client.post()等也会使用事件触发来统计数据。比如ResponseContextManager中success的实现：
def success(self):
"""
Report the response as successful
Example::
with self.client.get("/does/not/exist",
catch_response=True) as response:
if response.status_code == 404:
response.success()
"""
events.request_success.fire(
request_type=self.locust_request_meta["method"],
name=self.locust_request_meta["name"],
response_time=self.locust_request_meta["response_time"],
response_length=self.locust_request_meta["content_size"],
)
self._is_reported = True
这样的话，设计一个task来发送订阅并接受推送。作为前端人员来说，websocket都是用回调来处理推送，那么Python里面的websocket怎么办呢？实践成功的websocket模块连接后，也就是从connect()之后，self.ws本身是一个可迭代的，虽然每次next都是调用resv()方法，在循环里就好使。而self.ws.resv()这个函数是很神奇的，调用开始后他会等到接受推送了再返回，宏观角度来说这个函数的执行时间，应当等于服务器推送的间隔，一次可以判断推送是否出现了延迟。且这个函数如果返回的是空字符串，大致可以认为连接断掉，逻辑上增加统计和重连，能够保证一些测试准确度。
给出一个我实际使用的task：
@task
def test_ws(self):
# 这里可以使用这一个task来完成发送订阅和处理订阅
if not self.send_flag:
self.ws.send("""{
# 这里组织你需要发送的订阅数据
}
""")
result = json.loads(self.ws.recv())
# 这里来处理订阅的结果，如果订阅成功改变标志位，让这个task只执行接受
if result['code'] == 200:
self.send_flag = True
events.request_success.fire(
request_type="wss",
name="send_entrust",
response_time=100,
response_length=300)
else:
events.request_failure.fire(
request_type="wss",
name="send_entrust",
response_time=100,
exception=Exception(self.ws.status),
)
else:
flag = True
# 由于推送是不间断的直到发送取消订阅，故使用死循环来不断统计
while flag:
# self.ws本身是个迭代器，next和调用resv()是一样的，都可以
start_time = time.time()
resv = next(self.ws)
# 推送间隔就可以当做统计数据进行显示
total_time = int((time.time() - start_time) * 1000)
if resv != '':
events.request_success.fire(
request_type="wss",
name="resv_entrust",
response_time=total_time,
response_length=59)
else:
# 如果中断了可以增加重连方案
events.request_failure.fire(
request_type="wss",
name="resv_entrust",
response_time=total_time,
exception=Exception(""),
)
flag = False
else:
print("no resv")
总结
Locust总体来说还是比较容易上手的，提供的功能大概能够满足一般的测试需求。实践的过程中遇到过一些些问题，比如请求的数量不一定在页面上统计的及时，用户数也有可能会断掉一两个，具体原因不明。本人也不是专门的测试人员， 也不是Python native speaker，所以对于Python的理解和对测试分析的观点不一定对，只是分享一下我在使用的过程中如何思考问题如何解决问题，解决方案不一定最优，也有可能会造成很多困扰，但这是我的学习过程，想和大家分享，愿意和大家讨论并继续学习。
页:
[1]
