建设银行东莞招聘网站,出名的包装设计,桐乡市住房建设局网站公示,深圳市龙华区教育局官网1、gRPC流
从其名称可以理解#xff0c;流就是持续不断的传输。有一些业务场景请求或者响应的数据量比较大#xff0c;不适合使用普通的
RPC 调用通过一次请求-响应处理#xff0c;一方面是考虑数据量大对请求响应时间的影响#xff0c;另一方面业务场景的设计不一
定需…1、gRPC流
从其名称可以理解流就是持续不断的传输。有一些业务场景请求或者响应的数据量比较大不适合使用普通的
RPC 调用通过一次请求-响应处理一方面是考虑数据量大对请求响应时间的影响另一方面业务场景的设计不一
定需要一次性处理完所有数据这时就可以使用流来分批次传输数据。
HTTP/2中有两个概念流stream与帧frame其中帧作为HTTP/2中通信的最小传输单位通常一个请
求或响应会被分为一个或多个帧传输流则表示已建立连接的虚拟通道可以传输多次请求或响应。每个帧中包含
Stream Identifier标志所属流。HTTP/2通过流与帧实现多路复用对于相同域名的请求通过Stream
Identifier标识可在同一个流中进行从而减少连接开销。 而gRPC基于HTTP/2协议传输自然而然也实现了流式
传输其中gRPC中共有以下三种类型的流
1、服务端流式响应
2、客户端流式请求
3、两端双向流式
本篇主要讲讲如何实现gRPC三种流式处理。
gRPC的stream只需要在service的rpc方法描述中通过 stream 关键字指定启用流特性就好了。
1.1 单向流
单向流是指客户端和服务端只有一端开启流特性这里的单向特指发送数据的方向。 当服务端开启流时客户端和普通 RPC 调用一样通过一次请求发送数据服务端通过流分批次响应。 当客户端开启流时客户端通过流分批次发送请求数据服务端接完所有数据后统一响应一次。
1.1.1 服务端流
定义一个 MultiPong 方法在服务端开启流功能是接收到客户端的请求后响应10次 pong 消息。
ping.proto文件的编写
// ping.proto
// 指定proto版本
syntax proto3;
// 指定包名
package protos;
// 指定go包路径
option go_package protos/ping;
// 定义PingPong服务
service PingPong {// Ping发送ping请求,接收pong响应// 服务端流模式,在响应消息前添加stream关键字rpc MultiPong(PingRequest) returns (stream PongResponse);
}// PingRequest请求结构
message PingRequest {// value字段为string类型string value 1;
}// PongResponse 响应结构
message PongResponse {// value字段为string类型string value 1;
}ping.pb.go文件的生成
$ protoc --go_outpluginsgrpc:. ping.proto服务端实现server.go的编写第二个参数为 stream 对象的引用可以通过它的 Send 方法发送数据。
package mainimport (// 引入编译生成的包pb demo/protos/pinggoogle.golang.org/grpclognet
)// PingPongServer 实现 pb.PingPongServer 接口
type PingPongServer struct {pb.UnimplementedPingPongServer
}// MultiPong 服务端流模式
func (s *PingPongServer) MultiPong(req *pb.PingRequest, stream pb.PingPong_MultiPongServer) error {for i : 0; i 10; i {data : pb.PongResponse{Value: pong}// 发送消息err : stream.Send(data)if err ! nil {return err}}return nil
}// 启动server
func main() {srv : grpc.NewServer()// 注册 PingPongServerpb.RegisterPingPongServer(srv, PingPongServer{})lis, err : net.Listen(tcp, :7009)if err ! nil {log.Fatalln(err)}log.Println(listen on 7009)srv.Serve(lis)
}# 启动server
$ go run server.go
2023/02/10 20:51:04 listen on 7009客户端实现client.go的编写请求方式和普通 RPC 没有区别重点关注对响应数据流的处理通过一个 for
循环接收数据直到结束。
package mainimport (contextpb demo/protos/ping // 引入编译生成的包google.golang.org/grpciolog
)// Ping 单次请求-响应模式
func main() {conn, err : grpc.Dial(localhost:7009, grpc.WithInsecure())if err ! nil {log.Fatalln(err)}defer conn.Close()// 实例化客户端并调用client : pb.NewPingPongClient(conn)// 获得对 stream 对象的引用stream, err : client.MultiPong(context.Background(), pb.PingRequest{Value: ping})if err ! nil {log.Fatalln(err)}// 循环接收响应数据流for {msg, err : stream.Recv()if err ! nil {// 数据结束if err io.EOF {break}log.Fatalln(err)}log.Println(msg.Value)}
}# 客户端运行
$ go run client.go
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong# 目录结构
$ tree demo
demo
├── client.go
├── go.mod
├── go.sum
├── ping.proto
├── protos
│ └── ping
│ └── ping.pb.go
└── server.go2 directories, 6 files1.1.2 客户端流
定义一个 MultiPing 方法在客户端开启流功能是持续发送多个 ping 请求服务端统一响应一次。
ping.proto文件的编写
// ping.proto
// 指定proto版本
syntax proto3;
// 指定包名
package protos;
// 指定go包路径
option go_package protos/ping;
// 定义PingPong服务
service PingPong {// Ping 发送 ping 请求接收 pong 响应// 客户端流模式在请求消息前添加 stream 关键字rpc MultiPing(stream PingRequest) returns (PongResponse);
}// PingRequest 请求结构
message PingRequest {string value 1; // value字段为string类型
}// PongResponse 响应结构
message PongResponse {string value 1; // value字段为string类型
}ping.pb.go文件的生成
$ protoc --go_outpluginsgrpc:. ping.proto服务端实现server.go的编写只有一个参数为 stream 对象的引用可以通过它的 Recv 方法接收数据。使
用 SendAndClose 方法关闭流并响应服务端可以根据需要提前关闭。
package mainimport (fmt// 引入编译生成的包pb demo/protos/pinggoogle.golang.org/grpciolognet
)// PingPongServer 实现 pb.PingPongServer 接口
type PingPongServer struct {pb.UnimplementedPingPongServer
}// MultiPing 客户端流模式
func (s *PingPongServer) MultiPing(stream pb.PingPong_MultiPingServer) error {msgs : []string{}for {// 提前结束接收消息if len(msgs) 5 {return stream.SendAndClose(pb.PongResponse{Value: ping enough, max 5})}msg, err : stream.Recv()if err ! nil {// 客户端消息结束返回响应信息if err io.EOF {return stream.SendAndClose(pb.PongResponse{Value: fmt.Sprintf(got %d ping, len(msgs))})}return err}msgs append(msgs, msg.Value)}
}// 启动server
func main() {srv : grpc.NewServer()// 注册 PingPongServerpb.RegisterPingPongServer(srv, PingPongServer{})lis, err : net.Listen(tcp, :7009)if err ! nil {log.Fatalln(err)}log.Println(listen on 7009)srv.Serve(lis)
}# 启动server
$ go run server.go
2023/02/10 21:26:42 listen on 7009客户端实现client.go的编写调用 MultiPing 方法时不再指定请求参数而是通过返回的 stream 对象的
Send 分批发送数据。
package mainimport (contextpb demo/protos/ping // 引入编译生成的包google.golang.org/grpclog
)// Ping 单次请求-响应模式
func main() {conn, err : grpc.Dial(localhost:7009, grpc.WithInsecure())if err ! nil {log.Fatalln(err)}defer conn.Close()// 实例化客户端并调用client : pb.NewPingPongClient(conn)// 获得对stream对象的引用// 调用并得到stream对象stream, err : client.MultiPing(context.Background())if err ! nil {log.Fatalln(err)}// 发送数据for i : 0; i 6; i {data : pb.PingRequest{Value: ping}err stream.Send(data)if err ! nil {log.Fatalln(err)}}// 发送结束并获取服务端响应res, err : stream.CloseAndRecv()if err ! nil {log.Fatalln(err)}log.Println(res.Value)
}# 启动客户端
# 发送3个ping
$ go run client.go
2023/02/10 21:32:31 got 3 ping
# 发送6个ping
$ go run client.go
2023/02/10 21:32:31 ping enough, max 5# 项目结构
$ tree demo
demo
├── client.go
├── go.mod
├── go.sum
├── ping.proto
├── protos
│ └── ping
│ └── ping.pb.go
└── server.go2 directories, 6 files1.2 双向流
双向流是指客户端在发送数据和服务端响应数据的过程中都启用流特性实际上单向流只是双向流的特例有了上
面的基础双向流就很好理解了。
定义一个 MultiPingPong 方法在客户端和服务端都开启流功能是服务端每接收到两个 ping 就响应一次
pong。
ping.proto编写
// ping.proto
// 指定proto版本
syntax proto3;
// 指定包名
package protos;
// 指定go包路径
option go_package protos/ping;// 定义PingPong服务
service PingPong {// Ping 发送 ping 请求接收 pong 响应// 双向流模式rpc MultiPingPong(stream PingRequest) returns (stream PongResponse);
}// PingRequest 请求结构
message PingRequest {string value 1; // value字段为string类型
}// PongResponse 响应结构
message PongResponse {string value 1; // value字段为string类型
}ping.pb.go文件的生成
$ protoc --go_outpluginsgrpc:. ping.proto服务端实现server.go的编写同样通过 stream 的 Recv 和 Send 方法接收和发送数据。
package mainimport (pb demo/protos/ping // 引入编译生成的包google.golang.org/grpciolognet
)// PingPongServer 实现 pb.PingPongServer 接口
type PingPongServer struct {pb.UnimplementedPingPongServer
}func (s *PingPongServer) MultiPingPong(stream pb.PingPong_MultiPingPongServer) error {msgs : []string{}for {// 接收消息msg, err : stream.Recv()if err ! nil {if err io.EOF {break}return err}msgs append(msgs, msg.Value)// 每收到两个消息响应一次if len(msgs)%2 0 {err stream.Send(pb.PongResponse{Value: pong})if err ! nil {return err}}}return nil
}// 启动server
func main() {srv : grpc.NewServer()// 注册 PingPongServerpb.RegisterPingPongServer(srv, PingPongServer{})lis, err : net.Listen(tcp, :7009)if err ! nil {log.Fatal(err)}log.Println(listen on 7009)srv.Serve(lis)
}# 启动server
$ go run server.go
2023/02/10 21:26:42 listen on 7009客户端实现client.go的编写这里在另外一个 goroutine 里处理接收数据的逻辑来演示同时发送和接收数
据。
package mainimport (contextpb demo/protos/ping // 引入编译生成的包google.golang.org/grpciologtime
)// Ping 单次请求-响应模式
func main() {conn, err : grpc.Dial(localhost:7009, grpc.WithInsecure())if err ! nil {log.Fatal(err)}defer conn.Close()// 实例化客户端并调用client : pb.NewPingPongClient(conn)stream, err : client.MultiPingPong(context.Background())if err ! nil {log.Fatal(err)}// 在另一个goroutine中处理接收数据c : make(chan struct{})go func(stream pb.PingPong_MultiPingPongClient, c chan struct{}) {defer func() {c - struct{}{}}()for {msg, err : stream.Recv()if err ! nil {if err io.EOF {break}log.Fatal(err)}log.Printf(recv:%s\n, msg.Value)}}(stream, c)// 发送数据for i : 0; i 6; i {data : pb.PingRequest{Value: ping}err stream.Send(data)if err ! nil {log.Fatal(err)}log.Printf(send:%s\n, data.Value)// 延时一段时间发送等待响应结果time.Sleep(500 * time.Millisecond)}// 结束发送stream.CloseSend()// 等待接收完成-c
}# 启动客户端
$ go run client.go
2023/02/10 21:48:26 send:ping
2023/02/10 21:48:26 send:ping
2023/02/10 21:48:26 recv:pong
2023/02/10 21:48:27 send:ping
2023/02/10 21:48:27 send:ping
2023/02/10 21:48:27 recv:pong
2023/02/10 21:48:28 send:ping
2023/02/10 21:48:28 send:ping
2023/02/10 21:48:28 recv:pong# 项目结构
$ tree demo
demo
├── client.go
├── go.mod
├── go.sum
├── ping.proto
├── protos
│ └── ping
│ └── ping.pb.go
└── server.go2 directories, 6 files
