Locust完成gRPC协议的性能测试
1、gRPC知多少
对于分布式系统而言,不同的服务分布在不同的节点上,一个服务要完成自己的功能经常需要调用其他服务的接口,比如典型的微服务架构。通常这种服务调用方式有两种,一种是发送HTTP请求的方式,另一种则是RPC的方式,RPC是Remote Procedure Call(远程过程调用)的简称,可以让我们像调用本地接口一样使用远程服务。gRPC是一个由 google 推出的、高性能、开源、通用的 rpc 框架。它是基于 HTTP2 协议标准设计开发，默认采用 Protocol Buffers 数据序列化协议，支持多种开发语言。ProtoBuf buffer 是一种数据表达方式，以.proto 结尾的数据文件，可以类比 json、xml 等。
2、脚本设计
在我们开始性能测试之前，我们必须先有一个测试服务，那么我给你一个最简单的gRPC服务的例子。SUT服务代码下载
再打开这个例子后，第一步要通过如下的命令安装依赖
pip install -r requirements.txt
那么我们运行一下，通过如下命令启动服务端：python greeter_server.py
然后我们再次运行：python greeter_client.py
我们可以在服务端的控制台看到如下打印内容：
/usr/local/bin/python3.8 /greeter_server.py
Hello, you!
在客户端代码的控制台看见如下返回：
/usr/local/bin/python3.8 /greeter_client.py
Greeter client received: Hello, you!
Process finished with exit code 0
如果上面一切ok那么说明我们已经为利用Locust完成gRPC性能测试准备好了一个测试服务了。
3、撰写Locust脚本
python -m grpc_tools.protoc -I=server/proto --python_out=server/proto --grpc_python_out=server/proto server/proto/helloworld.proto
会生成两个python文件，helloworld_pb2.py helloworld_pb2_grpc.py，我们将其复制到我们的测试脚本代码所在目录下，然后创建性能测试脚本如下：
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2021/6/2 4:08 下午
# @Author : CrissChan
# @Site : https://blog.csdn.net/crisschan
# @File : load_test_grpc.py
# @Software: 这是调用gRPC的Locust脚本
import sys
import grpc
import inspect
import time
import gevent
from locust.contrib.fasthttp import FastHttpUser
from locust import task, events, constant
from locust.runners import STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP, WorkerRunner
import helloworld_pb2
import helloworld_pb2_grpc
def stopwatch(func):
def wrapper(*args, **kwargs):
previous_frame = inspect.currentframe().f_back
_, _, task_name, _, _ = inspect.getframeinfo(previous_frame)
start = time.time()
result = None
try:
result = func(*args, **kwargs)
except Exception as e:
total = int((time.time() - start) * 1000)
events.request_failure.fire(request_type="TYPE",
name=task_name,
response_time=total,
response_length=0,
exception=e)
else:
total = int((time.time() - start) * 1000)
events.request_success.fire(request_type="TYPE",
name=task_name,
response_time=total,
response_length=0)
return result
return wrapper
class GRPCMyLocust(FastHttpUser):
host = 'http://127.0.0.1:50051'# 服务端地址和端口号
wait_time = constant(0)
def on_start(self):
pass
def on_stop(self):
pass
@task
@stopwatch
def grpc_client_task(self):
try:# 服务端地址和端口号
with grpc.insecure_channel('127.0.0.1:50051') as channel:
stub = helloworld_pb2_grpc.GreeterStub(channel)
response = stub.SayHello(helloworld_pb2.HelloRequest(name='criss'))
print(response)
except (KeyboardInterrupt, SystemExit):
sys.exit(0)
4、控制台启动 在脚本所在的目录下执行：
locust -f load_test_grpc.py