本文代码地址
https://gitee.com/shoothzj/grpc-examples
python实现gRPC客户端服务端
添加依赖
pip install grpcio
pip install grpcio-tools
生成python代码
python -m grpc_tools.protoc -I .. --python_out=. --grpc_python_out=. ../message_proto2.proto
server侧代码
import logging
from concurrent import futures
import grpc
import message_proto2_pb2
import message_proto2_pb2_grpc
class EchoImpl(message_proto2_pb2_grpc.EchoProto2ServiceServicer):
def EchoProto2(self, request, context):
return message_proto2_pb2.EchoProto2Resp(
str_req=request.str_req,
str_opt=request.str_opt,
str_rep=request.str_rep,
int64_req=request.int64_req,
int32_opt=request.int32_opt,
comic=request.comic,
)
def serve():
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
message_proto2_pb2_grpc.add_EchoProto2ServiceServicer_to_server(EchoImpl(), server)
server.add_insecure_port('[::]:10240')
server.start()
server.wait_for_termination()
if __name__ == '__main__':
logging.basicConfig()
serve()
client侧代码
import grpc
import message_proto2_pb2
import message_proto2_pb2_grpc
def run():
with grpc.insecure_channel('localhost:10240') as channel:
stub = message_proto2_pb2_grpc.EchoProto2ServiceStub(channel)
response = stub.EchoProto2(message_proto2_pb2.EchoProto2Req(
str_req="strReq",
str_opt="strOpt",
str_rep=["str", "rep"],
int64_req=1,
int32_opt=2,
comic="Bleach",
))
print("client received: " + str(response))
if __name__ == '__main__':
run()
网友评论