「Protocol Buffer」之PB在gRPC中的應(yīng)用

相關(guān)庫(kù)的安裝

$ python -m pip ×××tall grpcio
$ python -m pip ×××tall grpcio-tools googleapis-common-protos

Demo程序功能概述

服務(wù)器端存在Test_service類(lèi)中定義了my_function方法，客戶(hù)端通過(guò)gRPC協(xié)議進(jìn)行遠(yuǎn)程調(diào)用；該方法實(shí)現(xiàn)的功能是將接受到的字符串內(nèi)容全部改為大寫(xiě)并返回

PB接口描述文件定義

syntax = "proto3";
package data;

#定義數(shù)據(jù)結(jié)構(gòu)
message Data{
    string text=1;
}

#定義服務(wù)接口，即聲明我要調(diào)用目標(biāo)服務(wù)器上的哪個(gè)函數(shù)
service Test_service{
    #定義調(diào)用函數(shù)名稱(chēng)、參數(shù)類(lèi)型與返回?cái)?shù)據(jù)類(lèi)型
    rpc my_function(Data) returns (Data) {}
}

編譯

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

客戶(hù)端代碼

#! /usr/bin/env python
# -*- coding: utf-8 -*-
import grpc
import data_pb2, data_pb2_grpc

_HOST = '127.0.0.1'
_PORT = '8080'

def run():
    conn = grpc.×××ecure_channel(_HOST + ':' + _PORT)
    client = data_pb2_grpc.Test_serviceStub(channel=conn)
    response = client.my_function(data_pb2.Data(text='hello,world!'))
    print("received: " + response.text)

if __name__ == '__main__':
    run()

服務(wù)器代碼

#! /usr/bin/env python
# -*- coding: utf-8 -*-
import grpc
import time
from concurrent import futures
import data_pb2, data_pb2_grpc

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

class Test_service(data_pb2_grpc.Test_serviceServicer):
    def my_function(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_Test_serviceServicer_to_server(Test_service(), grpcServer)
    grpcServer.add_×××ecure_port(_HOST + ':' + _PORT)
    grpcServer.start()
    try:
        while True:
            time.sleep(_ONE_DAY_IN_SECONDS)
    except KeyboardInterrupt:
        grpcServer.stop(0)

if __name__ == '__main__':
    serve()

客戶(hù)端運(yùn)行結(jié)果

$ python3 client.py 
received: HELLO,WORLD!

遠(yuǎn)程方法調(diào)用成功！