相信很多同学在测试RPC接口时会遇到很多困难,博主前段时间在测试时也一样,算是提前踩坑啦,下面就来介绍一下测试RPC接口的方法
RPC(Remote Procedure Call)是一种通信协议和模式,用于在分布式系统中实现远程过程调用。
比较常见的RPC框架有gRPC、Apache Thrift、CORBA、XML-RPC、JSON-RPC等。
博主这里给两种方案
如果只是暂时测一下而懒得写代码的话,可以考虑使用jmeter下载一个grpc的插件来测试RPC接口,方法如下
1.如果之前没有jmeter可以直接下载博主这个链接,自带插件
https://download.csdn.net/download/m0_57098592/86783488
2.如果有jmeter也行,将博主文件里的这几个 jar 包 放到jmeter解压后得 ext 目录下,重启jmeter即可apache-jmeter-5.4.1\lib\ext
jmeter-plugins-manager-1.7.jar
jmeter-plugins-graphs-basic-2.0.jar
jmeter-plugins-casutg-2.10.jar
3.重启后可以在右上角看见一个小标志,点击进去,搜索grpc下载
下载完成后再次重启即可
4.在线程组-取样器中添加GRPC即可
5.按图中方式进行
6.确认没问题后,直接监听器里面添加个查看结果树看结果就可以啦,当前如果你要压性能直接在jmeter里面也更方便一些
优点:扩展性较好 缺点:需要一定代码能力
博主在这里踩了不少坑,下面说一下用代码实现的逻辑
1.首先,proto文件仍然是必须的,我们需要开发提供给我们,然后放在我们的某个目录中,同时装包
grpc
grpcio==1.48.2
grpcio-tools==1.48.2
2.cd 进入到该目录
3.控制台运行
python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. kce.proto\n
4.运行完后,会在该目录下根据你的proto自动生成两个文件,我们假设proto文件名为A
A_pb2.py
A_pb2_grpc.py
5.通过以下代码调用RPC接口
import grpc
import A_pb2
import A_pb2_grpc
def run(request_data):
#创建gRPC通道。
with grpc.insecure_channel('ip:port') as channel: #你的ip地址和端口号
# 创建客户端 Stub。
stub = A_pb2_grpc.B(channel) #B为A_pb2_grpc文件中的类名
# 构造请求对象。
request = A_pb2.C(**request_data) #C为proto文件message名(看你的proto文件)
try:
#调用远程服务。
response = stub.D(request, timeout=60) #D为RPC对应的方法
#处理响应
if response:
print("成功")
return response
else:
print("No response received.")
except grpc.RpcError as e:
print(f"RPC error occurred: {e}")
except Exception as e:
print(f"An error occurred: {e}")
6.创建一个测试方法来调用RPC接口
import json
from google.protobuf.json_format import MessageToDict
from 文件 import run #上面封装好的run方法
def test_A(requests_file):
# 读取 JSON 文件并转换成 Python 字典
request_json = requests_file #把你的传参写入文件,或者直接给json也行
#这里需要加载文件逻辑
response = load_json(jsonfile=request_json)
response_json = run(response)
# 将CategoryDTO对象转换为字典
response_dict = MessageToDict(response_json)
# 将返回的结果保存下来
response_path = os.path.join(BASE_DIR, 'XXX/XX')
response_file = os.path.join(response_path, 'XXX.json')
with open(response_file, 'w') as file:
file.write(json.dumps(response_dict))
if __name__ == '__main__':
test_A('XXX.json')
7.通过以上方法就能调用RPC接口,如果后续开发有变更proto文件的话,控制台再执行一下就好了
python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. kce.proto\n
代码上面还是有些需要结合自己实际情况修改的地方,大家多看一下哈