前端搬磚工三天入門Locust壓力測試

什麼是Locust

Locust是一個基於Python的帶有可視化圖形界面的測試工具。

本文有什麼

本人不是專業的測試人員，Python也是先學先用的，因此不會涉及到太多的相關專業知識。本文主要是分享本身學習使用Locust的收穫，基於官方文檔和他人的博客，結合本身的使用體驗，因此不是一篇教學或者專業文。你在這篇文章中看不到Locust的嵌套task和多locustfile測試等。

名詞解釋

locust：本意是蝗蟲，在測試過程當中當作是一個用戶便可（後面所謂的HttpLocust只是帶有特殊性質的用戶）。
hatch rate：孵化率，表示每秒鐘產生的用戶數量，能夠模擬一個逐漸增長的過程，能夠根據曲線來考察接口的處理峯值。
TaskSet：顧名思義，是任務的集合，表明着每一個用戶具備的行爲，也就是你想要讓用戶對整個後臺功能作什麼。
max_wait/min_wait：最大等待時間/最小等待時間，這是用戶的休息時間，用戶在執行任務的過程當中，在這兩個值肯定的區間內隨機選擇一個時間進行休息，模擬真實用戶的緩衝，同時也能夠保證某些帶有評率限制的接口可以安全調用，不然可能會出現不少錯誤甚至被封IP等。
locustfile.py：文件名稱固然能夠隨便定義，可是這個文件的功能，至關於創建了一個用戶的模型，須要產生的大量用戶都將以這個文件定義爲原型。

locustfile

locust以這個腳本做爲用戶的原型孵化用戶，這個腳本包括兩個部分，TaskSet子類和Locust子類。TaskSet子類是用於描述用戶行爲的，提供on_start方法執行前置操做。Locust子類主要是使用HttpLocust子類，由於如今的測試主要是針對http的接口，固然也能夠經過一些特別的模塊實現websocket的測試（雖然感受websocket的測試也沒有什麼必要）

Locust/HttpLocust

這個類主要定義用戶的目標（host），至關於一個baseURL，用戶的行爲（請求）都會相對於這個目標進行。定義用戶的休息間隔（max_wait/min_wait）。定義用戶的行爲（TaskSet）。一個普通的HttpLocust多是如下這樣的：

class WebsiteUser(HttpLocust):
    host = "http://juejin.im"
    max_wait = 3000
    min_wait = 3000
    task_set = UserAction
複製代碼

你還能夠在這個類裏面經過類的靜態變量來提供一個全部用戶都能使用的變量。好比一個queue，一個list等。因爲類的靜態變量的特性，每一個用戶修改這個值能夠實現一個遞增的index之類的需求，參數化的邏輯能夠藉助這一點。

TaskSet

這個類負責定義每一個用戶的實際動做，一個用戶可能有不少動做，註冊啦，買東西啦，撤銷啦等等。on_start方法是給每一個用戶開始這些動做以前的一個前置準備，能夠在這個方法中初始化一些東西，好比登陸或者連接websocket。

TaskSet中能夠經過self.client來使用HttpSession class，也就是用戶的http請求能力的供應者，能夠理解成用戶的瀏覽器，用來發送請求和接受響應的。self.client是自帶cookie和session的能力的，徹底至關於一個瀏覽器環境。也就是說self.client發送了一個請求後，好比登陸請求，成功後再用這個self.client發送別的請求，會帶上登陸請求成功後服務器返回的cookie。

TaskSet還提供了一種神奇的東西ResponseContextManager class，當self.client發送的請求帶有catch_response=True的參數，像下面這樣：

with self.client.get("/", catch_response=True) as response:
    if response.content == "":
        response.failure("No data")
        # response.success()
複製代碼

通常的狀況下使用這樣的便可：

response = self.client.get("/",data=data)
複製代碼

這種普通的請求在響應狀態碼小於400的狀況下都認爲成功，然而如今大部分接口都是經過返回的數據來表達你的動做O不OK，這就須要ResponseContextManager經過success和failure來手動修正正確與錯誤的顯示。

TaskSet中self.locust能夠取到HttpLocust的實例，也就是說你在HttpLocust中的某些變量，你是可使用的。固然如何使用仍是取決於你的測試邏輯。

有了以上這些東西，基本就能夠開始定義本身的動做了。首先你須要引入locust提供的task裝飾器，TaskSet子類中的實例方法若是被task裝飾，那麼就會認爲是一個須要執行的動做。@task(weight)是支持權重安排的，也就是說多項任務能夠按照必定的頻繁度來調用，更接近用戶的某些實際體驗，好比購物過程當中翻頁可能要比添加購物車頻繁。

