Locust快速上手指南

本文首發於：行者AI

因爲目前公司性能測試需求較多，因此調研了目前比較流行的幾款壓測工具，因爲Jmeter與LoadRunner基於多線程實現併發，多線程由操做系統決定，因爲上下文切換頻繁，內核調度頻繁致使單臺機器很難產生大量的線程併發。以多線程方式運行會有不少線程切換的開銷致使資源的浪費，故而考慮用多協程的方式實現，Jmeter由Java語言編寫但Java不支持協程機制。 Python語言經過async/await的方式實現協程，剛好Locust正是基於python。因此此次把側重點放在Locust性能測試工具。

首先，咱們要明白只有產生壓力才能評估項目的性能，性能測試關鍵是要經過工具產生大量的併發，而併發的強弱取決於工具的實現原理，因爲Locust比較輕量，腳本設計較爲靈活，不少重量級壓測工具能夠實現的功能在Locust中一樣能夠實現。且python語言的locust則能夠以更輕量級的方式實現高併發，因此接下來將經過重要功能的介紹與代碼示例來帶你們看下如何快速上手Locust。

1. Locust的特色

（1）基於python開發腳本；
（2）開源免費,能夠二次開發；
（3）分佈執行。配置master和slave(主從機器)，在多要機器上對系統持續發起請求；
（4）基於事件驅動。與其餘工具使用進程和線程來模擬用戶不一樣，Locust藉助gevent庫對協程的支持，能夠達到更高數量級的併發；
（5）不支持監控被測機器，須要配合其餘工具的輔助；
（6）在Locust類中，具備一個client屬性，對應着虛擬用戶做爲客戶端所具有的請求能力，也就是咱們常說的請求方法；因此在使用Locust時，須要先繼承Locust類，而後在繼承子類中的client屬性中綁定客戶端的實現類；
（7）HttpUser使用到了requests.Session，所以後續全部任務執行過程當中就都具備登陸態；
（8）版本變更：1.0版本以後的更新重點是將HttpLocust替換爲Httpuser，task_set任務集須要數據類型爲列表類型，且task_set須要修改成tasks。

2. Locust的指標體系及經常使用使用流程

（1）響應時間：反應系統處理效率指標，從開始到完成某項工做所須要時間的度量，響應時間一般隨着負載的增長而增長；
（2）吞吐量：反應系統處理能力的指標，指單位時間內完成工做的度量，能夠從客戶端或服務端視角兩方面來進行綜合評估；
（3）事務處理能力（TPS在locust中爲RPS）：對一筆業務進行處理時的相應狀況，一般包含三個指標，一是處理該業務的響應時間，二是處理該業務的成功率，三是單位時間內（每秒鐘，每分鐘，每小時等）能夠處理的業務數量。

cmd命令執行腳本

web界面操做(web界面不會自動中止，須要手動stop)；
進入到項目目錄，py文件這一層級；
locust -f test.py 或者 locust -f test.py --host=http://example.com；
打開瀏覽器進入web界面 添入 模擬的用戶總數和每秒啓動的虛擬用戶數；
http://localhost:8089；
測試結果界面：

純命令運行

locust -f test.py --no-web -c 100 -r 20 -t 5 或者 locust -f test.py --host=http://example.com --no-web -c 100 -r 20 -t 5；
-c: 用戶數量；
-r: 每秒生成數量；
-t: 限制運行時間；
-n: 請求總次數；

3. Locust的語法格式

（1）定義一個任務類，這個類名稱本身隨便定義；
（2）繼承SequentialTaskSet 或 TaskSet類，因此要從locust中，引入SequentialTaskSet或TaskSet；
（3）當類裏面的任務請求有前後順序時繼承SequentialTaskSet類；
（4）沒有前後順序，可使用繼承TaskSet類；

import random
    from locust import HttpUser, task, between, SequentialTaskSet, tag
    class MyTaskCase(SequentialTaskSet):
            # 初始化方法，至關於 setup
            def on_start(self):
                pass
    
        # @task python中的裝飾器，告訴下面的方法是一個任務，
        # 這個裝飾器和下面的方法被複制屢次，改動一下，就能寫出多個接口
        # 裝飾器後面帶上(數字)表明在全部任務中，執行比例
        # 要用這個裝飾器，須要頭部引入 從locust中，引入 task
        @task
        @tag("leave_1")
        def regist_(self):  # 一個方法， 方法名稱能夠本身改
            url = '/erp/regist'  # 接口請求的URL地址
            # 定義請求頭爲類變量，這樣其餘任務也能夠調用該變量
            self.headers = {"Content-Type": "application/json"}
            self.username = "locust_" + str(random.randint(10000, 100000))
            self.pwd = '1234567890'
            # post請求的 請求體
            data = {"name": self.username, "pwd": self.pwd}
            # 使用self.client發起請求，請求的方法根據接口實際選,
            # catch_response 值爲True 容許爲失敗 ， 
            # name 設置任務標籤名稱   -----可選參數
            with self.client.post(url,
                                  json=data,
                                  headers=self.headers,
                                  catch_response=True) as rsp:
                if rsp.status_code > 400:
                    print(rsp.text)
                    rsp.failure('regist_ 接口失敗！')
    
        @task  # 裝飾器，說明下面是一個任務
        def login_(self):
            url = '/erp/loginIn'  # 接口請求的URL地址
            data = {"name": self.username, "pwd": self.pwd}
            with self.client.post(url,
                                  json=data,
                                  headers=self.headers,
                                  catch_response=True) as rsp:
                # 提取響應json 中的信息，定義爲 類變量
                self.token = rsp.json()['token']
                if rsp.status_code < 400 \
                        and rsp.json()['code'] == "200":
                    rsp.success()
                else:
                    rsp.failure('login_ 接口失敗！')
    
        @task  # 裝飾器，說明下面是一個任務
        def getuser_(self):
            url = '/erp/user'  # 接口請求的URL地址
            # 引用上一個任務的 類變量值   實現參數關聯
            headers = {"Token": self.token}  
           # 使用self.client發起請求，請求的方法 選擇 get
            with self.client.get(url, 
                                 headers=headers, 
                                 catch_response=True) as rsp: 
                if rsp.status_code < 400:
                    rsp.success()
                else:
                    rsp.failure('getuser_ 接口失敗！')
    
        # 結束方法， 至關於teardown
        def on_stop(self):
            pass
    
    # 定義一個運行類 繼承HttpUser類， 因此要從locust中引入 HttpUser類
    class UserRun(HttpUser):
        tasks = [MyTaskCase]
        # 設置運行過程當中間隔時間 須要從locust中 引入 between
        wait_time = between(0.1, 3)

