参考文档

https://developers.google.com/protocol-buffers/docs/reference/python-generated

安装

pip install grpcio grpcio-tools protobuf

定义Protocol Buffer文件

syntax="proto3";

package test;

service Bi {
    rpc HelloDewei(HelloDeweiReq) returns (HelloDeweiReply) {}
}


message HelloDeweiReq {
    string name = 1;
    int32 age = 2;
}

message HelloDeweiReply {
    string result = 1;
}

生成

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

编写服务端

import grpc
import time
import hello_pb2 as pb2
import hello_pb2_grpc as pb2_grpc

from concurrent import futures


class Bi(pb2_grpc.BiServicer):
    def HelloDewei(self, request, context):
        name = request.name
        age = request.age

        result = f'my name is {name}, i am {age} years old'
        return  pb2.HelloDeweiReply(result=result)


def run():
    mygrpc_server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=4)
    )

    pb2_grpc.add_BiServicer_to_server(Bi(), mygrpc_server)
    mygrpc_server.add_insecure_port('127.0.0.1:9999')
    print('server will start as 127.0.0.1:9999')
    mygrpc_server.start()

    try:
        while True:
            time.sleep(3600)
    except KeyboardInterrupt:
        mygrpc_server.stop(0)


if __name__ == '__main__':
    run()

编写客户端

import grpc
import hello_pb2 as pb2
import hello_pb2_grpc as pb2_grpc


def run():
    conn = grpc.insecure_channel("127.0.0.1:9999")
    client = pb2_grpc.BiStub(channel=conn)
    response = client.HelloDewei(pb2.HelloDeweiReq(
        name="dewei",
        age=33
    ))

    print(response.result)

if __name__ == '__main__':
    run()