Locust 负载测试：如何在所有其他 API 中为一个特定 API 分配单个用户

Posted 2023-03-23

【中文标题】Locust 负载测试：如何在所有其他 API 中为一个特定 API 分配单个用户

【英文标题】：Locust load testing : How to allocate single user for one particular API among all other APIs

【发布时间】：2021-09-15 15:16:30

【问题描述】：

我需要使用 locust 测试具有以下场景的 API 集。

有 8 个 HTTP GET API，如下所示。

    1. "/v1/api_1"
    2. "/v1/api_2"
    3. "/v1/api_3"
    4. "/v1/api_4"
    5. "/v1/api_5"
    6. "/v1/api_6"
    7. "/v1/api_7"
    8. "/v1/api_8"

我需要分配 3 个用户，这 3 个用户应该调用前五个 API 这 5 个 API 应该在 15 秒的等待时间和给定的权重下调用。 我需要为最后 3 个 API 分配另外 3 个用户，并且一个用户应该负责 1 个 API。 最后 3 个 API 中的每一个都应每 5 秒调用一次。

详情请看下图

我应该如何使用 locust 实现上述场景。

下面是我的代码。

class API_User(TaskSet):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    @task(10)
    def get_api_1(self):
        print(f"/v1/api_1")
        time.sleep(15)

    @task(50)
    def get_api_2(self):
        print(f"/v1/api_2")
        time.sleep(15)

    @task(20)
    def get_api_3(self):
        print(f"/v1/api_3")
        time.sleep(15)

    @task(10)
    def get_api_4(self):
        print(f"/v1/api_4")
        time.sleep(5)

    @task(10)
    def get_api_5(self):
        print(f"/v1/api_5")
        time.sleep(5)

    @task
    def get_api_6(self):
        print(f"/v1/api_6")
        time.sleep(5)

    @task
    def get_api_7(self):
        print(f"/v1/api_7")
        time.sleep(5)

    @task
    def get_api_8(self):
        print(f"/v1/api_8")
        time.sleep(5)

class API_User_Test(User):
    tasks = API_User

【参考方案1】：

你把它弄得太复杂了。每个单独的 http 调用不必是它自己的任务。您可以将任务设置为一堆调用以及介于两者之间的事情。为您需要运行的每个不同场景（如您所说的“用户”）制定一个任务。这意味着使用适当的睡眠时间调用前五个 API 的单个任务。适当地加权任务，然后启动要运行的 Locust 用户数量。使用您的示例代码：

class API_User(TaskSet):

    @task(3)
    def get_api_1(self):
        print(f"/v1/api_1")
        time.sleep(15)
        print(f"/v1/api_2")
        time.sleep(15)
        print(f"/v1/api_3")
        time.sleep(15)
        print(f"/v1/api_4")
        time.sleep(5)
        print(f"/v1/api_5")
        time.sleep(5)

    @task(1)
    def get_api_6(self):
        print(f"/v1/api_6")
        time.sleep(5)

    @task(1)
    def get_api_7(self):
        print(f"/v1/api_7")
        time.sleep(5)

    @task(1)
    def get_api_8(self):
        print(f"/v1/api_8")
        time.sleep(5)

 class API_User_Test(User):
    tasks = API_User

另外，如果你在__init__ 中除了调用 super 的 init 之外没有做任何其他事情，那么你不需要这样做。

【讨论】：

你能举个例子吗

编辑了我的答案以添加示例