参考文档
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()
