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