我需要一个关于如何使用自定义客户端(在我的案例中是WebSocket Server)编写蝗虫负载测试的示例或解释。我看到了蝗虫文档中给出的解释,但我无法确切了解挂钩蝗虫事件的函数__getattr__
和def wrapper(*args, **kwargs):
是如何通过蝗虫触发的。
我今天讨论了这个问题,还发现自定义示例有点棘手,因为很难判断魔法发生在哪里。这是一个更简单的客户端、任务和用户交互版本。我没有让decorator向客户端捕获任何任意方法,而是让它只能做一件事,这样你就可以看到时间和错误处理发生在哪里,以及它如何使用蝗虫事件挂钩发送成功和失败。
import time
from locust import TaskSet, task, Locust, events
class SimpleClient(object):
def __init__(self):
pass
def execute(self, name):
start_time = time.time()
try:
print("do your things cause stress and throw exceptions here.")
time.sleep(1)
except Exception as e:
total_time = int((time.time() - start_time) * 1000)
events.request_failure.fire(request_type="execute", name=name, response_time=total_time, exception=e)
else:
total_time = int((time.time() - start_time) * 1000)
events.request_success.fire(request_type="execute", name=name, response_time=total_time, response_length=0)
class SimpleTasks(TaskSet):
@task
def simple_task(self):
self.client.execute('simple_things')
class SimpleUser(Locust):
def __init__(self, *args, **kwargs):
super(Locust, self).__init__(*args, **kwargs)
self.client = SimpleClient()
task_set = SimpleTasks
min_wait = 1000
max_wait = 10000
然后你可以在你的终端上运行这个,比如:
locust -f simple_test.py --no-web --clients=2 --hatch-rate=2 --num-request=4 --print-stats --only-summary