GRPC 安装
安装 grpcio、grpcio-tools、protobuf、
pip install grpcio -i https://pypi.tuna.tsinghua.edu.cn/simple
pip install grpcio-tools -i https://pypi.tuna.tsinghua.edu.cn/simple
pip install protobuf -i https://pypi.tuna.tsinghua.edu.cn/simple
常用类型
proto 文件
// 选择 proto3
syntax = "proto3";// 名字
package test;// 服务{函数}
service Bilili {rpc Hello(HelloReq) returns (HelloReply) {} // 双向非流// rpc Hello(stream HelloReq) returns (stream HelloReply) {} // stream 保持链接
}// 双向非流 参数定义
// 输入函数
message HelloReq {string name = 1;int32 age = 2;repeated string data = 3;map<string, demo1> number = 4;
}message demo1 {string n1 = 1;int64 n2 = 2;bool n3 = 3;
}// 返回函数
message HelloReply {string result = 1;
}proto 文件 转换为 python 命令
我的文件名：test.proto
python -m grpc_tools.protoc -I. --python_out=. --pyi_out=. --grpc_python_out=. test.proto
客户端
# client.py
import grpc
import test_pb2_grpc as pb2_grpc
import test_pb2 as pb2def run():# 绑定地址conn = grpc.insecure_channel('127.0.0.1:5001')# 绑定对应服务client = pb2_grpc.BililiStub(channel=conn)# 绑定服务对应的函数response = client.Hello(pb2.HelloReq(name='SB',age=33,data=['1', '2', '3'],number={'sb': pb2.demo1(n1='q', n2=33, n3=True)}), )print(response.result)if __name__ == '__main__':run()服务端
# service.py
import time
import grpc
import test_pb2 as pb2
import test_pb2_grpc as pb2_grpc
from concurrent import futuresclass Bili(pb2_grpc.BililiServicer):def Hello(self, request, context):name = request.nameage = request.agedata = request.datamap_1 = request.numberresult = f'姓名： {name}, 年龄： {age} 数组：{data} 字典：{map_1}'return pb2.HelloReply(result=result)def run():# 服务grpc_server = grpc.server(# 设置了4个进程futures.ThreadPoolExecutor(max_workers=4),)# 注册服务pb2_grpc.add_BililiServicer_to_server(Bili(), grpc_server)# 绑定 地址grpc_server.add_insecure_port('0.0.0.0:5001')print('server start..')grpc_server.start()try:while True:time.sleep(3600)# 按Ctrl + cexcept KeyboardInterrupt:# 安全退出grpc_server.stop(0)if __name__ == '__main__':run()
