今天將繼續為大家帶來用Python進行gRPC接口測試的續集，上次主要講了一下前期準備工作和簡單RPC通信方式的實現，這次我們將著眼于另一類gRPC接口的通信形式——流式RPC。

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

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

首先接口協議是有區別的，我們來看三種形式的接口定義：

應答流式RPC：

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

請求流式RPC：

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

雙向流式RPC：

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

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

一個包含接口的完整proto協議文件（route_guide.proto）內容如下：

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;

}

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

二、客戶端實現

1、應答流式RPC

應答流式RPC返回的內容為流式，一次請求，n次返回。我們可以用for循環來接收返回的內容：

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次請求，一次返回。我們可以用迭代器來發送若干份請求數據：

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次請求，n次返回。我們可以用迭代器來發送若干份請求數據，通過for循環來接收返回結果：

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

三、實際應用

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

接口協議如下（僅為演示需要，只展示部分內容）：

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;

}

實現客戶端的關鍵代碼如下：

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

運行客戶端，可以成功返回結果：

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

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

運行腳本，即可自動化進行測試并得到結果，從中可以得到性能、穩定性相關指標：

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

小結

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

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

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。