性能测试工具-locust

news2025/1/10 9:42:15

简介

Locust是基于python语言的性能测试框架。

Locust支持分布式部署，单实例并发线程数可达1000，界面简约。

locust基于事件。

Locust的缺点

1、无可视化脚本编写功能，需要基于Python语言和locust框架进行脚本编写，纯代码编写。

2、需要安装Python环境。

环境搭建

Python安装和pip安装

Locust安装

数码内网安装：pip install locust。可以使用国内源安装

Locust使用

脚本编写

示例1

代码

from locust import HttpUser, task, between
from common.common import logger_print
import yaml
import time


class MyTest(HttpUser):
    wait_time = between(1, 1)
    with open('../config/config.yaml', 'r', encoding='utf-8') as f:
        config = yaml.safe_load(f)
        f.close()
    logger_file = config["logger"]["logger_file"]
    logger_level = config["logger"]["logger_level"]
    logger = logger_print(logger_file, logger_level)
    @task
    def test_demo(self):
        url = "***"
        headers = {'Content-Type': 'application/json;charset-utf-8'}
        json_data = {"facility": "qqg***"}
        start_time = time.time()
        with self.client.post(url=url, headers=headers, json=json_data, catch_response=True) as response:
            run_time = time.time() - start_time
            self.logger.info(response.text)
            self.logger.info("本次请求耗时: %.2f ms" % (run_time * 1000))
            try:
                if response.json()["status"] == 1 and response.json()["message"] == "查询**统计成功！":
                    response.success()
                    self.logger.info("请求消息体返回正确")
                else:
                    response.failure("返回消息不正确, %s" % response.text)
                    self.logger.error("返回消息不正确, %s" % response.text)
            except Exception as e:
                response.failure("返回消息不正确, %s, 报错: %s" % (response.text, e))
                self.logger.error("返回消息不正确, %s, 报错: %s" % (response.text, e))

说明

1、locust脚本需要编写测试类并继承HttpUser；

2、测试用例需要使用task进行标注；

3、需要设置单个请求发起的间隔时间，wait_time = between(1, 1)，between()方法入参最小等待时长和最大等待时长；

4、断言添加需要使用with self.client.post(url=url, headers=headers, json=json_data, catch_response=True) as response: ；

5、断言支持自定义，并使用response.success()和response.failure()进行标注成功或失败。

示例2

代码

from locust import HttpUser, TaskSet, task
from common.common import logger_print
import yaml
import time


class MyTestCases(TaskSet):
    with open('config/config.yaml', 'r', encoding='utf-8') as f:
        config = yaml.safe_load(f)
        f.close()
    logger_file = config["logger"]["logger_file"]
    logger_level = config["logger"]["logger_level"]
    logger = logger_print(logger_file, logger_level)
    @task
    def test_demo(self):
        url = "***"
        headers = {'Content-Type': 'application/json;charset-utf-8'}
        json_data = {"facility": "***"}
        start_time = time.time()
        with self.client.post(url=url, headers=headers, json=json_data, catch_response=True) as response:
            run_time = time.time() - start_time
            self.logger.info(response.text)
            self.logger.info("本次请求耗时: %.2f ms" % (run_time * 1000))
            try:
                if response.json()["status"] == 1 and response.json()["message"] == "查询**统计成功！":
                    response.success()
                    self.logger.info("请求消息体返回正确")
                else:
                    response.failure("返回消息不正确, %s" % response.text)
                    self.logger.error("返回消息不正确, %s" % response.text)
            except Exception as e:
                response.failure("返回消息不正确, %s, 报错: %s" % (response.text, e))
                self.logger.error("返回消息不正确, %s, 报错: %s" % (response.text, e))

class WebSitUser(HttpUser):
    tasks = [MyTestCases]
    min_wait = 1000
    max_wait = 1000