一個普通的TaskSet像是這樣：

class ForumPage(TaskSet):
 @task(100)
    def read_thread(self):
        pass
    
 @task(7)
    def create_thread(self):
        pass
複製代碼

雖然兩個任務什麼都沒幹，可是執行的頻率能夠看出來誰更頻繁。沒有特殊的設置下，兩個任務按照權重隨機的順序執行，也就是有可能在短時間內沒法按照實際的權重執行，也就是說每一個任務之間最好不要有順序以來（可是能夠經過一些操做來保證順序）。

一個普通的task像是這樣：

@task
def test_login(self):
    # 這裏經過類的靜態變量來控制一個全局的變量，保證惟一的index
    # WebsiteUser這就是Locust子類
    WebsiteUser.index += 1
    data = {
        # 這裏組織你的發送數據
        # 經過self.locust來訪問所屬的Locust子類
        "uid": self.locust.index
    }
    with self.client.post("/loginl", data=data, 
        catch_response=True) as response:
        # response提供了一些快捷的方法和屬性，讓你判斷請求的狀況
        if response.ok:
            # 若是響應的content是json格式的，能夠幫你轉成字典
            result = response.json()
            # 經過返回的數據來手動判斷是否成功
            if result['code'] == 200:
                response.success()
            else:
                response.failure(result['msg'])
        else:
            response.failure(bytes.decode(response.content))
複製代碼

咱們跑起來

有了locustfile.py就能夠開始跑起來了，有可視化界面仍是闊以的。命令行跑起來我就很少說了，瀏覽器打開localhost:8089，會讓你輸入總用戶量和孵化率。加入你輸入200和1，就表明每一秒產生一個用戶，直到200個用戶都產生。注意：到達總量時統計數據會重置。因爲用戶的行爲是儘量模擬實際狀況，因此考察數據的時候應當清楚一些。200個用戶，假設有兩個task，比重1：1，min_wait和max_wait都是1000ms，那麼理想狀況下某一時刻，200個用戶都發送了一個請求，且這個請求的響應時間能夠忽略，那麼咱們能夠認爲一秒的併發是200，等待時間是1s，故併發應當穩定在200。因此你須要考慮響應時間和等待時間，選擇合適的用戶數有可能纔到達所須要的併發量。

咱們須要知足其餘的需求

參數化

咱們在測試一個網站的功能時，咱們經過locust模擬的用戶須要不一樣的帳號來進行操做，所以咱們須要在TaskSet中作一些改造。以前咱們說過了TaskSet擁有on_start方法執行前置動做，且self.client能夠保持cookie和session，這不恰好能夠知足登錄後執行操做的需求嗎。那麼下面一個問題就是若是使用不一樣的帳號，這使用Python也是比較容易實現的。咱們在WebsiteUser中定義一個類靜態變量，好比下面這樣：

class WebsiteUser(HttpLocust):
    host = "http://juejin.im"
    max_wait = 3000
    min_wait = 3000
    task_set = UserAction
    # 這個屬性在locust啓動時被初始化好
    # 使用WebsiteUser.user_index來非競爭的更新，實現一個不重複的遞增序列
    user_index = 0 
複製代碼

使用遞增序列的方式，能夠快速註冊一大批用戶，只須要肯定好用戶名和密碼若是使用這個index進行產生便可。固然，若是不想使用這樣的方式，還有一種就是使用數據結構，好比queue：

class WebsiteUser(HttpLocust):
    host = 'http://juejin.im'
    task_set = UserBehavior
    user_data_queue = queue.Queue()
    # 一共300w用戶準備
    for index in range(1500000, 4500000):
        data = {
            # 準備你的註冊或者登錄數據
        }
        user_data_queue.put_nowait(data)
    min_wait = 500
    max_wait = 500
複製代碼

 @task
    def test_register(self):
        try:
            # 用queue來使每一個用戶取得不一樣的帳戶信息
            data = self.locust.user_data_queue.get()
        except queue.Empty:
            # 固然了，只出隊列會致使數據跑空
            # 若是須要循環，用完的數據從隊列的另外一頭塞回去就能夠重複利用了
            print('account data run out, test ended.')
            exit(0)
        with self.client.post('/register', data=data, 
            catch_response=True) as response:
            result = response.json()
            if result['code'] == 200:
                response.success()
            else:
                response.failure(result['msg'])
複製代碼

參數化的一些高級用法能夠參見深刻淺出開源性能測試工具Locust（腳本加強）會給你很多啓發。

非http測試

