运行 Locust 作为库时触发事件挂钩

Question

我正在尝试使用 Locust 库对 API 端点执行负载测试。在这里，我将运行 Locust 作为一个库，而不是使用 locust 命令。我正在尝试执行全局设置和全局拆卸，以便最初创建一个全局状态，供所有用户使用，然后在拆卸时清除（例如，下载一次 S3 文件，然后在结束时将其删除）。

有内置事件挂钩来添加此功能，例如 init and quitting which can be used when running the locustfile using locust command. But, I am unable to trigger these events when running it as a library. Based on the Locust's source code, I can check that these events are fired in locust main.py 文件，但当运行作为库时不会调用它。

当运行它作为一个库时如何添加这样的事件？我尝试了以下两种方法。添加事件侦听器并手动调用 event.fire() 是正确的方法还是直接为其创建和调用自定义方法而不是使用事件是更好的方法？
一般来说，init和quitting事件应该用于初始设置全局状态然后在结束时清除或者test_start和test_stop事件也可以取而代之？

参考源码：

方法 - 1（使用事件挂钩）

import gevent
from locust import HttpUser, task, between
from locust.env import Environment
from locust.stats import stats_printer, stats_history
from locust.log import setup_logging
from locust import events

setup_logging("INFO", None)

def on_init(environment, **kwargs):
    print("Perform global setup to create a global state")

def on_quit(environment, **kwargs):
    print('Perform global teardown to clear the global state')

events.quitting.add_listener(on_quit)
events.init.add_listener(on_init)

class User(HttpUser):
    wait_time = between(1, 3)
    host = "https://docs.locust.io"

    @tas
    def my_task(self):
        self.client.get("/")

    @task
    def task_404(self):
        self.client.get("/non-existing-path")


# setup Environment and Runner
env = Environment(user_classes=[User], events=events)
runner = env.create_local_runner()

### Fire init event and environment and local runner have been instantiated
env.events.init.fire(environment=env, runner=runner)  # Is it correct approach?
 
# start a WebUI instance
env.create_web_ui("127.0.0.1", 8089)

# start a greenlet that periodically outputs the current stats
gevent.spawn(stats_printer(env.stats))

# start a greenlet that save current stats to history
gevent.spawn(stats_history, env.runner)

# start the test
env.runner.start(1, spawn_rate=10)

# in 5 seconds stop the runner
gevent.spawn_later(5, lambda: env.runner.quit())

# wait for the greenlets
env.runner.greenlet.join()

### Fire quitting event when locust process is exiting
env.events.quitting.fire(environment=env, reverse=True) # Is it correct approach?

# stop the web server for good measures
env.web_ui.stop()

方法 - 2（创建自定义方法并直接调用它们）

import gevent
from locust import HttpUser, task, between
from locust.env import Environment
from locust.stats import stats_printer, stats_history
from locust.log import setup_logging
from locust import events

setup_logging("INFO", None)

class User(HttpUser):
    wait_time = between(1, 3)
    host = "https://docs.locust.io"

    @classmethod
    def perform_global_setup(cls):
        print("Perform global setup to create a global state")

    @classmethod
    def perform_global_teardown(cls):
        print('Perform global teardown to clear the global state')

    @task
    def my_task(self):
        self.client.get("/")

    @task
    def task_404(self):
        self.client.get("/non-existing-path")

# setup Environment and Runner
env = Environment(user_classes=[User])
runner = env.create_local_runner()

### Perform global setup 
for cls in env.user_classes:
    cls.perform_global_setup() # Is it correct approach?

# start a WebUI instance
env.create_web_ui("127.0.0.1", 8089)

# start a greenlet that periodically outputs the current stats
gevent.spawn(stats_printer(env.stats))

# start a greenlet that save current stats to history
gevent.spawn(stats_history, env.runner)

# start the test
env.runner.start(1, spawn_rate=10)

# in 5 seconds stop the runner
gevent.spawn_later(5, lambda: env.runner.quit())

# wait for the greenlets
env.runner.greenlet.join()

### Perform global teardown 
for cls in env.user_classes:
    cls.perform_global_teardown() # Is it correct approach?

# stop the web server for good measures
env.web_ui.stop()

Answer 1

两种方法都可以。如果您认为将来可能希望以正常（而不是库）方式运行，则使用事件挂钩更有意义，但如果这种情况不太可能发生，则选择感觉最自然的方法你.
init/quitting 仅在 gui 模式下执行多个运行时才有意义地不同于 test_start/stop（其中 test_start/stop 可能会发生多次).使用适合您在事件处理程序中执行的操作的那个，没有其他指南。

运行 Locust 作为库时触发事件挂钩

Firing Event Hooks when running Locust as a library

python

performance-testing

load-testing

locust