locust–python实现的分布式性能测试工具

1.locust特点：

        1.1 支持Python编写测试用例方案；

        1.2 使用requests发送http请求；

        1.3 使用协程实现，高并发时消耗更低；

        1.4 使用Flask提供 Web UI；

        1.5 有第三方插件支持扩展；

2.创建locust 性能测试用例步骤:

        2.1 创建locust.TaskSet的子类；

        2.2 为用例加上@locust.task()装饰器；

        2.3 使用self.client发送请求；

        2.4 制定await_time属性；

import os

import locust
from locust import task, TaskSet, HttpUser, constant


# 任务集 用户行为脚本
class UserBehavior(TaskSet):  # 必须继承TaskSet
    wait_time = locust.between(1, 2)  # 每个task间隔1~2秒

    @locust.task(1)    #@task装饰该方法表示为用户行为，括号里面参数表示该行为的执行权重：数值越大，执行频率越高，不设置默认是1；
    def getBaidu(self):
        # headers根据自己的接口实际情况填写即可
        headers = {'content-type': 'application/x-www-form-urlencoded',
                   'ua': 'xxx',
                   'Cookie': 'xxx',
                   }
        url = "https://www.baidu.com/"
        res = self.client.get(url, headers=headers)
        print(res.status_code)
        assert res.status_code == 200

    # wait_time = constant(1)  # 每个用户在每次任务执行之间等待5秒


class WebsiteUser(HttpUser):    #WebsiteUser()类用于设置生成负载的基本属性：
    tasks = [UserBehavior]    #task_set：指向定义了用户行为的类
    min_wait = 500    #min_wait：模拟负载的任务之间执行时的最小等待时间，单位为毫秒
    max_wait = 1000    #min_wait：模拟负载的任务之间执行时的最大等待时间，单位为毫秒
    host = "http://localhost:8089"


if __name__ == "__main__":
    os.system("locust -f %s " % __file__)  # 此处导入os，可以在pycharm中直接运行此py文件

3.locust执行压测

3.1通过web UI执行

        locust -f {用例文件.py}

3.2 通过命令行执行

        locust -f {文件名.py} –headless -u 500 -r 10 -t 1h30m [–host 192.168xx]

       -f:指定运行的用例文件；

        –headless:非GUI方式执行；

        -u:指定多少并发用户数；

        -r:指定每秒启动多少用户；

        -t：指定压测运行时间；

        –host：指定base host参数；

4.数据关联：

        通过变量self.token=res.json()[“token”]方式，token可被其他方法使用。

5.参数化：

        引入队列的概念 queue ，实现方式是将参数推入队列，测试时依次取出，全部取完后 locust 会自动停止。若是使用参数循环压测，需要将取出的参数再推入队尾。

import csv
import os, requests
import queue
from locust import TaskSet, task, HttpUser
from requests.packages.urllib3.exceptions import InsecureRequestWarning


# 禁用安全请求警告
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)


def fnReadData():
    f = open("uuid.text", "r") #读取参数文件
    data = []                  #声明空列表
    data = csv.reader(f)      #通过 csv读取文件内容
    s = queue.Queue()         #实例化一个queue对象
    for each in data:         #循环读取open里面的数据
        for key in each:
            try:
                s.put_nowait(key)    #put到队列中
            except queue.Full:
                print("Queue overflow")
    f.close()
    return s


class MyBlogs(TaskSet):
    # 访问我的博客首页
    @task(1)
    def get_blog(self):
        # 定义请求头
        header = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/69.0.3497.100 Safari/537.36"}
        data = self.user.queueData.get()
        req = self.client.get("/357712148/%s" % data, headers=header, verify=False)
        if req.status_code == 200:
            print("success")
        else:
            print("fails")


class httpGet(HttpUser):
    tasks = [MyBlogs]
    min_wait = 3000  # 单位为毫秒
    max_wait = 6000  # 单位为毫秒
    queueData = fnReadData()  # 队列实例化


if __name__ == "__main__":
    #通过好
    os.system("locust -f lcome.py --host=https://blog.51cto.com  --headless -u 1 -r 1 -t 1s")

代码007普通

打赏 收藏 海报 链接