本文首發於：行者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】，與行者一起討論吧！