1. 程式人生 > >Python gRPC 環境搭建及測試

Python gRPC 環境搭建及測試

小白一枚,啥也不懂,直接上手,雖然不知道我在幹什麼,但是我執行成功了,手動哈哈哈。。。

安裝:

  1. gRPC 的安裝:

$ pip install grpcio

  1. 安裝 ProtoBuf 相關的 python 依賴庫:

$ pip install protobuf

  1. 安裝 python grpc 的 protobuf 編譯工具:

$ pip install grpcio-tools

檔案生成

首先在IDE中新建專案,然後建立相對應的資料夾,橢圓圈圈是手動建立的檔案,矩形框是自動生成的,下面詳細說明。

1、在example中手動建立data.proto

輸入下面程式碼段:

syntax = "proto3";
package example;
service FormatData {
  rpc DoFormat(Data) returns (Data){}
}
message Data {
  string text = 1;
}

然後開啟命令窗(cmd或者是anaconda自帶的prompt):

cd到data.proto所在的位置:

執行:

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. ./data.proto 

就會自動生成上面矩形框中的兩個檔案。

測試:

client(main函式)

import grpc
import sys
import sys
sys.path.append('..')

from example import data_pb2
from example import data_pb2_grpc
# sys.path.append(r'C:\Users\huaru\PycharmProjects\GRPC_demo')

_HOST = 'localhost'
_PORT = '8080'

def run():
    conn = grpc.insecure_channel(_HOST + ':' + _PORT)
    client = data_pb2_grpc.FormatDataStub(channel=conn)
    response = client.DoFormat(data_pb2.Data(text='hello,world!'))
    print("received: " + response.text)

if __name__ == '__main__':
    run()

serve(main函式)

import grpc
import time
from concurrent import futures
import sys
sys.path.append('..')
#sys.path

from example import data_pb2#, data_pb2_grpc
from example import data_pb2_grpc

_ONE_DAY_IN_SECONDS = 60 * 60 * 24
_HOST = 'localhost'
_PORT = '8080'

class FormatData(data_pb2_grpc.FormatDataServicer):
    def DoFormat(self, request, context):
        str = request.text
        return data_pb2.Data(text=str.upper())

def serve():
    grpcServer = grpc.server(futures.ThreadPoolExecutor(max_workers=4))
    data_pb2_grpc.add_FormatDataServicer_to_server(FormatData(), grpcServer)
    grpcServer.add_insecure_port(_HOST + ':' + _PORT)
    grpcServer.start()
    try:
        while True:
            time.sleep(_ONE_DAY_IN_SECONDS)
    except KeyboardInterrupt:
        grpcServer.stop(0)

if __name__ == '__main__':
    serve()

先執行serve,再執行client,結果: