当前位置：首页 > news >正文

性能测试：Locust使用介绍(五)

news 2025/4/30 8:36:44

事件钩子

Locust附带了许多事件钩子，可用于以不同的方式扩展Locust。

例如，以下是如何设置一个事件监听器，该监听器将在请求完成后触发：

from locust import events@events.request.add_listener
def my_request_handler(request_type, name, response_time, response_length, response,context, exception, start_time, url, **kwargs):if exception:print(f"Request to {name} failed with exception {exception}")else:print(f"Successfully made a request to: {name}")print(f"The response was {response.text}")

[!NOTE]

在上面的示例中，通配符关键字参数（**kwargs）将为空，因为我们正在处理所有参数，但如果在Locust的未来版本中添加新参数，它可以防止代码中断。

此外，完全有可能实现一个不为此事件提供所有参数的客户端。例如，非HTTP协议甚至可能没有url或响应对象的概念。从侦听器函数定义中删除任何此类缺失的字段或使用默认参数。

在分布式模式下运行locust时，在运行测试之前在工作节点上进行一些设置可能会很有用。您可以通过检查节点运行器的类型来检查以确保您没有在主节点上运行：

from locust import events
from locust.runners import MasterRunner@events.test_start.add_listener
def on_test_start(environment, **kwargs):if not isinstance(environment.runner, MasterRunner):print("Beginning test setup")else:print("Started test from Master node")@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):if not isinstance(environment.runner, MasterRunner):print("Cleaning up test data")else:print("Stopped test from Master node")

请求上下文

请求事件有一个上下文参数，使您能够传递有关请求的数据（如用户名、标签等）。它可以直接在请求方法的调用中设置，也可以通过重写User.context()方法在用户级别设置。

请求方法的上下文：

class MyUser(HttpUser):@taskdef t(self):self.client.post("/login", json={"username": "foo"})self.client.get("/other_request", context={"username": "foo"})@events.request.add_listenerdef on_request(context, **kwargs):if context:print(context["username"])

用户实例的上下文：

class MyUser(HttpUser):def context(self):return {"username": self.username}@taskdef t(self):self.username = "foo"self.client.post("/login", json={"username": self.username})@events.request.add_listenerdef on_request(context, **kwargs):print(context["username"])

使用catch_response从响应中的值获取上下文：

with self.client.get("/", catch_response=True) as resp:resp.request_meta["context"]["requestId"] = resp.json()["requestId"]

添加Web路由

Locust使用Flask为web UI提供服务，因此很容易将web端点添加到web UI中。通过监听init事件，我们可以检索对Flask应用程序实例的引用，并使用该引用设置新路由：

from locust import events@events.init.add_listener
def on_locust_init(environment, **kw):@environment.web_ui.app.route("/added_page")def my_added_page():return "Another page"