python輕量級性能工具-Locust-有解無憂

Locust基于python的協程機制，打破了執行緒行程的限制，可以能夠在一臺測驗機上跑高并發

性能測驗基礎

　　1.快慢：衡量系統的處理效率：回應時間

　　2.多少：衡量系統的處理能力：單位時間內能處理多少個事務（tps）

性能測驗根據測驗需求最常見的分為下面三類

　　1 負載測驗load testing

　　　　不斷向服務器加壓，值得預定的指標或者部分系統資源達到瓶頸，目的是找到系統最大負載的能力

　　2 壓力測驗

　　　　通過高負載持續長時間，來驗證系統是否穩定

　　3 并發測驗：

　　　　同時像服務器提交請求，目的發現系統是否存在事務沖突或者鎖升級的現象

性能負載模型

locust安裝

安裝存在問題，可以通過豆瓣源下載

pip install locust

locust模板

基本上多數的場景我們都可以基于這個模板read.py去做修改

from locust import HttpUser, TaskSet, task, tag, events

# 啟動locust時運行
@events.test_start.add_listener
def setup(environment, **kwargs):
    # print("task setup")

# 停止locust時運行
@events.test_stop.add_listener
def teardown(environment, **kwargs):
    print("task teardown")

class UserBehavor(TaskSet):
    #虛擬用戶啟用task運行
    def on_start(self):
        print("start")
        locusts_spawned.wait()
    #虛擬用戶結束task運行
    def on_stop(self):
        print("stop")

    @tag('test1')
    @task(2)
    def index(self):
        self.client.get('/yetangjian/p/17320268.html')
    @task(1)
    def info(self):
        self.client.get("/yetangjian/p/17253215.html")

class WebsiteUser(HttpUser):
    def setup(self):
        print("locust setup")

    def teardown(self):
        print("locust teardown")

    host = "https://www.cnblogs.com"
    task_set = task(UserBehavor)
    min_wait = 3000
    max_wait = 5000

注：這里我們給了一個webhost，這樣我們可以直接在瀏覽器中打開locust

集合點lr_rendezvous

當然我們可以把集合點操作放入上述模板的setup中去運行起來

locusts_spawned = Semaphore()
locusts_spawned.acquire()

def on_hatch_complete(**kwargs):
    """
    select_task類的鉤子函式
    :param kwargs:
    :return:
    """
    locusts_spawned.release()

events.spawning_complete.add_listener(on_hatch_complete)
n = 0
class UserBehavor(TaskSet):
    def login(self):
        global n
        n += 1
        print(f"第{n}個用戶登陸")

    def on_start(self):
        self.login()
        locusts_spawned.wait()
    @task
    def test1(self):
        #catch_response獲取回傳
        with self.client.get("/yetangjian/p/17253215.html",catch_response=True):
            print("查詢結束")

class WebsiteUser(HttpUser):
    host = "https://www.cnblogs.com"
    task_set = task(UserBehavor)
    wait_time = between(1,3)

if __name__ == '__main__':
    os.system('locust -f read.py --web-host="127.0.0.1"')

比較常見的用法

在上面兩個例子中我們已經看到了一些，例如裝飾器events.test_start.add_listener；events.test_stop.add_listener用來在負載測驗前后進行一些操作，又例如on_start、on_stop，在task執行前后運行，又例如task，可以用來分配任務的權重

等待時間

# wait between 3.0 and 10.5 seconds after each task
#wait_time = between(3.0, 10.5)
#固定時間等待
# wait_time = constant(3)
#確保每秒運行多少次
constant_throughput（task_runs_per_second）
#確保每多少秒運行一次
constant_pacing(wait_time)

同樣也可以在User類下發重寫wait_time來達到自定義

tag標記

@tag('test1')
@task(2)
def index(self):
    self.client.get('/yetangjian/p/17320268.html')

通過對任務打標記，就可以在運行時候執行運行某一些任務：

#只執行標記test1
os.system('locust -f read.py --tags test1 --web-host="127.0.0.1"')
#不執行標記過的
os.system('locust -f read.py --exclude-tags --web-host="127.0.0.1"')
#除去test1執行所有
os.system('locust -f read.py --exclude-tags test1 --web-host="127.0.0.1"')

自定義失敗

#定義回應時間超過0.1就為失敗
with self.client.get("/yetangjian/p/17253215.html", catch_response=True) as response:
    if response.elapsed.total_seconds() > 0.1:
        response.failure("Request took too long")

#定義回應碼是200就為失敗
with self.client.get("/yetangjian/p/17320268.html", catch_response=True) as response:
    if response.status_code == 200:
        response.failure("回應碼200，但我定義為失敗")

自定義負載形狀

自定義一個shape.py通過繼承LoadTestShape并重寫tick

這個形狀類將以100塊為單位,20速率的增加用戶數，然后在10分鐘后停止負載測驗（從運行開始的第51秒開始user_count會round到100）

from locust import LoadTestShape


class MyCustomShape(LoadTestShape):
    time_limit = 600
    spawn_rate = 20

    def tick(self):
        run_time = self.get_run_time()

        if run_time < self.time_limit:
            # User count rounded to nearest hundred.
            user_count = round(run_time, -2)
            return (user_count, self.spawn_rate)

        return None

運行圖如下所示

通過命令列去觸發

os.system('locust -f read.py,shape.py --web-host="127.0.0.1"')

不同時間階段的例子

from locust import LoadTestShape

class StagesShapeWithCustomUsers(LoadTestShape):

    stages = [
        {"duration": 10, "users": 10, "spawn_rate": 10},
        {"duration": 30, "users": 50, "spawn_rate": 10},
        {"duration": 60, "users": 100, "spawn_rate": 10},
        {"duration": 120, "users": 100, "spawn_rate": 10}]

    def tick(self):
        run_time = self.get_run_time()

        for stage in self.stages:
            if run_time < stage["duration"]:
                tick_data = (stage["users"], stage["spawn_rate"])
                return tick_data

        return None

作者： yetangjian

出處： https://www.cnblogs.com/yetangjian/p/17366389.html

關于作者： yetangjian

本文著作權歸作者和博客園共有，歡迎轉載，但未經作者同意必須保留此段宣告，且在文章頁面明顯位置給出, 原文鏈接如有問題，可郵件（[email protected]）咨詢.

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/551541.html

標籤：其他

上一篇：基于ChatGPT用AI實作自然對話

下一篇：返回列表