在上节课内容中，我们学习了使用gRPC框架实现服务的调用编程。在gRPC框架中，诸如上节课我们学习的在客户端与服务端之间通过消息结构体定义的方式来传递数据，我们称之为“单项RPC”，也称之为简单模式。除此之外，gRPC中还有数据流模式的RPC调用实现，这正是我们本节课要学习的内容。
# 一 服务端流 RPC
在服务端流模式的RPC实现中，服务端得到客户端请求后，处理结束返回一个数据应答流。在发送完所有的客户端请求的应答数据后，服务端的状态详情和可选的跟踪元数据发送给客户端。服务端流RPC实现案例如下：
# 1.1 服务接口定义
在.proto文件中定义服务接口,使用服务端流模式定义服务接口,如下所示：
...
//订单服务service定义
service OrderService {
rpc GetOrderInfos (OrderRequest) returns (stream OrderInfo) {}; //服务端流模式
}
1
2
3
4
5
2
3
4
5
我们可以看到与之前简单模式下的数据作为服务接口的参数和返回值不同的是，此处服务接口的返回值使用了stream进行修饰。通过stream修饰的方式表示该接口调用时，服务端会以数据流的形式将数据返回给客户端。
# 1.2 编译.proto文件，生成pb.go文件
使用gRPC插件编译命令编译.proto文件，编译命令如下:
protoc --go_out=plugins=grpc:. message.proto
1
# 1.3 自动生成文件的变化
与数据结构体发送携带数据实现不同的时，流模式下的数据发送和接收使用新的功能方法完成。在自动生成的go代码程序当中，每一个流模式对应的服务接口，都会自动生成对应的单独的client和server程序，以及对应的结构体实现。具体编程如下图所示：
# 1.3.1 服务端自动生成
type OrderService_GetOrderInfosServer interface {
Send(*OrderInfo) error
grpc.ServerStream
}
type orderServiceGetOrderInfosServer struct {
grpc.ServerStream
}
func (x *orderServiceGetOrderInfosServer) Send(m *OrderInfo) error {
return x.ServerStream.SendMsg(m)
}
1
2
3
4
5
6
7
8
9
10
11
12
2
3
4
5
6
7
8
9
10
11
12
流模式下，服务接口的服务端提供Send方法，将数据以流的形式进行发送
# 1.3.2 客户端自动生成
type OrderService_GetOrderInfosClient interface {
Recv() (*OrderInfo, error)
grpc.ClientStream
}
type orderServiceGetOrderInfosClient struct {
grpc.ClientStream
}
func (x *orderServiceGetOrderInfosClient) Recv() (*OrderInfo, error) {
m := new(OrderInfo)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
流模式下，服务接口的客户端提供Recv()方法接收服务端发送的流数据。
# 1.4 服务编码实现
定义好服务接口并编译生成代码文件后，即可根据规则对定义的服务进行编码实现。具体的服务编码实现如下所示：
//订单服务实现
type OrderServiceImpl struct {
}
//获取订单信息s
func (os *OrderServiceImpl) GetOrderInfos(request *message.OrderRequest, stream message.OrderService_GetOrderInfosServer) error {
fmt.Println(" 服务端流 RPC 模式")
orderMap := map[string]message.OrderInfo{
"201907300001": message.OrderInfo{OrderId: "201907300001", OrderName: "衣服", OrderStatus: "已付款"},
"201907310001": message.OrderInfo{OrderId: "201907310001", OrderName: "零食", OrderStatus: "已付款"},
"201907310002": message.OrderInfo{OrderId: "201907310002", OrderName: "食品", OrderStatus: "未付款"},
}
for id, info := range orderMap {
if (time.Now().Unix() >= request.TimeStamp) {
fmt.Println("订单序列号ID：", id)
fmt.Println("订单详情：", info)
//通过流模式发送给客户端
stream.Send(&info)
}
}
return nil
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
GetOrderInfos方法就是服务接口的具体实现，因为是流模式开发，服务端将数据以流的形式进行发送,因此，该方法的第二个参数类型为OrderService_GetOrderInfosServer，该参数类型是一个接口，其中包含Send方法，允许发送流数据。Send方法的具体实现在编译好的pb.go文件中，进一步调用grpc.SeverStream.SendMsg方法。
# 1.5 服务的注册和监听的处理
服务的监听与处理与前文所学内容没有区别，依然是相同的步骤:
func main() {
server := grpc.NewServer()
//注册
message.RegisterOrderServiceServer(server, new(OrderServiceImpl))
lis, err := net.Listen("tcp", ":8090")
if err != nil {
panic(err.Error())
}
server.Serve(lis)
}
1
2
3
4
5
6
7
8
9
10
2
3
4
5
6
7
8
9
10
# 1.6 客户端数据接收
服务端使用Send方法将数据以流的形式进行发送，客户端可以使用Recv()方法接收流数据,因为数据流失源源不断的，因此使用for无限循环实现数据流的读取，当读取到io.EOF时，表示流数据结束。客户端数据读取实现如下：
...
for {
orderInfo, err := orderInfoClient.Recv()
if err == io.EOF {
fmt.Println("读取结束")
return
}
if err != nil {
panic(err.Error())
}
fmt.Println("读取到的信息：", orderInfo)
}
...
1
2
3
4
5
6
7
8
9
10
11
12
13
2
3
4
5
6
7
8
9
10
11
12
13
# 1.7 运行结果
按照先后顺序，依次运行server.go文件和client.go文件，可以得到运行结果。
# 1.7.1 服务端运行结果
服务端流 RPC 模式
订单序列号ID： 201907300001
订单详情： {201907300001 衣服 已付款 {} [] 0}
订单序列号ID： 201907310001
订单详情： {201907310001 零食 已付款 {} [] 0}
订单序列号ID： 201907310002
订单详情： {201907310002 食品 未付款 {} [] 0}
1
2
3
4
5
6
7
2
3
4
5
6
7
# 1.7.2 客户端运行结果
客户端请求RPC调用：服务端流模式
读取到的信息： OrderId:"201907310001" OrderName:"\351\233\266\351\243\237" OrderStatus:"\345\267\262\344\273\230\346\254\276"
读取到的信息： OrderId:"201907310002" OrderName:"\351\243\237\345\223\201" OrderStatus:"\346\234\252\344\273\230\346\254\276"
读取到的信息： OrderId:"201907300001" OrderName:"\350\241\243\346\234\215" OrderStatus:"\345\267\262\344\273\230\346\254\276"
读取结束
1
2
3
4
5
2
3
4
5
# 二 客户端流模式
上文演示的是服务端以数据流的形式返回数据的形式。对应的，也存在客户端以流的形式发送请求数据的形式。
# 2.1 服务接口的定义
与服务端同理,客户端流模式的RPC服务声明格式，就是使用stream修饰服务接口的接收参数,具体如下所示:
...
//订单服务service定义
service OrderService {
rpc AddOrderList (stream OrderRequest) returns (OrderInfo) {}; //客户端流模式
}
1
2
3
4
5
2
3
4
5
# 2.2 编译.proto文件
使用编译命令编译.protow文件。客户端流模式中也会自动生成服务接口的接口。
# 2.2.1 自动生成的服务流接口实现
type OrderService_AddOrderListServer interface {
SendAndClose(*OrderInfo) error
Recv() (*OrderRequest, error)
grpc.ServerStream
}
type orderServiceAddOrderListServer struct {
grpc.ServerStream
}
func (x *orderServiceAddOrderListServer) SendAndClose(m *OrderInfo) error {
return x.ServerStream.SendMsg(m)
}
func (x *orderServiceAddOrderListServer) Recv() (*OrderRequest, error) {
m := new(OrderRequest)
if err := x.ServerStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
SendAndClose和Recv方法是客户端流模式下的服务端对象所拥有的方法。
# 2.2.2 自动生成的客户端流接口实现
type OrderService_AddOrderListClient interface {
Send(*OrderRequest) error
CloseAndRecv() (*OrderInfo, error)
grpc.ClientStream
}
type orderServiceAddOrderListClient struct {
grpc.ClientStream
}
func (x *orderServiceAddOrderListClient) Send(m *OrderRequest) error {
return x.ClientStream.SendMsg(m)
}
func (x *orderServiceAddOrderListClient) CloseAndRecv() (*OrderInfo, error) {
if err := x.ClientStream.CloseSend(); err != nil {
return nil, err
}
m := new(OrderInfo)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
Send和CloseAndRecv是客户端流模式下的客户端对象所拥有的方法。
# 2.3 服务的实现
客户端流模式的服务接口具体实现如下：
//订单服务实现
type OrderServiceImpl struct {
}
//添加订单信息服务实现
func (os *OrderServiceImpl) AddOrderList(stream message.OrderService_AddOrderListServer) error {
fmt.Println(" 客户端流 RPC 模式")
for {
//从流中读取数据信息
orderRequest, err := stream.Recv()
if err == io.EOF {
fmt.Println(" 读取数据结束 ")
result := message.OrderInfo{OrderStatus: " 读取数据结束 "}
return stream.SendAndClose(&result)
}
if err != nil {
fmt.Println(err.Error())
return err
}
//打印接收到的数据
fmt.Println(orderRequest)
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 2.4 服务的注册和监听处理
依然是采用相同的服务注册和监听处理方式对服务进行注册和监听处理。
func main() {
server := grpc.NewServer()
//注册
message.RegisterOrderServiceServer(server, new(OrderServiceImpl))
lis, err := net.Listen("tcp", ":8090")
if err != nil {
panic(err.Error())
}
server.Serve(lis)
}
1
2
3
4
5
6
7
8
9
10
11
12
2
3
4
5
6
7
8
9
10
11
12
# 2.5 客户端实现
客户端调用send方法流数据到服务端，具体实现如下：
...
//调用服务方法
addOrderListClient, err := orderServiceClient.AddOrderList(context.Background())
if err != nil {
panic(err.Error())
}
//调用方法发送流数据
for _, info := range orderMap {
err = addOrderListClient.Send(&info)
if err != nil {
panic(err.Error())
}
}
for {
orderInfo, err := addOrderListClient.CloseAndRecv()
if err == io.EOF {
fmt.Println(" 读取数据结束了 ")
return
}
if err != nil {
fmt.Println(err.Error())
}
fmt.Println(orderInfo.GetOrderStatus())
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# 2.6 程序运行
# 2.6.1 服务端
运行案例，程序输出如下：
客户端流 RPC 模式
201907300001 衣服 已付款
201907310001 零食 已付款
201907310002 食品 未付款
读取数据结束
客户端流 RPC 模式
201907300001 衣服 已付款
201907310001 零食 已付款
201907310002 食品 未付款
读取数据结束
1
2
3
4
5
6
7
8
9
10
2
3
4
5
6
7
8
9
10
# 2.6.2 客户端
客户端运行程序输出如下：
客户端请求RPC调用：客户端流模式
读取数据结束
读取数据结束了
1
2
3
2
3
# 三 双向流模式
上文已经讲过了服务端流模式和客户端流模式。如果将客户端和服务端两种流模式结合起来,就是第三种模式，双向流模式。即客户端发送数据的时候以流数据发送，服务端返回数据也以流的形式进行发送，因此称之为双向流模式。
# 3.1 双向流服务的定义
//订单服务service定义
service OrderService {
rpc GetOrderInfos (stream OrderRequest) returns (stream OrderInfo) {}; //双向流模式
}
1
2
3
4
2
3
4
# 3.2 编译.proto文件
# 3.2.1 服务端接口实现
type OrderService_GetOrderInfosServer interface {
Send(*OrderInfo) error
Recv() (*OrderRequest, error)
grpc.ServerStream
}
type orderServiceGetOrderInfosServer struct {
grpc.ServerStream
}
func (x *orderServiceGetOrderInfosServer) Send(m *OrderInfo) error {
return x.ServerStream.SendMsg(m)
}
func (x *orderServiceGetOrderInfosServer) Recv() (*OrderRequest, error) {
m := new(OrderRequest)
if err := x.ServerStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 3.2.2 客户端接口实现
type OrderService_GetOrderInfosClient interface {
Send(*OrderRequest) error
Recv() (*OrderInfo, error)
grpc.ClientStream
}
type orderServiceGetOrderInfosClient struct {
grpc.ClientStream
}
func (x *orderServiceGetOrderInfosClient) Send(m *OrderRequest) error {
return x.ClientStream.SendMsg(m)
}
func (x *orderServiceGetOrderInfosClient) Recv() (*OrderInfo, error) {
m := new(OrderInfo)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 3.3 服务实现
//实现grpc双向流模式
func (os *OrderServiceImpl) GetOrderInfos(stream message.OrderService_GetOrderInfosServer) error {
for {
orderRequest, err := stream.Recv()
if err == io.EOF {
fmt.Println(" 数据读取结束 ")
return err
}
if err != nil {
panic(err.Error())
return err
}
fmt.Println(orderRequest.GetOrderId())
orderMap := map[string]message.OrderInfo{
"201907300001": message.OrderInfo{OrderId: "201907300001", OrderName: "衣服", OrderStatus: "已付款"},
"201907310001": message.OrderInfo{OrderId: "201907310001", OrderName: "零食", OrderStatus: "已付款"},
"201907310002": message.OrderInfo{OrderId: "201907310002", OrderName: "食品", OrderStatus: "未付款"},
}
result := orderMap[orderRequest.GetOrderId()]
//发送数据
err = stream.Send(&result)
if err == io.EOF {
fmt.Println(err)
return err
}
if err != nil {
fmt.Println(err.Error())
return err
}
}
return nil
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
# 3.4 服务端及客户端的编程实现
# 3.4.1 服务端实现
func main() {
server := grpc.NewServer()
//注册
message.RegisterOrderServiceServer(server, new(OrderServiceImpl))
lis, err := net.Listen("tcp", ":8092")
if err != nil {
panic(err.Error())
}
server.Serve(lis)
}
1
2
3
4
5
6
7
8
9
10
11
2
3
4
5
6
7
8
9
10
11
# 3.4.2 客户端实现
func main() {
//1、Dail连接
conn, err := grpc.Dial("localhost:8092", grpc.WithInsecure())
if err != nil {
panic(err.Error())
}
defer conn.Close()
orderServiceClient := message.NewOrderServiceClient(conn)
fmt.Println("客户端请求RPC调用：双向流模式")
orderIDs := []string{"201907300001", "201907310001", "201907310002"}
orderInfoClient, err := orderServiceClient.GetOrderInfos(context.Background())
for _, orderID := range orderIDs {
orderRequest := message.OrderRequest{OrderId: orderID}
err := orderInfoClient.Send(&orderRequest)
if err != nil {
panic(err.Error())
}
}
//关闭
orderInfoClient.CloseSend()
for {
orderInfo, err := orderInfoClient.Recv()
if err == io.EOF {
fmt.Println("读取结束")
return
}
if err != nil {
return
}
fmt.Println("读取到的信息：", orderInfo)
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38