Skip to main content
 首页 » 编程设计

python之使用 Locust 自定义客户端支持编写 websocket 服务器负载测试的示例

2025年05月04日33dudu

我需要一个例子或解释如何使用自定义客户端(在我的例子中是 WebSocket 服务器)编写蝗虫负载测试。看到locust中给出的解释documentation但我不知道函数 __getattr__ 和 def wrapper(*args, **kwargs): 究竟是如何通过蝗虫触发钩子(Hook)事件的。

请您参考如下方法:

我今天在玩这个,还发现自定义示例有点棘手,因为很难说出神奇的地方。这是客户端和任务以及用户交互的一个更简单的版本。我没有让装饰器捕获任何客户端的任意方法,而是让它只能做一件事,这样您就可以看到计时和错误处理发生的位置,以及它如何使用蝗虫事件 Hook 发送成功和失败。

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