上期回顾:用Python进行gRPC接口测试
一、流式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()
https://mp.weixin.qq.com/s/Y2sHs_Sq4lB3hBhKGSvaNg
语: 本文以幽默诙谐的方式,介绍gRPC的4种client-server服务模式的开发实践及应用场景
前言:为什么要写这篇文章?
gRPC client-server服务模式
这里先将重要的结论写出来,方便以后查阅,具体介绍见下文。找不到准确的中文来翻译这几种模式,就保留了英文。
应用场景:
-
agent上报CPU,内存等数据到server
-
客户端心跳
-
客户端并发调用细小粒度的接口。比如有5个后台接口A B C D E,客户端在不同页面,可以调用不同的接口组合。比如在个人页,就调用ABC;在动态页面,就调用CDE,后台都只会有一个rsp。这种模式的好处就是让后台可以将接口的粒度细化,客户端调用灵活,减少重复代码,提高复用率
应用场景:
-
股票app。客户端向服务端发送一个股票代码,服务端就把该股票的实时数据源源不断的返回给客户端
-
app的在线push。client先发请求到server注册,然后server就可以发在线push了
应用场景:
-
聊天机器人
-
有状态的游戏服务器进行数据交换。比如LOL,王者荣耀等竞技游戏,client和server之间需要非常频繁地交换数据
接下来就让我用一个男女生之间的交往故事来说明这4种服务模式。内容纯属虚构,并故意写得比较搞笑。如果有不恰当的描述,请告诉我。
男女生交往之gRPC的4种模式
一些说明
基础代码说明
conn, err := grpc.Dial(love_const.Address, grpc.WithInsecure())
(左滑可查看完整代码,下同)
lis, err := net.Listen("tcp", love_const.Address)
grpcServer := grpc.NewServer()
love_proto.RegisterBehaviorServer(grpcServer, newServer())
grpcServer.Serve(lis)
0x1: A simple RPC
Selina: "在哪"
Jhon: "刚起床,在打游戏呢"
这种模式就是我们用得最多的模式,一发一收
response, err := client.WhereAreYou(ctx, message)
func (s *loveServer) WhereAreYou(ctx context.Context, message *love_proto.Message) (*love_proto.Response, error) {
rsp := new(love_proto.Response)
rsp.Words = "刚起床,在打游戏呢"
return rsp, nil
}
0x2: A client-to-server streaming RPC
Selina: "你在干嘛"
Selina: "我不开心"
Selina: "我要和你说话"
Selina: "你怎么还不打电话过来"
Selina: "我来大姨妈了"
Jhon: "哦,多喝热水"
这种模式是client先建立长连接,然后发送多个request给server,server最后统一回一个rsp。
应用场景
-
agent收集CPU,内存等数据到server
-
客户端心跳
-
客户端并发调用细小粒度的接口。比如有5个后台接口A B C D E,客户端在不同页面,可以调用不同的接口组合,比如在个人页,就调用ABC;在动态页面,就调用CDE,后台都只会有一个rsp。这种模式的好处就是让后台可以将接口的粒度细化,客户端调用灵活,减少重复代码,提高复用率
stream, err := client.ContinuousCall(ctx)
if err != nil {
log.Fatalf("%v.ContinuousCall(_) = _, %v", client, err)
}
for _, message := range messages {
fmt.Printf("message words: %s
", message.Words)
if err := stream.Send(message); err != nil {
log.Fatalf("%v.Send(%v) = %v", stream, message, err)
}
}
reply, err := stream.CloseAndRecv()
if err != nil {
log.Fatalf("%v.CloseAndRecv() got error %v, want %v", stream, err, nil)
}
for {
message, err := stream.Recv()
if err == io.EOF {
rsp := new(love_proto.Response)
rsp.Words = "哦,多喝热水"
return stream.SendAndClose(rsp)
}
if err != nil {
return err
}
printGirlWords(message.Words)
}
0x3: A server-to-client streaming RPC
Selina: "这么久不给我打电话,你是不是不爱我了?"
Jhon: "啊,宝贝你怎么了?"
Jhon: "我刚刚在玩游戏,那一局刚开,我走不开啊"
Jhon: "你不能不讲道理啊"
Jhon: "你都20分钟不回我了"
Jhon: "好了,宝贝,我错了,都是我的错"
Jhon: "你在家等我,我过去接你,带你去买包包"
这种模式是client先发一个请求到server,然后server不停的回包
应用场景
-
股票app。客户端向服务端发送一个股票代码,服务端就把该股票的实时数据源源不断的返回给客户端
-
app的在线push。client先发请求到server注册,然后server就可以发在线push了
stream, err := client.LoveOrNot(ctx, message)
if err != nil {
log.Fatalf("%v.LoveOrNot(_) = _, %v", client, err)
}
for {
response, err := stream.Recv()
if err == io.EOF {
break
}
if err != nil {
log.Fatalf("%v.LoveOrNot(_) = _, %v", client, err)
}
log.Printf("rsp: %s", response.Words)
}
for _, response := range responses {
if err := stream.Send(response); err != nil {
return err
}
}
0x4: A Bidirectional streaming RPC
Selina: "好呀好呀"
Jhon: "宝贝我很快就到啦"
Selina: "等我化妆"
Jhon: "宝贝我很快就到啦"
Selina: "亲爱的你最好啦"
Jhon: "宝贝我很快就到啦"
Selina: "么么哒"
Jhon: "宝贝我很快就到啦"
这种模式就是建立一个长连接,然后client和server可以随意收发数据
应用场景
-
聊天机器人
-
有状态的游戏服务器进行数据交换。比如LOL,王者荣耀等竞技游戏,client和server之间需要非常频繁地交换数据
stream, err := client.LoveChat(ctx)
if err != nil {
log.Fatalf("%v.LoveChat(_) = _, %v", client, err)
}
waitc := make(chan struct{})
go func() {
for {
response, err := stream.Recv()
if err == io.EOF {
close(waitc)
log.Printf("EOF return")
return
}
if err != nil {
log.Fatalf("Failed to receive a note : %v", err)
}
log.Printf("rsp: %s", response.Words)
}
}()
for _, message := range messages {
time.Sleep(1 * time.Second)
if err := stream.Send(message); err != nil {
log.Fatalf("Failed to send a message: %v", err)
}
}
stream.CloseSend()
<-waitc
for {
message, err := stream.Recv()
if err == io.EOF {
return nil
}
if err != nil {
return err
}
for _, response := range responses {
if err := stream.Send(response); err != nil {
return err
}
}
}
男生们看过来,看过来
-
既然标题写了“用gRPC教程序员谈恋爱”,那么肯定就得给点干货。女生可能天生比男生更感性一些,所以当女生在倾诉问题的时候,千万不要只去想解决方案,而是要尽快地安慰女生,倾听她的吐槽,顺着她的思路去稍微吐槽下。可以换位思考,如果你自己心情不好的时候,肯定也想有人来安慰,而不是想听到”你要看开一点,这样工作效率才会高”之类的解决方案的话
-
在倾听和安慰之后,然后把问题解决,或者帮助女生把问题解决。做和说同样重要。