Locust性能测试工具核心技术@task和@events

Tasks和Events是Locust性能测试工具的核心技术,有了它们,Locust才能称得上是一个性能工具。

locustfile中的User类和HttpUser类

本书作者:测试开发刚哥,文章版权归个人所有,未经授权请勿转载。 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 ... ...

Python技术栈性能测试工具Locust入门

Locust是一款Python技术栈的开源的性能测试工具。Locust直译为蝗虫,寓意着它能产生蝗虫般成千上万的并发用户: