Locust作為開源的壓力測試工具，這幾年備受關(guān)注，用法已經(jīng)被大神們開發(fā)的差不多了太惠。最近版本大更新，我重新學(xué)習(xí)一些基礎(chǔ)用法疲憋，卻發(fā)現(xiàn)參數(shù)化這一塊的代碼不能滿足我的需求凿渊。所以這次我從一個小白的角度出發(fā)，爭取用簡單的方式缚柳，實現(xiàn)Locust獲取外部csv數(shù)據(jù)并參數(shù)化埃脏。
在直接編寫代碼之前，要了解兩個庫：pandas與queue秋忙。（想直接看代碼麻煩手動跳到最后剂癌。）
pandas——Python的數(shù)據(jù)分析支持庫
Python Data Analysis Library 或 pandas 是基于NumPy 的一種工具，該工具是為了解決數(shù)據(jù)分析任務(wù)而創(chuàng)建的翰绊。Pandas 納入了大量庫和一些標(biāo)準(zhǔn)的數(shù)據(jù)模型佩谷，提供了高效地操作大型數(shù)據(jù)集所需的工具。
簡言之监嗜，pandas高效谐檀，方便，好用裁奇。
它有兩種獨有的數(shù)據(jù)結(jié)構(gòu)：Series 和 DataFrame桐猬。
- Series：與Python基本的數(shù)據(jù)結(jié)構(gòu)List相近，Series中只允許存儲相同的數(shù)據(jù)類型
- DataFrame：二維的表格型數(shù)據(jù)結(jié)構(gòu)刽肠。很多功能與R中的data.frame類似溃肪。可以將DataFrame理解為Series的容器音五。
在獲取csv數(shù)據(jù)時惫撰，我使用了DataFrame數(shù)據(jù)結(jié)構(gòu)，這種表格形式更方便理解與操作躺涝，下面是基本用法：
from pandas import Series,DataFrame
# 使用dict定義DataFrame
data = {"name":['google','baidu','yahoo'],"marks":[100,200,300],"price":[1,2,3]}
# 讀取csv文件
reader = pandas.read_csv(file_path)
# 將數(shù)據(jù)轉(zhuǎn)換為DataFrame格式
df = pd.DataFrame(reader)
# 遍歷行數(shù)據(jù)
for index, row in dFrame.iterrows():
print(index, row[i]) i為列序號厨钻，從0開始
queue——一種同步的隊列類
隊列queue 多應(yīng)用在多線程應(yīng)用中，多線程訪問共享變量坚嗜。對于多線程而言夯膀，訪問共享變量時，隊列queue是線程安全的苍蔬。從queue隊列的具體實現(xiàn)中诱建，可以看出queue使用了1個線程互斥鎖(pthread.Lock())，以及3個條件標(biāo)量(pthread.condition()),來保證了線程安全碟绑“吃常【線程安全：一個函數(shù)茎匠，當(dāng)且僅當(dāng)被多個并發(fā)線程反復(fù)調(diào)用時，能夠一直產(chǎn)生正確的結(jié)果辜荠，才能夠被稱為線程安全的（thread-safe）】
我主要使用queue.Queue來實例化參數(shù)隊列，將數(shù)據(jù)put到對列中抓狭，再從隊列中g(shù)et出來伯病，傳給相應(yīng)請求。下面為基本用法：
import queue
# 實例化一個隊列
queue_data = queue.Queue()
# put數(shù)據(jù)
queue_data.put_nowait(user)
# get數(shù)據(jù)
queue_data.get_nowait(user)
put_nowait()與get_nowait()實際為block=False的put()與get()否过，消息隊列如果沒有空間可寫入/獲取午笛，則會立刻拋出“queue.Full/queue.Empty”異常
pandas與queue結(jié)合，實現(xiàn)參數(shù)化
import queue
import pandas as pd
import sys
from locust import task, HttpUser, between, TaskSet
# 預(yù)防棧溢出苗桂，設(shè)置嵌套層數(shù)上限為10000
sys.setrecursionlimit(10000)
# 測試數(shù)據(jù)csv文件路徑
file_path = "test_data.csv"
# 讀取csv文件
reader = pd.read_csv(file_path)
# 將數(shù)據(jù)轉(zhuǎn)換為DataFrame格式
df = pd.DataFrame(reader)
# 定義csv文件的單列數(shù)據(jù)獲取函數(shù)
def queue_data(dFrame, i, **kwargs):
data = queue.Queue() # 先進(jìn)先出
# queue_data = queue.LifoQueue() 后進(jìn)先出
for index, row in dFrame.iterrows():
# print(index, row[i]) i為列序號药磺，從0開始
try:
data.put_nowait(row[i])
except queue.Full:
print("隊列溢出")
return data
class MyTaskSet(TaskSet):
# 在task中使用queue數(shù)據(jù)
@task
def task1(self):
# 使用queue中的數(shù)據(jù)進(jìn)行參數(shù)化
url = "/postData"
try:
row0_data = self.user.row0.get_nowait()
row1_data = self.user.row1.get_nowait()
payload = {"row0": row0_data, "row1": row1_data}
print(payload)
res = self.client.post(url, json=payload, name="使用queue中的數(shù)據(jù)進(jìn)行參數(shù)化")
except queue.Empty:
# 隊列取空后退出
print("queue is empty")
exit()
class MyUser(HttpUser):
host = "http://example.com"
tasks = [MyTaskSet]
wait_time = between(1, 3)
# 獲取單列數(shù)據(jù)放入對應(yīng)隊列中
row0 = queue_data(df, 0)
row1 = queue_data(df, 1)