「Protocol Buffer」のPBのgRPCへの応用
関連ライブラリのインストール $ python -m pip ×××tall grpcio
$ python -m pip ×××tall grpcio-tools googleapis-common-protos
Demoプログラム機能の概要
サーバ側にTest_が存在するサービスクラスでmy_が定義されていますfunctionメソッドは、クライアントがgRPCプロトコルによってリモートコールを行う.この方法は,受け取った文字列の内容をすべて大文字に変更して返す機能を実現する.
PBインタフェース記述ファイル定義 syntax = "proto3";
package data;
#
message Data{
string text=1;
}
# ,
service Test_service{
# 、
rpc my_function(Data) returns (Data) {}
}
コンパイル python3 -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. ./data.proto
クライアントコード #! /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()
サーバコード #! /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()
クライアント実行結果 $ python3 client.py
received: HELLO,WORLD!
リモートメソッド呼び出しに成功しました!
$ python -m pip ×××tall grpcio
$ python -m pip ×××tall grpcio-tools googleapis-common-protos
サーバ側にTest_が存在するサービスクラスでmy_が定義されていますfunctionメソッドは、クライアントがgRPCプロトコルによってリモートコールを行う.この方法は,受け取った文字列の内容をすべて大文字に変更して返す機能を実現する.
PBインタフェース記述ファイル定義 syntax = "proto3";
package data;
#
message Data{
string text=1;
}
# ,
service Test_service{
# 、
rpc my_function(Data) returns (Data) {}
}
コンパイル python3 -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. ./data.proto
クライアントコード #! /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()
サーバコード #! /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()
クライアント実行結果 $ python3 client.py
received: HELLO,WORLD!
リモートメソッド呼び出しに成功しました!
syntax = "proto3";
package data;
#
message Data{
string text=1;
}
# ,
service Test_service{
# 、
rpc my_function(Data) returns (Data) {}
}
python3 -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. ./data.proto
クライアントコード #! /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()
サーバコード #! /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()
クライアント実行結果 $ python3 client.py
received: HELLO,WORLD!
リモートメソッド呼び出しに成功しました!
#! /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()
#! /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()
クライアント実行結果 $ python3 client.py
received: HELLO,WORLD!
リモートメソッド呼び出しに成功しました!
$ python3 client.py
received: HELLO,WORLD!