locust入门

简介

locust是python语言的一个开源压测框架, 利用gevent协程来产生大量的并发. 官网: https://locust.io/, 文档: https://docs.locust.io/en/stable/use-as-lib.html

相关文章

https://zhuanlan.zhihu.com/p/118470760

安装

pip install locust

运行

# master
locust -f lc.py  --master --headless

# worker
locust -f lc.py --worker --master-host=127.0.0.1

启动master之后访问 http://localhost:8089/即可看到主界面, 主界面配置和一些参数的解释:

• --headless: no web模式

• Number of total users to simulate: 要模拟的用户总数

• Spawn rate : 每秒产生的用户数,将多少用户添加到当前用户，直到用户总数

• rps: 每秒请求数

• wait_time = between(0, 0) : 用于确定模拟用户在执行任务之间应等待多长时间(自己在测试的时候,用比较少的User然后0,0的between 往往会产生比较高的RPS.因为此时每个任务中间是没有等待时间的, 正常的测试应该设置1秒的等待时间, 使用between(1,1)或constant(1))

web参数解释

# 当设置为:
Number of total users to simulate: 500
Spawn rate: 20

则脚本每秒产生20个用户,并一直到500个为止. 然后开始使用500个用户来产生并发

报告参数解释

参数	解释
Type	请求类型
Name	请求路径
#Request	当前请求的数量
Fails	当前请求失败的数量
Median (ms)	中间值,单位毫秒.
90%ile (ms)	百分之九十的响应时间，因为大于100ms的响应时间被四舍五入，您可能会看到它比最大响应时间大
Average (ms)	所有请求的平均响应时间，毫秒
Min (ms)	请求的最小的服务器响应时间，毫秒
Max (ms)	请求的最大服务器响应时间，毫秒
Average size (bytes)	请求的平均大小,单位字节
Current RPS	当前每秒请求数
Current Failures/s	当前每秒失败的次数

charts

参数	参数
Total Requests per Seconde	吞吐量/每秒响应事务数（rps）实时统计
Response Times(ms)	平均响应时间/平均事务数实时统计
Number of User	虚拟用户数运行

示例代码

from locust import FastHttpUser, task, between, TaskSet

import queue

q = queue.Queue()
tel_data_list = [str(13812120000+i) for i in range(100)]
for i in tel_data_list:
    q.put_nowait(i)


class UserTask(TaskSet):
    headers = {'access-token': 'eyJ0eXAiOiJKV1QiLCJjdHkiOiJKV1QiLCJ6aXAiOiJERUYiLCJhbGciOiJSUzUxMiJ9.eNqEkMtqwzAQRf9l1l7oYatSliktCSQEQrLoqkjWiBhsKchyCTH-98o1bbNql3O491yYEYIe0mXrXYCVH9q2gKHHuNwjmOb-HCzCCl43u_c9FNAPZv0DK1cr7ajjpdNKVKaSlHBjy5zLzWNo59D6_PZyzKRL9XlW28wEJdIRZS2VlCGVVkhtSMmW4n-xEqYC8HZtIp6aLm9QwXn1JBShTKgvxeGKUafwp4bntTqiTr8WxiT5tnQY64v26fE3efRxrYAPjH0TfIbL47yeTXN4-gQAAP__.X8P5sYn6rtKfUzcSor2LMYq7QkI-Luux1L9HVirPw5aaoaQbgsoZTGTYEZrl1-au0lnhieEJPv4Ucj3kDBlZsNoCfxmGSSB6ouYHMDiBcw2dnni4nE5NyPfQpI8PxFaqjTjLfI1idQAPV_r9ZrWS6oZBXunRFCk7fXkkH9AyiJ0', 'merchant-id': '5fc9af1f34fa965b58103bd4', 'bizCode': 'FHL_M', 'channel-id': ''}
    
    @task
    def get_shop_function(self):
        tel = q.get()
        print(f"tel:{tel}")
        
        p = {'src': 'web', 'time': '1631955153000'}
        with self.client.get(
                path="fhl_buyerapi/pub/v1/shop_functions",
                aders=self.headers,
                params=p,
                catch_response=True,
                name="Demo"
        ) as response:
            print(f"response.status_code:{response.status_code}")
            try:
                if response.status_code == 200 and response.json().get("statusCode") != 200:
                    print(f"try:{response.text}")
                    response.failure("Got wrong response")
            except:
                print(response.text)
                response.failure("Got wrong response")


class WebUser(FastHttpUser):
    tasks = [UserTask]
    wait_time = between(5, 10)
    host = "https://shop-stage.ticketmars.com/"