Step01: Python腳本開(kāi)發(fā)
文件路徑:D://wl//testproject//Fone-grpc//project1//test_client.py
Python 腳本作用:
-
通過(guò) grpc 調(diào)用底層 c++ 的接口,做數(shù)據(jù)庫(kù)的數(shù)據(jù)插入與查詢操作,然后將返回的結(jié)果進(jìn)行拼接與輸出。
2. 代碼里面將每一次調(diào)用后返回的內(nèi)容進(jìn)行拼接后,并做了成功信息的統(tǒng)計(jì),輸出成功的次數(shù)為3,輸出后會(huì)轉(zhuǎn)給 jmeter 里面的腳本獲取。
from?__future__?import?print_function
import?logging
import?sys
import?grpc
import?Storage_pb2??????#?Storage_pb2.py?which?contains?our?generated?request?and?response?classes
import?StorageService_pb2_grpc?#?StorageService_pb2_grpc.py?which?contains?our?generated?client?and?server?classes.
host="10.10.1.117:50066"
def?run():
????result=""
????with?grpc.insecure_channel(host)?as?channel:
????????resp0=StorageService_pb2_grpc.FOneStorageStub(channel).InsertKVS(Storage_pb2.PUpsertKVS(
????????????DBName="MergeLog",TableName="MergeLog",KeyValues=[{"Key":b'',"Key":b''}],Upsert=True,Transaction=True
????????))
????????result+=str(resp0)
????with?grpc.insecure_channel(host)?as?channel:
????????stud1=StorageService_pb2_grpc.FOneStorageStub(channel)
????????resp1=stud1.InsertV(Storage_pb2.PInsertV(DBName="MergeLog",?TableName="MergeLog",?Value=b"1"))
????????result?+=?str(resp1)
?????
????with?grpc.insecure_channel(host)?as?channel:
????????stud2=StorageService_pb2_grpc.FOneStorageStub(channel)
????????resp2=stud2.FindOne(Storage_pb2.PFindK(DBName="MergeLog",?TableName="MergeLog",?Key="1"))
????????result?+=?str(resp2)
????result=result.replace("\n",?"?")
????result=result.count("Successful?return:?0")
????print(result)
if?__name__?==?'__main__':
????logging.basicConfig()
????run()
PS:在本地調(diào)試好腳本,確保符合預(yù)期。
Step02: Jmeter 創(chuàng)建 beanshell 取樣器
腳本內(nèi)容如下:
import?java.io.BufferedReader;
import?java.io.InputStreamReader;
log.info("----------------------------start?to?exe");
String?command?=?"cmd?/c?python?D://wl//testproject//Fone-grpc//project1//test_client.py";?//定義要執(zhí)行的python文件路徑
String?var;
Runtime?rt?=?Runtime.getRuntime();
Process?pr?=?rt.exec(command);??//執(zhí)行上面的python腳本
pr.waitFor()
BufferedReader?b?=?new?BufferedReader(new?InputStreamReader(pr.getInputStream()));?//?轉(zhuǎn)換執(zhí)行腳本后的返回響應(yīng)
String?line?=?"";
StringBuilder?response?=?new?StringBuilder();
while?((line?=?b.readLine())?!=?null)?{
?????response.append(line);
}?
String?response_data?=?response.toString();
log.info(response_data);
if(response_data.equals("3")){?//?判斷響應(yīng)的內(nèi)容是否有3次,如果不是3次就表示執(zhí)行失敗了
?log.info("---------------------------success?end?exe");
}else{
?log.error("--------------------------failed?end?exe");?//這里的輸出一次會(huì)記錄一次失敗次數(shù)
}
b.close();
Step03: 執(zhí)行效果
通過(guò) jmeter 工具進(jìn)行并發(fā)操作,對(duì)后端底層接口進(jìn)行壓力測(cè)試,效果不錯(cuò),效果圖如下所示。
?Step04: Jmeter日志輸出
2023-02-22 17:26:08,539 INFO o.a.j.t.JMeterThread: Stopping because end time detected by thread: 線程組--python 1-33
2023-02-22 17:26:08,539 INFO o.a.j.t.JMeterThread: Thread finished: 線程組--python 1-33
2023-02-22 17:26:08,597 INFO o.a.j.u.BeanShellTestElement: 3
2023-02-22 17:26:08,597 INFO o.a.j.u.BeanShellTestElement: ---------------------------success end exe
2023-02-22 17:26:08,598 INFO o.a.j.t.JMeterThread: Stopping because end time detected by thread: 線程組--python 1-82
2023-02-22 17:26:08,598 INFO o.a.j.t.JMeterThread: Thread finished: 線程組--python 1-82
2023-02-22 17:26:08,608 INFO o.a.j.u.BeanShellTestElement: 2
2023-02-22 17:26:08,608 ERROR o.a.j.u.BeanShellTestElement: --------------------------failed end exe
END
?今天的分享就到此結(jié)束了 ,如果文章對(duì)你有幫助,記得點(diǎn)贊,收藏,加關(guān)注。會(huì)不定期分享一些干貨哦......
最后感謝每一個(gè)認(rèn)真閱讀我文章的人,看著粉絲一路的上漲和關(guān)注,禮尚往來(lái)總是要有的,雖然不是什么很值錢(qián)的東西,如果你用得到的話可以直接拿走:
這些對(duì)從事【軟件測(cè)試】的朋友來(lái)說(shuō)應(yīng)該是最全面最完整的備戰(zhàn)倉(cāng)庫(kù),這個(gè)倉(cāng)庫(kù)也陪伴我走過(guò)了最艱難的路程,希望也能幫助到你!凡事要趁早,特別是技術(shù)行業(yè),一定要提升技術(shù)功底。希望對(duì)大家有所幫助……文章來(lái)源:http://www.zghlxwxcb.cn/news/detail-785806.html
文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-785806.html
到了這里,關(guān)于一文3000字教你用Python + Jmeter 實(shí)現(xiàn)自動(dòng)化性能壓測(cè)的文章就介紹完了。如果您還想了解更多內(nèi)容,請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章,希望大家以后多多支持TOY模板網(wǎng)!