今天將繼續(xù)為大家?guī)碛肞ython進行g(shù)RPC接口測試的續(xù)集，上次主要講了一下前期準備工作和簡單RPC通信方式的實現(xiàn)，這次我們將著眼于另一類gRPC接口的通信形式——流式RPC。

一、流式RPC的三種具體形式

流式RPC不同于簡單RPC只有“單發(fā)單收“一種形式，而是可以分為三種不同的形式——“應(yīng)答流式RPC”，“請求流式RPC”，“雙向流式RPC”。對于這三種不同的形式，python有不同的請求及接收方式，下面就讓我們來具體了解一下。（對于下面操作有疑問的同學(xué)可以去看上一期的內(nèi)容）

首先接口協(xié)議是有區(qū)別的，我們來看三種形式的接口定義：

應(yīng)答流式RPC：

rpc ListFeatures(Rectangle) returns (stream Feature) {}

請求流式RPC：

rpc RecordRoute(stream Point) returns (RouteSummary) {}

雙向流式RPC：

rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}

可以看到，請求和響應(yīng)參數(shù)中流式內(nèi)容的前面會有一個stream標識，代表這是一個流式的內(nèi)容。應(yīng)答流式RPC只有返回是流式的，請求流式RPC只有請求是流式的，而雙向流式RPC請求和返回都是流式的。

一個包含接口的完整proto協(xié)議文件（route_guide.proto）內(nèi)容如下：

syntax = "proto3";

option java_multiple_files = true;

option java_package = "io.grpc.examples.routeguide";

option java_outer_classname = "RouteGuideProto";

option objc_class_prefix = "RTG";

package routeguide;

service RouteGuide {

rpc ListFeatures(Rectangle) returns (stream Feature) {}

rpc RecordRoute(stream Point) returns (RouteSummary) {}

rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}

}

message Point {

int32 latitude = 1;

int32 longitude = 2;

}

message Rectangle {

Point lo = 1;

Point hi = 2;

}

message Feature {

string name = 1;

Point location = 2;

}

message RouteNote {

Point location = 1;

string message = 2;

}

message RouteSummary {

int32 point_count = 1;

int32 feature_count = 2;

int32 distance = 3;

int32 elapsed_time = 4;

}

根據(jù)協(xié)議文件生成route_guide_pb2.py、route_guide_pb2_grpc.py兩個必要的模塊文件，然后就可以根據(jù)他們來創(chuàng)建客戶端了。

二、客戶端實現(xiàn)

1、應(yīng)答流式RPC

應(yīng)答流式RPC返回的內(nèi)容為流式，一次請求，n次返回。我們可以用for循環(huán)來接收返回的內(nèi)容：

def guide_list_features(stub):

rectangle = route_guide_pb2.Rectangle(

lo=route_guide_pb2.Point(latitude=400000000, longitude=-750000000),

hi=route_guide_pb2.Point(latitude=420000000, longitude=-730000000))

print("Looking for features between 40, -75 and 42, -73")

features = stub.ListFeatures(rectangle)

for feature in features:

print("Feature called %s at %s" % (feature.name, feature.location))

2、請求流式RPC

請求流式RPC請求的內(nèi)容為流式，n次請求，一次返回。我們可以用迭代器來發(fā)送若干份請求數(shù)據(jù)：

def generate_route(feature_list):

for _ in range(0, 10):

random_feature = feature_list[random.randint(0, len(feature_list) - 1)]

print("Visiting point %s" % random_feature.location)

yield random_feature.location

def guide_record_route(stub):

feature_list = route_guide_resources.read_route_guide_database()

route_iterator = generate_route(feature_list)

route_summary = stub.RecordRoute(route_iterator)

print("Finished trip with %s points " % route_summary.point_count)

print("Passed %s features " % route_summary.feature_count)

print("Travelled %s meters " % route_summary.distance)

print("It took %s seconds " % route_summary.elapsed_time)

其中route_iterator為一個迭代器。

3、雙向流式RPC

雙向流式RPC請求的內(nèi)容為流式，返回內(nèi)容也為流式，n次請求，n次返回。我們可以用迭代器來發(fā)送若干份請求數(shù)據(jù)，通過for循環(huán)來接收返回結(jié)果：

