locust性能测试脚本模板

locust性能测试脚本模板

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import time

from locust import HttpLocust, TaskSet, task, seq_task


class BlogDemo(TaskSet):

    def get_parameters(self):
        # 定义获取参数
        pass

    def regx(self):
        # 定义正则表达式
        pass

    @seq_task(2)  # 第二个执行
    @task(1)
    def open_blog(self):
        # 定义requests的请求头和url
        url = "/111testing"
        header = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/68.0.3440.106 Safari/537.36"}

        r = self.client.get(url,  headers=header, name="进入我的博客")
        print(r.status_code)
        assert r.status_code == 200

    @seq_task(1)  # 第一个执行
    @task(1)
    # @task(1) 和 @task(2) 运行的请求数量是1:2
    def open_blog_firstPage(self):
        url = "/"
        header = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/68.0.3440.106 Safari/537.36"}

        self.client.get(url,  headers=header, name="进入博客园首页")


class websitUser(HttpLocust):
    task_set = BlogDemo
    # 等待的区域时间不写的话，默认是1秒
    min_wait = 3000  # 单位毫秒
    max_wait = 6000  # 单位毫秒
    host = "https://www.cnblogs.com"


if __name__ == "__main__":
    import os

    # 如果上面写了host，那么os.system这里就不用再写host了
    os.system("locust -f demo.py")

    # 如果上面没有写host，那么os.system这里就需要写host
    # os.system("locust -f demo.py --host=https://www.cnblogs.com")

    # 本机为主测试机时运行下面命令，单独的主测试机运行起来也没有用（无法跑并发请求的），必须结合子测试机使用
    # os.system("locust -f demo.py --master")

    # 本机为子测试机时运行下面命令，运行子测试机的前提是必须已经运行了主测试机
    # os.system("locust -f demo.py --slave")

    # 本机为子测试机时，且主测试机在另外一台机器时运行下面命令
    # os.system("locust -f demo.py --slave --master-host=192.168.xx.xxx")

    # 一般主测试机多放在windows系统运行，子测试机多放在Linux系统运行，因为windows运行locust性能很差
    # 运行一次请求机型调试：locust -f demo.py --no_web -c 1 -n 1

locust集合点设置

有时测试需要让所有并发用户完成初始化后再进行压力测试，这就需要类似于LoadRunner中的集合点的概念，由于框架本身没有直接封装，可借助python的gevent库来实现