属性及方法
wait_time属性
用户
内置了三个等待时间功能:
例如,使每个用户在每次任务执行之间等待0.5到10秒:
from locust import User, task, between class MyUser(User): @task def my_task(self): print("executing my_task") wait_time = between(0.5, 10)
也可以在类上直接声明自己的wait_time方法。例如,下面的User类将休眠一秒钟,然后休眠两个,然后休眠三个,依此类推。
class MyUser(User): last_wait_time = 0 def wait_time(self): self.last_wait_time += 1 return self.last_wait_time ...
如果文件中存在多个用户类,并且在命令行上未指定任何用户类,则Locust将产生相等数量的每个用户类。您还可以通过将它们作为命令行参数传递,来指定要从同一locustfile中使用哪些用户类:
$ locust -f locust_file.py WebUser MobileUser
如果您希望模拟更多特定类型的用户,则可以在这些类上设置一个weight属性。举例来说,网络用户的可能性是移动用户的三倍:
class WebUser(User): weight = 3 ... class MobileUser(User): weight = 1 ...
Host属性
host属性是要加载的主机的URL前缀(即“http//:xxx.com ”)。通常,这是在蝗虫--host
启动时在Locust的Web UI或命令行中使用该 选项指定的。
如果在用户类中声明了主机属性,则--host
task属性
User类可以使用
self.environment.runner.quit()
如果在独立蝗虫实例上运行,则将停止整个运行。如果在工作程序节点上运行,它将停止该特定节点。
on_start和on_stop方法
用户(和
@task装饰器
为用户添加任务的最简单方法是使用
from locust import User, task, constant class MyUser(User): wait_time = constant(1) @task def my_task(self): print("User instance (%r) executing my_task" % self)
from locust import User, task, between class MyUser(User): wait_time = between(5, 15) @task(3) def task1(self): pass @task(6) def task2(self): pass
tasks
启动负载测试后,将为每个模拟用户创建一个User类的实例,并且它们将在其自己的绿色线程中运行。这些用户运行时,他们选择执行的任务,休眠一会儿,然后选择一个新任务,依此类推。
tasks属性
定义用户任务的另一种方法是设置
在任务属性为任务的列表,或<任务:int>的字典,其中,任务或者是一个可调用蟒或
这是一个声明为普通python函数的User任务的示例:
from locust import User, constant def my_task(user): pass class MyUser(User): tasks = [my_task] wait_time = constant(1)
{my_task: 3, another_task: 1}
my_task的执行可能性是another_task的3倍。
在内部,上面的dict实际上将扩展为如下所示的列表(并且tasks
[my_task, my_task, my_task, another_task]
@tag标签装饰器
通过使用
from locust import User, constant, task, tag class MyUser(User): wait_time = constant(1) @tag('tag1') @task def task1(self): pass @tag('tag1', 'tag2') @task def task2(self): pass @tag('tag3') @task def task3(self): pass @task def task4(self): pass
--exclude-tags`会以完全相反的方式表现。所以,如果你开始测试--exclude-tags tag3 ,只有TASK1,TASK2和task4将被执行。排除总是胜于包容,因此,如果任务具有包含的标签和包含的标签,则该任务将不会执行。
Events
如果要在测试中运行某些安装代码,通常将其放在locustfile的模块级别就足够了,但是有时您需要在运行的特定时间进行操作。为此,Locust提供了事件。
test_start和test_stop
如果需要在负载测试的开始或结束时运行一些代码,则应使用
from locust import events @events.test_start.add_listener def on_test_start(environment, **kwargs): print("A new test is starting") @events.test_stop.add_listener def on_test_stop(environment, **kwargs): print("A new test is ending")
init
该init
事件在每个蝗虫过程的开始时触发。这在分布式模式下特别有用,在分布式模式下,每个工作进程(而不是每个用户)都需要机会进行一些初始化。例如,假设您有一个全局状态,此过程中产生的所有用户都需要:
from locust import events from locust.runners import MasterRunner @events.init.add_listener def on_locust_init(environment, **kwargs): if isinstance(environment.runner, MasterRunner): print("I'm on master node") else: print("I'm on a worker or standalone node")