wait_time 属性

1、设置固定时间

from loguru import logger
import time
from locust import HttpUser, constant, task


class WebUser(HttpUser):
    # 1、wait_time 属性 模拟指定用户在任务执行期间的等待时间
    # constant指定一个常量
    wait_time = constant(3)

    @task
    def app_list(self):
        logger.info(f"web 访问 ---> /app -")
        logger.info(f"---{time.time()}---")
        self.client.get("/app", name="获取应用列表-web")

2、设置随机时间区间

from loguru import logger
import time
from locust import HttpUser, between, task


class WebUser(HttpUser):
    # 1、wait_time 属性 模拟指定用户在任务执行期间的等待时间
    # between指定一个时间区间
    wait_time = between(1, 5)

    @task
    def app_list(self):
        logger.info("web 访问 ---> /app -")
        logger.info(f"---{time.time()}---")
        self.client.get("/app", name="获取应用列表-web")

3、自适应时间，确保任务每 X 秒运行一次（最多）

from loguru import logger
import time
from locust import HttpUser, task, constant_pacing


class MobileUser(HttpUser):
    host = "https://www.baidu.com"
    weight = 20

    # 指定时间内，一定会执行一次
    wait_time = constant_pacing(3)

    @task
    def app_list(self):
        logger.info("phone 访问 ---> /app")
        logger.info(f"---{time.time()}---")
        self.client.get("/app", name="获取应用列表-phone")

4、自定义等待逻辑

from loguru import logger
import time
from locust import HttpUser, task


class WebUser(HttpUser):
    # 默认等待时间
    last_wait_time = 0
    # 配置虚拟用户的权重
    weight = 60
    host = "https://www.baidu.com"
    
    # 自定义等待时间逻辑
    def wait_time(self):
        self.last_wait_time += 1
        return self.last_wait_time

    @task
    def app_list(self):
        logger.info(f"web 访问 ---> /app - {self.last_wait_time}")
        logger.info(f"---{time.time()}---")
        self.client.get("/app", name="获取应用列表-web")

设置 locust task

1、@task装饰器

from loguru import logger
from locust import HttpUser, task, between


class LoadApiUser(HttpUser):
    # 1、指定host和用户访问接口的休眠时间
    host = "https://www.baidu.com"
    wait_time = between(3, 5)

    # 2、@task装饰器指定权重
    @task(2)
    def user_info(self):
        logger.info("访问 ---> /user")
        # 3、使用requests模块发送http请求，name可以设置别名
        self.client.get("/user", name="获取用户信息")

2、TaskSet任务类

from loguru import logger
from locust import HttpUser, task, between, TaskSet


# 参数为 HttpUser
def app_list(user):
    logger.info("访问 ---> /app")
    user.client.get("/app", name="获取应用列表")


# 参数为 HttpUser
def school_list(user):
    logger.info("访问 ---> /school")
    user.client.get("/school", name="获取学校列表")


class SchoolTaskSet(TaskSet):

    @task(1)
    def school_list(self):
        logger.info("访问 ---> /school")
        self.client.get("/school", name="获取学校列表 class")

    @task(2)
    def user_info(self):
        logger.info("访问 ---> /user")
        # 3、使用requests模块发送http请求，name可以设置别名
        self.client.get("/user", name="获取用户信息 class")


class LoadApiUser(HttpUser):
    # 1、指定host和用户访问接口的休眠时间
    host = "https://www.baidu.com"
    wait_time = between(1, 3)

    # 列表：不指定权重
    # tasks = [app_list, ]
    # 字典：指定权重
    tasks = {app_list: 3, school_list: 1, SchoolTaskSet: 2}

    # 使用 random.choice() 从列表中选取任务

    # 2、@task装饰器指定权重
    @task(1)
    def user_info(self):
        logger.info("访问 ---> /user")
        # 3、使用requests模块发送http请求，name可以设置别名
        self.client.get("/user", name="获取用户信息")

3、tasks属性

from loguru import logger
from locust import HttpUser, task, between


# 参数为 HttpUser
def app_list(user):
    logger.info("访问 ---> /app")
    user.client.get("/app", name="获取应用列表")


# 参数为 HttpUser
def school_list(user):
    logger.info("访问 ---> /school")
    user.client.get("/school", name="获取学校列表")


