Locust 集合点

时间:2023-03-10 04:26:21
Locust 集合点

直接编写接口事务脚本对后台接口进行测试;有时测试需要让所有并发用户完成初始化后再进行压力测试,这就需要类似于LoadRunner中的集合点的概念,由于框架本身没有直接封装,有如下办法实现:

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 //挂载到locust钩子函数(所有的Locust实例产生完成时触发) 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() //限制在所有用户准备完成前处于等待状态

思路:

通过locust得基于gevent并发得机制,引入gevent的锁的概念,代入到locust的钩子函数中,实现集合点统一并发概念