在计时器上安排蝗虫任务答案

【问题标题】：Scheduling locust tasks on a timer在计时器上安排蝗虫任务
【发布时间】：2020-07-28 19:05:49
【问题描述】：

我希望能够有一个以指定时间间隔运行的任务。例如，假设我想测试 3 个端点，但我想每 30 秒 ping 一次 /status/ 端点

class SomeTasks(TaskSet):
    @task
    self.get("/home/")

    @task
    self.get("/info/")
    
    @task
    self.get("/products/")

    @task
    self.get("/status/")

class SomeUser(FastHttpUser):
    tasks = [SomeTasks]
    wait_time = between(0, 0)

不确定如何运行它，以便它像往常一样随机选择/home/、/info/、/products/，但每 30 秒检查一次状态。也许这可以放在SomeUser 类中？

【问题讨论】：

标签： python performance locust

【解决方案1】：

听起来您希望 ping /status/ 与您的其他用户工作分开。作为测试的一部分，您希望 Locust 生成许多用户重复点击 /home/、/info/、/products/，但您希望单个“用户”每 30 秒 ping /status/ 一次？如果是这样，这是可能的，但不是很简单。我会在正常的 @tasks 之外创建一个单独的函数，该函数在自己的计时器上重复运行，但仍向 Locust 报告。

有多种方法可以做到这一点，但我倾向于使用 Locust events，例如 @events.test_start.add_listener，让 Locust 在您的测试开始时调用一个函数。在函数中，创建一个 greenlet，每 30 秒循环 ping /status/ 端点并向 Locust 报告。下面的基本文件和关于使循环以测试状态为条件的部分是可选的，取自Locust examples。

from locust import HttpUser, TaskSet, task, constant, events
from locust.runners import STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP, WorkerRunner
from locust.clients import HttpSession

import gevent
import time
import os


@events.test_start.add_listener
def on_test_start(environment, **_kwargs):
    gevent.spawn(ping_status, environment)

def ping_status(environment):
    session = HttpSession(base_url=environment.host, request_success=environment.events.request_success, request_failure=environment.events.request_failure)
    while not environment.runner.state in [STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP]:
        time.sleep(30)
        session.get("/status/")

class UserTasks(TaskSet):

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

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

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

class WebsiteUser(HttpUser):
    host = "http://127.0.0.1:8089"
    wait_time = constant(2)
    tasks = [UserTasks]

【讨论】：