少數狀況咱們想考量websocket的性能，好比同時鏈接的客戶端以及推送到接受的延遲狀況。Python強大的模塊庫確定是有websocket相關的，經過搜索和嘗試，pip2安裝的websocket模塊和pip3的好像不同，使用pip3安裝的實踐成功。
通常來講咱們在on_start方法中初始化鏈接：

self.ws = websocket.WebSocket()
self.ws.connect("ws://xxx")
複製代碼

而且將ws實例做爲實例屬性方便後續的操做。注意：websocket鏈接端在測試過程當中斷掉也是極有可能的，因此提供一個重連的方法可能會有必要。有了ws的鏈接以後呢，通常須要發送訂閱信息讓服務器來主動推送數據，若是ws在接受了訂閱後會先返回一條訂閱的結果，若是在可視化界面上經過數據統計出來，這就須要locust提供的hook。
from locust import TaskSet, task, HttpLocust, events將events導入進來，而後觸發一些事件，皆可讓locust統計相應的數據：

events.request_success.fire(
    request_type="wss", 
    name="send_entrust", 
    response_time=100,
    response_length=300
)
複製代碼

request_type至關於請求的類型，能夠隨便填。name至關於請求url，response_time至關於響應事件，response_length至關於響應體大小。實際上self.client.post()等也會使用事件觸發來統計數據。好比ResponseContextManager中success的實現：

def success(self):
    """ Report the response as successful Example:: with self.client.get("/does/not/exist", catch_response=True) as response: if response.status_code == 404: response.success() """
    events.request_success.fire(
        request_type=self.locust_request_meta["method"],
        name=self.locust_request_meta["name"],
        response_time=self.locust_request_meta["response_time"],
        response_length=self.locust_request_meta["content_size"],
    )
    self._is_reported = True
複製代碼

這樣的話，設計一個task來發送訂閱並接受推送。做爲前端人員來講，websocket都是用回調來處理推送，那麼Python裏面的websocket怎麼辦呢？實踐成功的websocket模塊鏈接後，也就是從connect()以後，self.ws自己是一個可迭代的，雖然每次next都是調用resv()方法，在循環裏就好使。而self.ws.resv()這個函數是很神奇的，調用開始後他會等到接受推送了再返回，宏觀角度來講這個函數的執行時間，應當等於服務器推送的間隔，一次能夠判斷推送是否出現了延遲。且這個函數若是返回的是空字符串，大體能夠認爲鏈接斷掉，邏輯上增長統計和重連，可以保證一些測試準確度。 給出一個我實際使用的task：

@task
def test_ws(self):
    # 這裏可使用這一個task來完成發送訂閱和處理訂閱
    if not self.send_flag:
        self.ws.send("""{ # 這裏組織你須要發送的訂閱數據 } """)
        result = json.loads(self.ws.recv())
        # 這裏來處理訂閱的結果，若是訂閱成功改變標誌位，讓這個task只執行接受
        if result['code'] == 200:
            self.send_flag = True
            events.request_success.fire(
                request_type="wss", 
                name="send_entrust", 
                response_time=100,
                response_length=300)
        else:
            events.request_failure.fire(
                request_type="wss",
                name="send_entrust",
                response_time=100,
                exception=Exception(self.ws.status),
            )
    else:
        flag = True
        # 因爲推送是不間斷的直到發送取消訂閱，故使用死循環來不斷統計
        while flag:
            # self.ws自己是個迭代器，next和調用resv()是同樣的，均可以
            start_time = time.time()
            resv = next(self.ws)
            # 推送間隔就能夠當作統計數據進行顯示
            total_time = int((time.time() - start_time) * 1000)
            if resv != '':
                events.request_success.fire(
                    request_type="wss", 
                    name="resv_entrust", 
                    response_time=total_time,
                    response_length=59)
            else:
                # 若是中斷了能夠增長重連方案
                events.request_failure.fire(
                    request_type="wss",
                    name="resv_entrust",
                    response_time=total_time,
                    exception=Exception(""),
                )
                flag = False
        else:
            print("no resv")
複製代碼

總結

Locust整體來講仍是比較容易上手的，提供的功能大概可以知足通常的測試需求。實踐的過程當中遇到過一些些問題，好比請求的數量不必定在頁面上統計的及時，用戶數也有可能會斷掉一兩個，具體緣由不明。本人也不是專門的測試人員， 也不是Python native speaker，因此對於Python的理解和對測試分析的觀點不必定對，只是分享一下我在使用的過程當中如何思考問題如何解決問題，解決方案不必定最優，也有可能會形成不少困擾，但這是個人學習過程，想和你們分享，願意和你們討論並繼續學習。