4. 單接口壓測示例

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from locust import task,TaskSet,HttpUser
    
    class UserTasks(TaskSet):
        # 聲明下面是一個任務
        @task
        def getIndex(self):
            # self.client是TaskSet的成員，至關於一個request對象
            self.client.get("/path")
            print("here")
    class WebUser(HttpUser):
        # 聲明執行的任務集是哪一個類
        tasks = [UserTasks]
        # 最小等待時間和最大等待時間   請求間的間隔時間
        min_wait = 1000
        max_wait = 2000

運行：

在終端中輸入：locust -f 被執行的locust文件.py --host=http://被測服務器域名或ip端口...，也能夠不指定host，如 "locust -f locust_test.py --host=http://localhost:8082"；當命令執行成功，會提示服務端口，如：*：8089。此時，則可經過瀏覽器訪問機器ip:8089,看到任務測試頁面；

Number of total users to simulate 模擬的用戶數
Spawn rate (users spawned/second) 每秒產生的用戶數

5. 業務用例壓測示例

下面以一個登陸接口和獲取id的接口爲例

# #!/usr/bin/env python
   # # -*- coding: utf-8 -*-
   from locust import task,TaskSet,HttpUser,tag
   from random import randint
   import json
   class UserTasks(TaskSet):
   
       def on_start(self):
       # 準備測試，請求接受的大可能是字典格式
           self.loginData = [{"username":"1","paswordword":"1"},
                             {"username":"2","paswordword":"2"},
                             {"username":"3","paswordword":"3"}]
       print("-----------on_start----------------")
   
       # 聲明下面是一個登陸任務
       @task(1)
       def doLogin(self):
           ranIndex = randint(1,len(self.loginData))
           # 發送請求並響應
           response = self.client.post("/path",data = self.loginData[ranIndex],catch_response=True)
           if "login-pass" in response.text:
               response.success()
           else:
               response.failure("Can not login!")
   
       # 聲明下面是一個獲取商品任務
       @task
       def get_goods(self):
           body = {"good_name" :"apple"}
           # 發送請求並響應
           response = self.client.post("/path2",data = body,catch_response=True)
           newText = json.loads(response.txt)
           if "game_id" in newText[0].keys():
               response.success()
           else:
               response.failure("Can not get!")
       
       #爲任務分配權重
       tasks = {doLogin:1, get_goods:2}
       
   #增長 tag 標籤，在執行時，能夠用 -T \ --tags 指定標籤執行、-E \ --exclude-tags 排除指定標籤執行
   class WebUser(User):
       @task
       @tag("tag1", "tag2")
       def my_task(self):
           pass

   class WebUser(HttpUser):
       # 聲明執行的任務集是哪一個類，任務集中的任務按已分配的1:2權重執行
       tasks = [UserTasks]
       # 最小等待時間和最大等待時間   請求間的間隔時間
       min_wait = 1000
       max_wait = 2000

   # locust -f locust_test.py --host=http://localhost:8082
   # Number of total users to simulate   模擬的用戶數
   # Spawn rate (users spawned/second)   每秒產生的用戶數

注：若是任務接口的請求值須要其餘接口返回值中的參數，這些非任務請求也會在locust的統計面板中顯示出來。若想只關注任務接口的統計數據，則依賴的請求需用原生requests庫。

6. 數據監測工具推薦

（1）若是公司有搭建監測系統，可請運維協助在平臺查看便可，好比Grafana；
（2）linux檢測工具Nmon；
（3）windows自帶perfmon；
（4）使用python的psuil庫自定義檢測頻率與指標參數，須要對數據單獨進行處理；

7. 總結

    相比較jmeter等工具基於python語言locust的自由度高了不少，能夠自定義特殊協議的實現方法，且locust基於協程更容易構造性能自動化測試平臺，若是壓測機的配置有限，又想知足足夠高的併發量。用locust來實現是一個明智的選擇。

---

PS：更多技術乾貨，快關注【公衆號 | xingzhe_ai】，與行者一塊兒討論吧！