gRPC （四）python gRPC样例

本文代码地址

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()