Locust完成gRPC协议的性能测试

Posted 2021-06-05 独行高飞

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