python 3.7安裝locust壓測工具

簡介

• Locust（蝗蟲）是一種易於使用、可編寫腳本且可擴展的性能測試工具。
• 使用python編寫，可以在常規的python代碼中定義用戶的行為。
• 分布式和可拓展，可以支持十萬並發用戶，使用gevent支持協程處理，單個進程可以處理數千個並發用戶，並且開銷低。
• 帶有web用戶界面，實時顯示測試進度，甚至可以在測試運行時更改負載。也可以在沒有UI的情況下運行，易於CI/CD測試。

安裝

使用pip安裝即可（當前版本2.2.1，需要python版本3.6以上）

pip3 install locust

查看版本locust -V

編寫腳本

文件名locustfile.py（文件命名需要為此，或者放在locustfiles文件夾下）

import time
from locust import HttpUser, task, between

# 這里為所有虛擬用戶定義了一個繼承自HttpUser的類，每個虛擬用戶都提供了一個client屬性
# 該屬性是HttpSession的實例，可以用於向我們需要測試的目標發起http請求
class QuickStartUser(HttpUser):
    wait_time = between(1, 5)  # 模擬用戶在每個任務執行后等待1-5秒

    @task  # task 任務，對於每個正在運行的用戶，locust都會創建一個greenlet（協程）
    def hello_world(self):  
        self.client.get("/hello")
        self.client.get("/world")

    @task(3)  # 這是第二個task，后面的3表示權重，運行QuickStartUser時，會從多個task任務中隨機選擇一個，權重增加了他的選擇幾率
    def view_items(self):
        for item_id in range(10):
            self.client.get(f"/item?id={item_id}", name="/item")  # 統計信息是按照URL來分組，這里是為了將這些鏈接都歸於item組內
            time.sleep(1)

    def on_start(self):  # 每個用戶啟動都會調用此方法 on_stop則是每個用戶停止時運行
        self.client.post("/login", json={"username": "foo", "password": "bar"})

當測試開始之后，locust會為每個創建一個HttpSession的實例，每個用戶都運行在自己的green gevent thread下

執行

web ui方式執行
在命令行中輸入locust即可

然后在本地瀏覽器打開http://localhost:8089/，就可以設置用戶數，啟動數，host了。

命令行方式執行

locust --headless --users 10 --spawn-rate 1 -H http://your-server.com

當使用CI服務執行時

locust -f locust_files/my_locust_file.py --headless -u 1000 -r 100
# -f 指定文件
# -u 指定生成的用戶數
# -r 指定生成的速率（即每秒啟動的用戶數）在測試運行時，可以手動更改用戶計數，即使在加速完成后也是如此。按 w 添加 1 個用戶或按 W 添加 10。按 s 刪除 1 或按 S 刪除 10。
# --run-time 指定測試的運行時間 例如--run-time 1h30m  這里時間到立馬關閉，可能還有部分請求沒有完成
# --stop-timeout <seconds> 可以配合上一個使用，等待最后一個任務在指定時間內完成迭代
# --expect-workers 分布式啟動來指定節點

設置退出碼
例如如果滿足以下任何條件，則將退出代碼設置為非零：

• 超過 1% 的請求失敗
• 平均響應時間大於 200 ms
• 響應時間的第 95 個百分位數大於 800 毫秒

import logging
from locust import events

@events.quitting.add_listener
def _(environment, **kw):
    if environment.stats.total.fail_ratio > 0.01:
        logging.error("Test failed due to failure ratio > 1%")
        environment.process_exit_code = 1
    elif environment.stats.total.avg_response_time > 200:
        logging.error("Test failed due to average response time ratio > 200 ms")
        environment.process_exit_code = 1
    elif environment.stats.total.get_response_time_percentile(0.95) > 800:
        logging.error("Test failed due to 95th percentile response time > 800 ms")
        environment.process_exit_code = 1
    else:
        environment.process_exit_code = 0

常用屬性記錄

等待時間屬性
使用wait_time方法，可以很簡單的引入延遲，如果沒有此方法則下一個任務立即執行，這一點有點類似loadrunner中的think_time

• constant固定時間內
• between在最大和最小值之間的隨機時間
• constant_throughput確保任務每秒運行（最多）多少次的自適應時間
• constant_pacing確保任務每多少秒運行一次的自適應時間

例如我想每秒運行500次任務迭代，可以使用wait_time = constant_throughput(1)和500的用戶數
等待時間適用於任務，而不是請求，例如我指定wait_time = constant_throughput(2)，並且每個任務里面有兩個請求，那么每秒請求數（RPS）則為4

權重屬性
如果文件中存在多個用戶類，默認會生成相同數量的每個用戶類。可以通過將類名作為參數傳遞來指定使用哪些類，例如

locust -f locust_file.py WebUser MobileUser

對每個類指定權重

class WebUser(User):
    weight = 3
    ...

class MobileUser(User):
    weight = 1
    ...

主機屬性
主機屬性就是要加載的服務器URL的前綴（類似"http://www.baidu.com"），一般在web ui上的host指定或者命令行中指定--host。
如果在user類中聲明了一個host屬性，那么當沒有--host或者在web ui上的host指定時生效

環境屬性
environment是對用戶正在運行的環境的引用，可以與當前環境進行互動，或者是runner，例如停止運行

self.environment.runner.quit()

如果是獨立的locust實例，則此時整個停止運行。如果是在Node上運行，則會停止這個node

驗證響應
如果 HTTP 響應代碼正常（<400），則認為請求成功，但對響應進行一些額外驗證通常很有用。
可以使用catch_response參數、with語句和對response.failure()的調用將請求標記為失敗

with self.client.get("/", catch_response=True) as response:
    if response.text != "Success":
        response.failure("Got wrong response")
    elif response.elapsed.total_seconds() > 0.5:
        response.failure("Request took too long")

還可以將請求標記為成功，即使響應代碼是錯誤的

with self.client.get("/does_not_exist/", catch_response=True) as response:
    if response.status_code == 404:
        response.success()

甚至可以通過拋出異常然后在with塊之外捕獲它來完全避免記錄請求。或者可以拋出一個locust異常，就像下面的例子一樣，讓Locust捕獲它

from locust.exception import RescheduleTask
...
with self.client.get("/does_not_exist/", catch_response=True) as response:
    if response.status_code == 404:
        raise RescheduleTask()

初始化
在每個用戶之前初始化是on_start函數，在每個用戶之后初始化是on_stop函數。
在每個進程之前初始化是init事件，方便在分布式模式中使用。

from locust import events
from locust.runners import MasterRunner

@events.init.add_listener
def on_locust_init(environment, **kwargs):
    if isinstance(environment.runner, MasterRunner):
        print("I'm on master node")
    else:
        print("I'm on a worker or standalone node")

在整個負載測試開始或者停止時運行可以使用test_start和test_stop事件

from locust import events

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    print("A new test is starting")

@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    print("A new test is ending")

@tag裝飾器
對於@tag裝飾的任務，可以使用--tags和--exclude-tags參數對在測試期間執行的任務進行排除，例如

from locust import User, constant, task, tag

class MyUser(User):
    wait_time = constant(1)

    @tag('tag1')
    @task
    def task1(self):
        pass

    @tag('tag1', 'tag2')
    @task
    def task2(self):
        pass

    @tag('tag3')
    @task
    def task3(self):
        pass

    @task
    def task4(self):
        pass

執行task2和task3為--tags tag2 tag3
排除執行task3為--exclude-tags tag3