本书作者:测试开发刚哥,文章版权归个人所有,未经授权请勿转载。
locustfile 是什么? locustfile 是 Locust 性能测试工具的用户脚本,描述了单个用户的行为。
locustfile 是个普通的 Python 模块,如果写作locustfile.py,那么路径切换到文件所在目录,直接执行命令就能运行:
$ locust 如果换个名字,那么只能通过-f参数指定文件名运行:
$ locust -f locust_files/my_locust_file.py 与一般 Python 模块不同的是:locustfile 必须至少定义一个类,且继承自 User 类。
User 类 User 类表示性能测试的模拟用户,Locust 会在运行时创建 User 类的实例。
wait_time 属性 设置等待时间,默认值不等待,立即执行。
Locust 支持 4 种方式设置 wait_time 属性。
为了便于理解实际意义,我把源码贴在了下面。
constant函数,常量,任务执行完毕等待 X 秒开始下一任务。
def constant(wait_time): """ Returns a function that just returns the number specified by the wait_time argument Example:: class MyUser(User): wait_time = constant(3) """ return lambda instance: wait_time between函数,区间随机值,任务执行完毕等待 X-Y 秒(中间随机取值)开始下一任务。
def between(min_wait, max_wait): """ Returns a function that will return a random number between min_wait and max_wait. Example:: class MyUser(User): # wait between 3.0 and 10.5 seconds after each task wait_time = between(3.0, 10.5) """ return lambda instance: min_wait + random.random() * (max_wait - min_wait) constant_pacing函数,自适应,若任务耗时超过该时间,则任务结束后立即执行下一任务;若任务耗时不超过该时间,则等待达到该时间后执行下一任务。
def constant_pacing(wait_time): """ Returns a function that will track the run time of the tasks, and for each time it's called it will return a wait time that will try to make the total time between task execution equal to the time specified by the wait_time argument. In the following example the task will always be executed once every second, no matter the task execution time:: class MyUser(User): wait_time = constant_pacing(1) @task def my_task(self): time.sleep(random.random()) If a task execution exceeds the specified wait_time, the wait will be 0 before starting the next task. """ def wait_time_func(self): if not hasattr(self, "_cp_last_run"): self._cp_last_wait_time = wait_time self._cp_last_run = time() return wait_time else: run_time = time() - self._cp_last_run - self._cp_last_wait_time self._cp_last_wait_time = max(0, wait_time - run_time) self._cp_last_run = time() return self._cp_last_wait_time return wait_time_func 自定义wait_time方法,比如每次等待时间 1 秒 2 秒 3 秒递增:
class MyUser(User): last_wait_time = 0 def wait_time(self): self.last_wait_time += 1 return self.last_wait_time ... ...