Python + Jmeter 实现自动化性能压测

Step01: Python脚本开发

文件路径：D://wl//testproject//Fone-grpc//project1//test_client.py
Python 脚本作用：
1.通过 grpc 调用底层 c++ 的接口，做数据库的数据插入与查询操作，然后将返回的结果进行拼接与输出。
2.代码内里将每一次调用后返回的内容进行拼接后，并做了成功信息的统计，输出成功的次数为3，输出后会转给 jmeter 内里的脚本获取。

1. from __future__ import print_function
2. import logging
3. import sys
4. import grpc
5. import Storage_pb2      # Storage_pb2.py which contains our generated request and response classes
6. import StorageService_pb2_grpc # StorageService_pb2_grpc.py which contains our generated client and server classes.
8. host="10.10.1.117:50066"
10. def run():
11.     result=""
12.     with grpc.insecure_channel(host) as channel:
13.         resp0=StorageService_pb2_grpc.FOneStorageStub(channel).InsertKVS(Storage_pb2.PUpsertKVS(
14.             DBName="MergeLog",TableName="MergeLog",KeyValues=[{"Key":b'',"Key":b''}],Upsert=True,Transaction=True
15.         ))
16.         result+=str(resp0)
18.     with grpc.insecure_channel(host) as channel:
19.         stud1=StorageService_pb2_grpc.FOneStorageStub(channel)
20.         resp1=stud1.InsertV(Storage_pb2.PInsertV(DBName="MergeLog", TableName="MergeLog", Value=b"1"))
21.         result += str(resp1)
22.      
23.     with grpc.insecure_channel(host) as channel:
24.         stud2=StorageService_pb2_grpc.FOneStorageStub(channel)
25.         resp2=stud2.FindOne(Storage_pb2.PFindK(DBName="MergeLog", TableName="MergeLog", Key="1"))
26.         result += str(resp2)
28.     result=result.replace("\n", " ")
29.     result=result.count("Successful return: 0")
30.     print(result)
33. if __name__ == '__main__':
34.     logging.basicConfig()
35.     run()

复制代码

PS：在当地调试好脚本，确保符合预期。
Step02: Jmeter 创建 beanshell 取样器

脚本内容如下：

1. import java.io.BufferedReader;
2. import java.io.InputStreamReader;
4. log.info("----------------------------start to exe");
5. String command = "cmd /c python D://wl//testproject//Fone-grpc//project1//test_client.py"; //定义要执行的python文件路径
6. String var;
7. Runtime rt = Runtime.getRuntime();
8. Process pr = rt.exec(command);  //执行上面的python脚本
9. pr.waitFor()
10. BufferedReader b = new BufferedReader(new InputStreamReader(pr.getInputStream())); // 转换执行脚本后的返回响应
11. String line = "";
12. StringBuilder response = new StringBuilder();
13. while ((line = b.readLine()) != null) {
14.      response.append(line);
15. }
16. String response_data = response.toString();
17. log.info(response_data);
18. if(response_data.equals("3")){ // 判断响应的内容是否有3次，如果不是3次就表示执行失败了
19. log.info("---------------------------success end exe");
20. }else{
21. log.error("--------------------------failed end exe"); //这里的输出一次会记录一次失败次数
22. }
23. b.close();

复制代码

Step03: 实行效果

通过 jmeter 工具进行并发操作，对后端底层接口进行压力测试，效果不错，效果图如下所示。

Step04: Jmeter日志输出

1. 2023-02-22 17:26:08,539 INFO o.a.j.t.JMeterThread: Stopping because end time detected by thread: 线程组--python 1-33
2. 2023-02-22 17:26:08,539 INFO o.a.j.t.JMeterThread: Thread finished: 线程组--python 1-33
3. 2023-02-22 17:26:08,597 INFO o.a.j.u.BeanShellTestElement: 3
4. 2023-02-22 17:26:08,597 INFO o.a.j.u.BeanShellTestElement: ---------------------------success end exe
5. 2023-02-22 17:26:08,598 INFO o.a.j.t.JMeterThread: Stopping because end time detected by thread: 线程组--python 1-82
6. 2023-02-22 17:26:08,598 INFO o.a.j.t.JMeterThread: Thread finished: 线程组--python 1-82
7. 2023-02-22 17:26:08,608 INFO o.a.j.u.BeanShellTestElement: 2
8. 2023-02-22 17:26:08,608 ERROR o.a.j.u.BeanShellTestElement: --------------------------failed end exe

复制代码

假如本日的分享对你有帮助的话，请绝不犹豫：关注、分享、点赞、在看、收藏呀~
最后： 下方这份完备的软件测试视频学习教程已经整理上传完成，朋友们假如需要可以自行免费领取 【保证100%免费】

这些资料，对于【软件测试】的朋友来说应该是最全面最完备的备战仓库，这个仓库也陪伴上万个测试工程师们走过最艰难的路程，盼望也能帮助到你!

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。