class LoadApiUser(HttpUser):
    # 1、指定host和用户访问接口的休眠时间
    host = "https://www.baidu.com"
    wait_time = between(1, 3)

    # 列表：不指定权重
    # tasks = [app_list, ]
    # 字典：指定权重
    tasks = {app_list: 3, school_list: 1}

    # 使用 random.choice() 从列表中选取任务

    # 2、@task装饰器指定权重
    @task(1)
    def user_info(self):
        logger.info("访问 ---> /user")
        # 3、使用requests模块发送http请求，name可以设置别名
        self.client.get("/user", name="获取用户信息")

设置任务Tag标签:@tag

from loguru import logger
from locust import HttpUser, task, between, tag

"""
# 排除某些带标签的任务
locust -f locust_task_tag.py --exclude-tags user

# 执行某些带标签的任务
locust -f locust_task_tag.py --tags user app
"""


class LoadApiUser(HttpUser):
    host = "https://www.baidu.com"
    wait_time = between(1, 5)

    # 通过tag装饰器设置标签，并可以同时设置多个
    @tag("user", "P0", "info")
    @task(2)
    def user_info(self):
        logger.info("访问 ---> /user")
        self.client.get("/user", name="获取用户信息")

    @tag("app", "P1", "app_list")
    @task(1)
    def app_list(self):
        logger.info("访问 ---> /app")
        self.client.get("/app", name="获取应用列表")

事件注册

1、负载执行开始和结束执行

from loguru import logger
from locust import events

# test_start, test_stop 设置负载测试的开始或停止时运行一些代码
@events.test_start.add_listener
def on_test_start(**kwargs):
    logger.info("一个新的测试启动")


@events.test_stop.add_listener
def on_test_stop(**kwargs):
    logger.info("一个新的测试关闭")

2、每个工作进程初始化

from loguru import logger
from locust import events
from locust.runners import MasterRunner

# init事件: 每个工作进程执行初始化
@events.init.add_listener
def on_locust_init(environment, **kwargs):
    if isinstance(environment.runner, MasterRunner):
        logger.info("master")
    else:
        logger.info("worker")

3、请求成功或失败执行

from loguru import logger
from locust import events

# 请求成功后执行
@events.request_success.add_listener
def my_success_handler(request_type, name, response_time, response_length, **kw):
    logger.debug("--请求成功--: %s" % name)


# 请求失败后执行
@events.request_failure.add_listener
def my_fail_handler(request_type, name, response_time, response_length, exception, **kw):
    logger.debug(f"--{name} 请求异常 {exception}--")

4、扩展locust的web页面

from locust import events
from flask import jsonify

# 扩展 locust web 页面
@events.init.add_listener
def on_locust_init(web_ui, **kw):
    @web_ui.app.route("/hello")
    def my_added_page():
        return jsonify({"msg": "Hello, World!"})

5、启动后台协程，执行自定义逻辑

import time

import gevent
from locust.runners import STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP, WorkerRunner
from loguru import logger
from locust import events

def checker(environment):
    while environment.runner.state not in [STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP]:
        time.sleep(1)
        if environment.runner.stats.total.fail_ratio > 0:
            logger.error(f"出现异常 {environment.runner.stats.total.fail_ratio}, 直接退出")
            environment.runner.quit()
            return


# 启动一个后台协程，处理自定义逻辑
@events.init.add_listener
def on_locust_init(environment, **_kwargs):
    if not isinstance(environment.runner, WorkerRunner):
        gevent.spawn(checker, environment)

自定义响应验证

from locust import HttpUser, between, task, tag
from loguru import logger

class LoadApiUser(HttpUser):
    host = "https://www.baidu.com"
    wait_time = between(1, 5)

    @tag("app", "P1", "app_list")
    @task(1)
    def app_list(self):
        logger.info("访问 ---> /app")
        with self.client.get("/app", catch_response=True, name="获取应用列表") as response:
            # 检查响应文本
            if response.text != "Success":
                response.failure("Got wrong response")
            # 检查响应时间
            elif response.elapsed.total_seconds() > 0.5:
                response.failure("Request took too long")

    # 4、每个虚拟用户启动时，调用
    def on_start(self):
        self.login()
        logger.info("压测开始")

    def login(self):
        logger.info("模拟用户登陆")
        # self.client.post("/login", name="登陆-获取Token")
        with self.client.get("/login", catch_response=True, name="登陆-获取Token") as response:
            # 自定义成功逻辑
            if response.status_code == 404:
                response.success()