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/"