def generate_messages():

messages = [

make_route_note("First message", 0, 0),

make_route_note("Second message", 0, 1),

make_route_note("Third message", 1, 0),

make_route_note("Fourth message", 0, 0),

make_route_note("Fifth message", 1, 0),

]

for msg in messages:

print("Sending %s at %s" % (msg.message, msg.location))

yield msg

def guide_route_chat(stub):

responses = stub.RouteChat(generate_messages())

for response in responses:

print("Received message %s at %s" %

(response.message, response.location))

三、實際應(yīng)用

在錄音筆項目中，需要對轉(zhuǎn)寫后的文本進行分段語義整理，由于文本內(nèi)容可能較多，服務(wù)端需要采用流式的方式進行接收，并通過流式的方式將結(jié)果返給客戶端，于是這里采用了雙向流式RPC形式的接口。

接口協(xié)議如下（僅為演示需要，只展示部分內(nèi)容）：

syntax = "proto3";

package sogou.parrot.inner.semantic.v1;

import "google/protobuf/duration.proto";

import "record.proto";

option go_package = "git.speech.sogou/semantic/v1;semantic";

service discourse_understand{

rpc UnderstandFullText(stream UnderstandFullTextRequest) returns(stream UnderstandFullTextResponse);

}

message UnderstandFullTextRequest{

repeated SubSentence sub_sentences = 1;

repeated sogou.parrot.record.v1.NonSpeechSoundInfo sound_infos = 2;

repeated sogou.parrot.record.v1.AIMark ai_marks = 3;

}

message UnderstandFullTextResponse{

UnderstandFullTextResult result = 2;

}

實現(xiàn)客戶端的關(guān)鍵代碼如下：

def gen_iterator(request):

for r in [request]:

yield r

def get_understand_full_textresponse(stub, ai_marks, sound_infos, sub_sentences):

request = UnderstandFullTextRequest()

request.sub_sentences.extend(sub_sentences)

request.sound_infos.extend(sound_infos)

request.ai_marks.extend(ai_marks)

request_iter = gen_iterator(request)

try:

resps = stub.UnderstandFullText(request_iter)

for resp in resps:

resp_str = json.dumps(json.loads(MessageToJson(resp)),indent=4, ensure_ascii=False)

print(resp_str)

except Exception as e:

print (e)

def run():

ai_marks, sound_infos, sub_sentences = extract_data()

with grpc.insecure_channel(sys.argv[2]) as channel:

stub = discourse_understandStub(channel)

print("-------------- UnderstandFullText --------------")

get_understand_full_textresponse(stub, ai_marks, sound_infos, sub_sentences)

if __name__ == '__main__':

run()

運行客戶端，可以成功返回結(jié)果：

進一步，如果需要對接口進行并發(fā)下的穩(wěn)定性測試，依然可以將客戶端編譯成可執(zhí)行程序或利用shell腳本，再結(jié)合jmeter等自動化測試工具進行測試，以編譯可執(zhí)行程序的方法為例：

首先利用pyinstaller工具將腳本編譯為可執(zhí)行程序，接下來用jmeter編寫自動化測試腳本，在線程組下添加OS Process Sampler，傳入所需參數(shù)（下面的三個參數(shù)值為：文本，地址，句子起始編號）：

運行腳本，即可自動化進行測試并得到結(jié)果，從中可以得到性能、穩(wěn)定性相關(guān)指標：

此外還可以結(jié)合jmeter的參數(shù)化功能和隨機功能設(shè)置一些參數(shù)值，比如文本文件和句子起始id，從而更加全面地對接口進行測試：

小結(jié)

本文介紹了用python實現(xiàn)其他三種形式gRPC接口測試的方法，這樣四種形式的gRPC接口我們就都可以比較方便地進行測試了，對于今后需要測試gRPC接口的同學(xué)可以提供一些借鑒，當(dāng)然有更好地方法歡迎大家一起討論交流。~

本文內(nèi)容不用于商業(yè)目的，如涉及知識產(chǎn)權(quán)問題，請權(quán)利人聯(lián)系博為峰小編(021-64471599-8017)，我們將立即處理

總結(jié)

以上是生活随笔為你收集整理的python grpc 并发_用Python进行gRPC接口测试（二）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。