性能测试工具 Locust 性能测试 - 添加集合点

在路上 · 2018年03月02日 · 最后由盖水回复于 2021年12月03日 · 3498 次阅读

Locust 作为一款基于 Python 轻量化的性能测试工具，中小型的团队做接口压测还是比较方便的，直接编写接口事务脚本对后台接口进行测试；但有时测试需要让所有并发用户完成初始化后再进行压力测试，这就需要类似于 LoadRunner 中的集合点的概念，由于框架本身没有直接封装，google 后实践有效，分享出来：

Python 相关代码

from locust import events
from gevent._semaphore import Semaphore
all_locusts_spawned = Semaphore()
all_locusts_spawned.acquire()

def on_hatch_complete(**kwargs):
    all_locusts_spawned.release()

events.hatch_complete += on_hatch_complete

class TestTask(TaskSet):
    def on_start(self):
        """ on_start is called when a Locust start before any task is scheduled """
        self.login()
        all_locusts_spawned.wait()

5 个赞

如果觉得我的文章对您有用，请随意打赏。您的支持将鼓励我继续创作！

打赏支持

共收到 11 条回复时间点赞

1楼已删除

我是老罗 #2 · 2018年08月08日

我在使用的时候，由于要使用 login（）中的返回数据当做实际压测接口的入参，按你这个方法在 on_start 函数只执行了一次，所以压测的数据只有第一次的并发所生成，无法实现动态得掉用不同的登录账号，求大神指点

在路上 #3 · 2018年08月08日 Author

压测期间如果需要动态更换帐号，则需要在压测的事务场景中再次调用 login 手工切换帐号了

多读书 #4 · 2018年08月17日

两个问题请教下：
1、“压测期间如果需要动态更换帐号，则需要在压测的事务场景中再次调用 login 手工切换帐号了”，这句话没理解。比如我并发 5，不是在 onstart 就起来了五个用户吗，用户起完后不会再执行 login 了，此时再调用 login 切换账号复用返回值，拿到的值对应这五个用户之一？
2、集合点如何验证？login 没有 all_locusts_spawned.wait() 的时候和有 all_locusts_spawned.wait() 的时候，两次执行结果，并未发现明显的同时刻请求的证据，因为我只能 print response 看到结果响应时间，不能看到请求发出的时间。请大佬帮忙抽空指点下

在路上 #5 · 2018年08月17日 Author

对

多读书回复

回复：
1、切换帐号不代表变更并发用户数，意思是同一个并发用户下测试业务场景时需要更换帐号登录才会这样做，如果你的场景压测期间不需要切换帐号那就只需要在 onstart 中登录一次就可以；
2、集合点请求发出的时间如果需要获取的话，则需要自己在集合点完成后的压测场景脚本中加入打印时间了，框架目前没有提供这类监控支持；
locust 是个轻量级的压测框架，做的都是最基本的，但方便之处在于自己可根据自己的压测需要，方便二次开发；

多读书 #6 · 2018年08月17日

对

在路上回复

谢谢大佬抽空解答。
“1” 是当初没理解到，你一提我明白了；
“2” 的话，其实我是想要个比较确切的答案，就是加入您提供的这个方法后，是否真的就合入了集合点的概念了呢？因为我理解到的，集合点的概念现在好像比较弱化，没有必要严格遵循，是否是这样呢？如果说中小并发确认是没有必要设置集合点，大并发还是需要的，那么这个 “大并发” 应该怎么定义，大概是个什么具体值呢？1000？

####################### 用例从这里开始执行 ######################
class LRdemo_inquiry(TaskSet):
"""压力测试样本"""

################### 并发用户从这里启动 ######################
def on_start(self):
global j
self.userId = userId_list[j]
j = j + 1
if j >= len(userId_list):
j = 0

url = Ip + "/passport/login"
body_data = {
"logintype":logintype,
"username":self.userId,
"password":Password,
}
self.client.post(url,body_data)

all_locusts_spawned.wait()
################## 并发用户从这里启动 ######################

################# 这里是示例 ###################
@task(1)
def inquirty_vin(self):

for k in range(len(userId_list)):
if self.userId == userId_list[k]:
self.Inquiry_list = txt_list[k]
Inquiry = random.choice(self.Inquiry_list)

url = Ip + "/agentBuy/inquiryDetailData/"+Inquiry

response = self.client.get(url)

print(self.userId)
print(url)
#print(response.text)
################# 这里是示例主题 ###################
####################### 用例从这里开始执行 ######################

我调试以上脚本（贴出部分），发现有没有 “all_locusts_spawned.wait()” 没有什么差别，也不知道哪里有问题，还是确实就是这样~
望大佬再帮忙抽空指点下。
顿首百拜