官方文档：What is Locust? — Locust 2.14.2 documentation

webUI模式跑起来没有数据。。。。？？？；

E:\T_Work\other\WB_Locust\my_locustfiles>locust
[2023-02-14 09:57:44,530] PC-20190108TSZQ/INFO/locust.main: Starting web interface at http://127.0.0.1:8089
[2023-02-14 09:57:44,575] PC-20190108TSZQ/INFO/locust.main: Starting Locust 2.14.2
[2023-02-14 09:57:44,708] PC-20190108TSZQ/WARNING/root: Exception in keyboard input poller: (8, 'PeekConsoleInput', '存储空间不足，无法处理
此命令。')
[2023-02-14 10:00:13,088] PC-20190108TSZQ/INFO/locust.runners: Ramping to 5 users at a rate of 1.00 per second
[2023-02-14 10:00:17,090] PC-20190108TSZQ/INFO/locust.runners: All users spawned: {"Boon_User": 1, "JUM_User": 1, "My_App": 1, "My_User": 1, "My_talent_rig": 0, "My_talent_swagger": 0, "Myc_User": 0, "VI_Us
[2023-02-14 10:00:17,090] PC-20190108TSZQ/INFO/locust.runners: All users spawned: {"Boon_User": 1, "JUM_User": 1, "My_App": 1, "My_User": 1,
er": 0, "ZJ_User": 1} (5 total users)
End-My_App
[2023-02-14 10:01:52,372] PC-20190108TSZQ/INFO/locust.runners: Ramping to 1 users at a rate of 1.00 per second
[2023-02-14 10:01:52,373] PC-20190108TSZQ/INFO/locust.runners: All users spawned: {"Boon_User": 1, "JUM_User": 0, "My_App": 0, "My_User": 0,
er": 0, "ZJ_User": 0} (1 total users)
WARNING: gevent: Unhandled error with no watcher
Traceback (most recent call last):
  File "E:\Python\Lib\site-packages\gevent\libuv\loop.py", line 44, in python_queue_callback
    def python_queue_callback(self, watcher_ptr, revents):
KeyboardInterrupt
Traceback (most recent call last):
  File "E:\Python\Lib\site-packages\gevent\_ffi\loop.py", line 270, in python_check_callback
    def python_check_callback(self, watcher_ptr): # pylint:disable=unused-argument
KeyboardInterrupt

原因是gevent（协程，locust 压测要用到的）这个库版本不一致导致的，更新一下：

python.exe -m pip install --upgrade gevent==21.12.0 -i https://mirrors.aliyun.com/pypi/simple/

再次运行就可以打开Web-ui界面了

指定访问页面地址
locust --web-host="127.0.0.1"

 

 

安装Locust

命行模式：python.exe -m pip install locust

代码段主要作用为创建用户或用户组角色【按wait_time、weight自定义选择执行】

User :表示要生成进行负载测试的系统的 HTTP“用户”

---HttpUser：比User类更常用，因为它添加了一个client属性，用来发送HTTP请求

task :表示用户要进行的操作【用户行为】

• 方法前添加@task才会执行操作
• 访问首页 → 登录 → 增、删改查 → homPage

tasks:指定为一个列表，那么每次执行任务时，它将从tasks属性中随机选择

tasks = {My_talent_rig: 1} 
--->每次执行

tasks = {My_talent_rig: 1,My_talent_swagger:5} 
--->My_talent_swagger执行次数是My_talent_rig的五倍

wait_time：时间间隔

• 固定时间， 由constant(wait)函数提供

• 区间随机时间: `between(min_wait, max_wait)函数

• 自适应节奏时间: constant_pacing用于确保任务每 X 秒（最多）运行一次

• 固定吞吐量，由constant_throughput函数提供

weight

        测试中，存在多个User Class，默认情况下locust将为每个User Class的实例的数量是相同的。通过设置weight属性，来控制locust为我们的User Class生成不同数量的实例。

命令行模式启动：locust --headless --users 5 --spawn-rate 1 --run-time 30m --csv=wb_example

框架是通过命令locust运行的，常用参数有：

• -H：指定测试的主机地址（注：会覆盖Locust类指定的主机地址）
• -f：指定测试脚本地址（注：脚本中必须包含一个Locust的衍生类）
• --no-web：不启动web网页，而是直接开始运行测试，需提供属性-c和-r
• -u or users：并发的用户数，与--no-web一起使用
• -r or spawn-rate：每秒启动的用户数，与--no-web一起使用
• -t or run-time：运行时间（单位：秒），与--no-web一起使用
• csv : 保存CSV报告数据，默认保存文件【wb_example_exceptions.csv、wb_example_failures.csv、wb_example_stats.csv、wb_example_stats_history.csv】

import os,sys
pathS = os.getcwd()
Project_Path = os.path.dirname(os.path.dirname(os.path.dirname(pathS)))
root_Path = os.path.dirname(pathS)
sys.path.append(Project_Path)
sys.path.append(root_Path)

from locust import HttpUser, task , between , User,constant,constant_pacing, constant_throughput,events
import os
from other.WB_Locust.my_locustfiles.locustfile01 import B_Set_UP,M_Get
from other.WB_Locust.my_locustfiles.locustfile_php import My_talent_swagger
from other.WB_Locust.my_locustfiles.locustfile_php import My_talent_rig
from other.WB_Locust.my_locustfiles.locustfile01 import My_App

class My_User(User): # 请求一个任务
    host = "http://m-test.xxxxtest.cn"
    wait_time = between(0.5, 5)
    weight = 1
    @task
    def My_User(self):
        pass

class VI_User(User): # 请求一个任务
    host = "http://m-test.xxxxtest.cn"
    wait_time = constant(3)
    weight = 1
    @task
    def VI_User(self):
        pass

class JUM_User(User): # 请求一个任务
    host = "http://m-test.xxxxtest.cn"
    wait_time = constant_pacing(3)
    weight = 2
    @task
    def JUM_User(self):
        pass

class Boon_User(User): # 请求一个任务
    host = "http://m-test.xxxxtest.cn"
    wait_time = constant_throughput(5)
    weight = 2
    @task
    def Boon_User(self):
        pass

class ZJ_User(HttpUser): # 请求一个任务组【TaskSet】
    host = "http://m-test.xxxxtest.cn"
    wait_time = between(2,3)
    weight = 2
    tasks = {B_Set_UP:1}
    @task
    def ZJ_User(self):
        pass

class Myc_User(HttpUser): # 请求一个任务组【TaskSet】
    host = "http://member-service.xxxxtest.cn"
    tasks = {M_Get:2}
    @task
    def Myc_User(self):
        pass
class My_App(HttpUser): # 请求一个任务组【TaskSet】
    host = "http://api-service.xxxxtest.cn"
    tasks = {My_App:1}
    @task
    def My_App(self):
        pass

class My_talent_swagger(HttpUser): 
    wait_time = between(10,20)
    host = "http://member-service.xxxxtest.cn"
    tasks = {My_talent_swagger: 1}
    @task
    def My_talent_swagger(self):
        pass
class My_talent_rig(HttpUser):
    wait_time = between(5,6)
    host = "http://rig-portal.xxxxtest.cn"
    tasks = {My_talent_rig: 1}
    @task
    def My_talent_rig(self):
        pass

if __name__ == '__main__':
    os.system("locust --headless --users 5 --spawn-rate 1 --run-time 30m --csv=wb_example")

# 命令行执行locust
# locust --headless --users 10 --spawn-rate 1 --tun-time 1m -l -H https://m-test.xxxxtest.cn
# --tags WB 只执行标记为WB的任务[debug调试]
# locust --headless --users 10 --spawn-rate 1 --tun-time 1m -l -H https://m-test.xxxxtest.cn --tags WB
# exclude-tags WB 执行标识不为WB的任务

前端--逻辑代码文件

请求值参：catch_response = True 为检查点，结合with使用

@ 多重嵌套任务
class xxx(TaskSet):
        class AAA(TaskSet):
                @task
                def BBB(self):
                    print('000')
        class CCC(TaskSet):
                @task
                def EEE(self):
                    print('123')
                    with self.client.get("/shop/introduce/59/",catch_response=True) as Res:
                        if "万表网" in Res.text:
                            Res.success()
                        else:
                            Res.failure("No member message!")

# locustfile01.py 



import os,sys
pathS = os.getcwd()
Project_Path = os.path.dirname(os.path.dirname(os.path.dirname(pathS)))
root_Path = os.path.dirname(pathS)
sys.path.append(Project_Path)
sys.path.append(root_Path)

from locust import HttpUser, task , between , tag,User,constant,constant_pacing,TaskSet,constant_throughput,events
from other.WB_Locust.common.api_token import *
from other.WB_Locust.Config import Config
import json,time,random,requests

class SSet_UP():
    def START_009(self):
        K_token = WB_ApiMD.MD5_token(self,1,1)
        Api_HeaderToKen = HeaderToken.Header_token(self,3, K_token)
        Member_HeaderToKen = HeaderToken.Header_token(self, 5, K_token)
        # PHP专用
        # PHP_mParams = Config.php_mParams
        # P_token = WB_ApiMD.MD5_token(self, 7, 5)
        # Php_HeaderToKen = HeaderToken.Header_token(self, 2, P_token)
        PHP_mParams = ""
        P_token = ""
        Php_HeaderToKen = ""
        return K_token,Api_HeaderToKen,Member_HeaderToKen,Php_HeaderToKen,P_token,PHP_mParams

class B_Set_UP(TaskSet):
    @task
    class Test_HttpUser(TaskSet):
        @tag('42470')
        @task(3)
        def view_goods(self):
            self.client.get("/goods/42470")

        @tag('shoubiao')
        @task(2)
        def view_shoubiao(self):
            self.client.get("/shoubiao.html")

        @tag('fenlei')
        @task(1)
        def view_fenlei(self):
            self.client.get("/fenlei.html")
            self.interrupt()

    @task
    class Woman_UserA(TaskSet):
        @task
        def Woman_UserA(self):
            self.client.get("/shop/59/")
            self.interrupt()

    @task
    class Woman_UserB(TaskSet):
        @task
        def Woman_UserB(self):
            self.client.get("/goods/42471")

        @task
        def Woman_UserC(self):
            self.client.get("/goods/67")
            self.interrupt()

    @task
    class Woman_UserS(TaskSet):
        @task
        def Woman_UserB1(self):
            self.client.get("/goods/16641")

        @task
        def Woman_UserC2(self):
            self.client.get("/goods/42823")
            self.interrupt()

    @task
    class SWoman_UserS(TaskSet):
        '商城静态页面接口请求'
        @tag('WB')
        @task(2)
        def SWoman_UserB1(self):
            with self.client.get("/shop/introduce/59/",catch_response=True) as Res:
                if "万表网" in Res.text:
                    Res.success()
                else:
                    Res.failure("No member message!")
        @tag('WB')
        @task(2)
        def SWoman_UserC2(self):
            with self.client.get("/shop/dynamic/59/",catch_response=True) as Res:
                if "万表网" in Res.text:
                    Res.success()
                else:
                    Res.failure("No member message!")
            self.interrupt()

class M_Get(TaskSet):
    '会员资料接口请求'
    def on_start(self):
        Data_key = SSet_UP().START_009()
        self.My_y = Data_key[2]

    def on_stop(self):
        self.client.get('/member/loginOut',headers=self.My_y)

    @tag('WB')
    @task(5)
    def login(self):  # 会员登录接口请求
        Url_headers = {'Content-Type': 'application/json', 'Connection': 'keep-alive',
                    'User-Agent': 'Dalvik/2.1.0 (Linux; U; Android 8.1.0; Redmi Note 5 MIUI/V9.6.4.0.OEICNFD) WBIAO_MARKET_20190471_8.1.0 com.wbiao.wbapp 3.4.9#huawei# runningIn_1'}
        Pcontent = {"identity": '188000000' + str(random.randint(00,99)), "identityType": "1", "password": 111111, "platformSource": "Mall",
                    "remember": "true", "terminalType": "PC"}
        Url_loring = 'http://member-service.xxxxtest.cn/member/login'
        with self.client.post(url=Url_loring,data=json.dumps(Pcontent),catch_response=True,headers=Url_headers) as Res:
            if "memberCode" in Res.text:
                Res.success()
            else:
                Res.failure("Wrong")

    @tag('SH')
    @task
    def member_get(self):  # 获取个人资料
        with self.client.get('/member/center/get',catch_response=True, headers=self.My_y) as Res:
            if "5102252" in Res.text:
                Res.success()
            else:
                Res.failure("No member message!")

    @tag('SH')
    @task
    def member_message(self):  # 获取会员基本信息
        with self.client.get('/member/center/getMember',
                             catch_response=True,
                             headers=self.My_y) as Res:
            if "5102252" in Res.text:
                Res.success()
            else:
                Res.failure("No member message!")

class My_App(TaskSet): #api-APP请求
    def on_start(self):
        Data_key = SSet_UP().START_009()
        self.K_token = Data_key[0]
        self.HeaderToKen = Data_key[1]
    def on_stop(self):
        print("End-My_App")

    @tag('debug')
    @task
    def My_Cart(self):
        app_mParams = ""
        api_key = WB_ApiMD.api_get(app_mParams,'1','1')
        K_word = api_key + '&wb-token=' + self.K_token + Config.T_keyword
        with self.client.get('/api/cart/getList?' + api_key + WB_ApiMD.sign_R(K_word),
                             headers=self.HeaderToKen,
                             catch_response=True) as Res:
            if "cartGoodsList" in Res.text:
                Res.success()
            else:
                Res.failure("Faile")
'''
class My_talent_swagger(TaskSet): # Member swagger
    def on_start(self):
        Data_key = SSet_UP().START_009()
        self.K_token = Data_key[0]
        self.HeaderToKen = Data_key[3]
    def on_stop(self):
        self.interrupt()
    @task
    def talent_get(self): # 达人-获取图文信息
        with self.client.get('/member/talentImageText/get?talentImageTextId=122',
                             headers=self.HeaderToKen,
                             catch_response=True) as Res:
            if "talentImageTextId" in Res.text:
                Res.success()
            else:
                Res.failure("Wrong")
    @task
    def talent_list(self): # 达人-查询列表
        list_data = {
          "pager": {
            "now": 0,
            "size": 5,
            "total": 5
          },
          "search": [
            {
              "name": "string",
              "value": "string"
            }
          ],
          "sort": [
            {
              "asc": 'true',
              "name": "string"
            }
          ]
        }
        with self.client.post('/member/talentImageText/getList',
                              data=json.dumps(list_data),
                              headers=self.HeaderToKen,
                              catch_response=True) as Res:
            if "talentImageTextId" in Res.text:
                Res.success()
            else:
                Res.failure("Wrong")

class My_talent_rig(TaskSet): # Rig Console
    def on_start(self):
        Data_key = SSet_UP().START_009()
        self.K_token = Data_key[4]
        self.HeaderToKen = Data_key[3]
        self.php_mParams = Data_key[5]
    def on_stop(self):
        self.interrupt()
    @task
    def rig_t_get(self): # 达人类目详情
        with self.client.get('/back/member/talent/class/get?id=1'+ self.php_mParams +'&sign='+self.K_token,
                             headers=self.HeaderToKen,
                             catch_response=True) as Res:
            if 'name' in Res.text:
                Res.success()
            else:
                Res.failure('Wrong')

    @task
    def rig_t_list(self): # 达人类目列表
        with self.client.get('/back/member/talent/class/list?' + self.php_mParams + '&sign=' + self.K_token,
                             headers=self.HeaderToKen,
                             catch_response=True) as Res:
            if 'name' in Res.text:
                Res.success()
            else:
                Res.failure('Wrong')

    @task
    def rig_t_highCommissionGoods(self):  # 高佣金比例的商品列表
        Pcontent = {"brandCodes":[70,176],"goodsCode":"","models":""}
        with self.client.post('/back/member/talent/highCommissionGoods/list?' + self.php_mParams + '&sign=' + self.K_token,
                              data= json.dumps(Pcontent),
                              headers=self.HeaderToKen,
                              catch_response=True) as Res:
            if 'name' in Res.text:
                Res.success()
            else:
                Res.failure('Wrong')
'''

# class My_cart(HttpUser):
#     host = "http://api-service.xxxxtest.cn"
#     tasks = {My_cart:1}
# if __name__ == '__main__':
#     os.system("locust -f locustfile01.py --headless --users 1 --spawn-rate 1 --run-time 10")

后台-逻辑代码文件

# locustfile_php.py



import os,sys
pathS = os.getcwd()
Project_Path = os.path.dirname(os.path.dirname(os.path.dirname(pathS)))
root_Path = os.path.dirname(pathS)
sys.path.append(Project_Path)
sys.path.append(root_Path)

from locust import HttpUser, task , between , tag,User,constant,constant_pacing,TaskSet,constant_throughput,events
from other.WB_Locust.common.api_token import *
from other.WB_Locust.Config import Config
import json,time,random,requests

class SSet_UP():
    def START_009(self):
        # K_token = WB_ApiMD.MD5_token(self,1,1)
        # Api_HeaderToKen = HeaderToken.Header_token(self,3, K_token)
        # Member_HeaderToKen = HeaderToken.Header_token(self, 5, K_token)
        # PHP专用
        PHP_mParams = Config.php_mParams
        P_token = WB_ApiMD.MD5_token(self, 7, 5)
        Php_HeaderToKen = HeaderToken.Header_token(self, 2, P_token)
        # Rig_HeaderToKen = HeaderToken.Header_token(self, 2, K_token)
        # app_mParams = ""
        # api_key = WB_ApiMD.api_get(app_mParams, str(1), str(1))
        # K_word = api_key + '&wb-token=' + K_token + Config.T_keyword
        # Res = requests.get(url='http://api-service.xxxxtest.cn/api/cart/getList?' + api_key + WB_ApiMD.sign_R(K_word),
        #                    headers=HeaderToKen)
        # print(Res.text)
        # print(Res.url)
        # print(api_key)
        # print(K_word)
        # return K_token,Api_HeaderToKen,Member_HeaderToKen,Php_HeaderToKen,P_token,PHP_mParams
        return Php_HeaderToKen, P_token, PHP_mParams

class My_talent_swagger(TaskSet): # Member swagger
    def on_start(self):
        print("My_talent_swagger")
        Data_key = SSet_UP().START_009()
        self.HeaderToKen = Data_key[0]
    def on_stop(self):
        self.interrupt()
    @task
    def talent_get(self): # 达人-获取图文信息
        with self.client.get('/member/talentImageText/get?talentImageTextId=122',
                             headers=self.HeaderToKen,
                             catch_response=True) as Res:
            if "talentImageTextId" in Res.text:
                Res.success()
            else:
                Res.failure("Wrong")
    @task
    def talent_list(self): # 达人-查询列表
        list_data = {
          "pager": {
            "now": 0,
            "size": 5,
            "total": 5
          },
          "search": [
            {
              "name": "string",
              "value": "string"
            }
          ],
          "sort": [
            {
              "asc": 'true',
              "name": "string"
            }
          ]
        }
        with self.client.post('/member/talentImageText/getList',
                              data=json.dumps(list_data),
                              headers=self.HeaderToKen,
                              catch_response=True) as Res:
            if "talentImageTextId" in Res.text:
                Res.success()
            else:
                Res.failure("Wrong")

class My_talent_rig(TaskSet): # Rig Console
    def on_start(self):
        print("My_talent_rig")
        Data_key = SSet_UP().START_009()
        self.K_token = Data_key[1]
        self.HeaderToKen = Data_key[0]
        self.php_mParams = Data_key[2]
    def on_stop(self):
        self.interrupt()
    @task
    def rig_t_get(self): # 达人类目详情
        with self.client.get('/back/member/talent/class/get?id=1'+ self.php_mParams +'&sign='+self.K_token,
                             headers=self.HeaderToKen,
                             catch_response=True) as Res:
            if 'name' in Res.text:
                Res.success()
            else:
                Res.failure('Wrong')

    @task
    def rig_t_list(self): # 达人类目列表
        with self.client.get('/back/member/talent/class/list?' + self.php_mParams + '&sign=' + self.K_token,
                             headers=self.HeaderToKen,
                             catch_response=True) as Res:
            if 'name' in Res.text:
                Res.success()
            else:
                Res.failure('Wrong')

    @task
    def rig_t_highCommissionGoods(self):  # 高佣金比例的商品列表
        Pcontent = {"brandCodes":[70,176],"goodsCode":"","models":""}
        with self.client.post('/back/member/talent/highCommissionGoods/list?' + self.php_mParams + '&sign=' + self.K_token,
                              data= json.dumps(Pcontent),
                              headers=self.HeaderToKen,
                              catch_response=True) as Res:
            if 'name' in Res.text:
                Res.success()
            else:
                Res.failure('Wrong')



# class My_cart(HttpUser):
#     host = "http://api-service.xxxxtest.cn"
#     tasks = {My_cart:1}
# if __name__ == '__main__':
#     os.system("locust -f locustfile01.py --headless --users 1 --spawn-rate 1 --run-time 10")

执行

locust --headless --users 5 --spawn-rate 1 --run-time 30m --csv=wb_example

• Type： 请求的类型，例如GET/POST。

• Name：请求的路径。

• reqs：当前请求的数量。

• Fails：当前请求失败的数量。

• Med：中间值，单位毫秒，一半的服务器响应时间低于该值，而另一半高于该值。

• Avg：平均值，单位毫秒，所有请求的平均响应时间。

• Min：请求的最小服务器响应时间，单位毫秒。

• Max：请求的最大服务器响应时间，单位毫秒。

• Average size：平均请求的大小，单位字节。

• Req/s：是每秒钟请求的个数。

• Failures/s：瞬时每秒失败数。

WebUI启动，均需要在命令行切换到执行文件的目录下，执行【locust】,对应打开链接即可