博客

官方python教程

https://www.grpc.io/docs/languages/python/quickstart/

前期准备

1.了解protobuf语法

编写.proto文件，使用对应工具生成目标语言的协议代码

官方教程

2.下载依赖

grpcio为需要使用的依赖，grpcio-tools用来转化.proto文件为.py文件

pip install grpcio grpcio-tools

使用教程

1. 编写data.proto

// 使用proto3语法
syntax = "proto3";

// 服务名称
service TestService {
  // rpc接口名称Infer，请求数据为ReqData，返回数据为RespData，ReqData和RespData结构体下面给出
  rpc Infer(ReqData) returns (RespData) {}
}

// 客户端传输数据格式
message ReqData
{
    // int32 类型变量id,为第一个参数
    int32 id = 1;
    // string 类型变量name,为第二个参数
    string name = 2;
    // 数组使用repeated
    // 如: repeated int32 id = 1;
}

// 服务端返回数据格式
message RespData
{
    string json = 1;
}

在data.proto目录下执行以下代码生成协议文件

python -m grpc_tools.protoc --python_out=. --grpc_python_out=. -I. data.proto

执行结束后，会在同级目录生成以下两个文件

• data_pb2.py
• data_pb2_grpc.py

2. 编写服务端

文件夹同级新建文件server.py，内容如下:

# 当前python版本3.7.11
from concurrent import futures
import grpc
# 引入两个生成的文件
import data_pb2
import data_pb2_grpc
import json

# test的服务，TestServiceServicer类型在data_pb2_grpc.py自动生成
class TestServicer(data_pb2_grpc.TestServiceServicer):
	# test的infer方法
    def Infer(self, request, context):
    	# 打印接收到的参数， request为ReqData结构数据
        print("id: {}".format(request.id))
        print("name: {}".format(request.name))
        # tuple保存数据
        json_data = {}
        json_data["id"] = request.id
        json_data["name"] = request.name
        # tuple转字符串返回RespData结构数据给客户端
        return data_pb2.RespData(json=json.dumps(json_data))
# 启动服务端
def start():
  # 开启总的服务
  # max_workers控制线程数量
  server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))
  # 将子服务添加到总的服务
  # 函数在data_pb2_grpc.py自动生成
  data_pb2_grpc.add_TestServiceServicer_to_server(TestServicer(), server)
  # 设置端口
  server.add_insecure_port("[::]:50051")
  server.start()
  print("grpc server start...")
  # 保持常驻状态
  server.wait_for_termination()

if __name__ == '__main__':
  start()

3. 编写客户端

3.1 一般客户端

文件夹同级新建文件client.py，内容如下:

import grpc
# 引入两个生成的文件
import data_pb2
import data_pb2_grpc

def run():
  # 连接到服务端
  channel = grpc.insecure_channel('localhost:50051')
  # 使用对应服务的stub
  # TestServiceStub函数在data_pb2_grpc.py自动生成
  stub = data_pb2_grpc.TestServiceStub(channel)
  # 远程调用TestServicer的函数Infer，并发送定义的ReqData结构数据
  response = stub.Infer(data_pb2.ReqData(id=1, name="test"))
  # response为RespData结构的数据，打印查看json的内容
  print(response.json)

if __name__ == "__main__":
  run()

3.2 flask作为客户端调用

如神经网络推理服务作为rpc服务端，flask的接口接收到请求然后调用rpc得到推理结果再返回给客户端

from flask import Flask, abort, request, jsonify, render_template, redirect, url_for
import grpc
import data_pb2
import data_pb2_grpc

HOST = "0.0.0.0"
PORT = 60000
app = Flask(__name__)


@app.route('/test', methods=['GET', 'POST'])
def test():
    channel = grpc.insecure_channel('localhost:50051')
    stub = data_pb2_grpc.TestServiceStub(channel)
    response = stub.Infer(data_pb2.ReqData(id=1, name="test"))
    data ={}
    # 补充相关业务
    return jsonify(data)

if __name__ == "__main__":
    app.run(host=HOST, port=PORT, debug=True)

启动后访问http://127.0.0.1:60000/test

4.一般客户端测试

先启动服务端，再启动客户端，服务端返回将客户端信息打包的json

字符串


以下为客户端发出请求后，返回的结果

{"id": 1, "name": "test"}