【Golang | gRPC】gRPC-Bidirectional Streaming双向流实战

  • A+
所属分类:golang

环境:
Golang: go1.18.2 windows/amd64
grpc: v1.47.0
protobuf: v1.28.0

完整代码:
https://github.com/WanshanTian/GolangLearning
cd GolangLearning/RPC/gRPC-BidirectionalStreaming

1. 简介

前文【Golang | gRPC】HTTP的连接管理——从HTTP/1.0到HTTP/2.0的演进 简单介绍了gRPC中流模式主要分为客户端流、服务端流、双向流以及流传输模式的优点,下面通过一个demo来说明gRPC双向流的使用

2. 实践

现有下面一种场景:服务端保存着用户的年龄信息,客户端通过stream多次发送含用户姓名的message,服务端通过stream接收message,返回指定用户对应的年龄

2.1 proto文件

2.1.1 新建gRPC-BidirectionalStreaming文件夹,使用go mod init初始化,创建pb文件夹,新建query.proto文件

syntax = "proto3";
package pb;
option go_package= ".;pb";

// 定义查询服务包含的方法
service Query {
  // 双向流模式
  rpc GetAge (stream userInfo) returns (stream ageInfo){}
}

// 请求用的结构体,包含一个name字段
message userInfo {
  string name = 1;
}

// 响应用的结构体,包含一个age字段
message ageInfo {
  int32 age = 1;
}

服务端实现一个查询(Query)服务,包含一个方法GetAge。方法GetAge的入参和返回值前加关键字stream来表明该方法启用双向流

2.1.2 在.\gRPC-BidirectionalStreaming\pb目录下使用protoc工具进行编译,在pb文件夹下直接生成.pb.go和_grpc.pb.go文件。关于protoc的详细使用可以查看【Golang | gRPC】使用protoc编译.proto文件

protoc --go_out=./ --go-grpc_out=./ .\query.proto

【Golang | gRPC】gRPC-Bidirectional Streaming双向流实战

2.2 grpc.pb.go文件

2.2.1 查看query_grpc.pb.go中生成的客户端流和服务端流的接口定义以及服务端QueryServer服务的定义

// 客户端流
type Query_GetAgeClient interface {
   
	Send(*UserInfo) error
	Recv() (*AgeInfo, error)
	grpc.ClientStream
}
// 服务端流
type Query_GetAgeServer interface {
   
	Send(*AgeInfo) error
	Recv() (*UserInfo, error)
	grpc.ServerStream
}
// Query服务的客户端接口
type QueryClient interface {
   
	GetAge(ctx context.Context, opts ...grpc.CallOption) (Query_GetAgeClient, error)
}
// Query服务的服务端接口
type QueryServer interface {
   
	GetAge(Query_GetAgeServer) error
	mustEmbedUnimplementedQueryServer() 
}
  • 无论是客户端流还是服务端流,都包含两种方法SendRecv,分别用于向流发送和接收message
  • grpc.ClientStream接口实现了CloseSend() error方法,用于关闭流的发送方向
  • 客户端GetAge方法的第一个返回值是Query_GetAgeClient,表明生成了一条流,用于发送和接收message;如果有多个方法,则每个方法可以各自生成一条流
  • 服务端GetAge方法的入参是Query_GetAgeServer(流),具体方法需要用户自行实现,可以从流中接收和发送message

2.3 服务端

在gRPC-BidirectionalStreaming目录下新建Server文件夹,新建main.go文件

2.3.1 下面我们通过Query这个结构体具体实现QueryServer接口

// 用户信息
var userinfo = map[string]int32{
   
	"foo": 18,
	"bar": 20,
}

// Query 结构体,实现QueryServer接口
type Query struct {
   
	pb.UnimplementedQueryServer // 涉及版本兼容
}

func (q *Query) GetAge(ServerStream pb.Query_GetAgeServer) error {
   
	log.Println("start of stream")
	for {
   
		// 接受message
		userinfoRecv, err := ServerStream.Recv()
		// 待客户端主动关闭流后,退出for循环
		if err == io.EOF {
   
			log.Println("end of stream")
			break
		}
		log.Printf("The name of user received is %s", userinfoRecv.GetName())
		// 返回响应message
		log.Printf("send message about the age of %s", userinfoRecv.GetName())
		err = ServerStream.Send(&pb.AgeInfo{
   Age: userinfo[userinfoRecv.Name]})
		if err != nil {
   
			log.Panic(err)
		}
	}
	return nil
}
  • 服务端每收到一个message,返回对应用户的年龄
  • Recv方法会一直阻塞直到从stream中接收到message,或者直到客户端调用CloseSend方法
  • 当客户端调用CloseSend方法时,服务端调用Recv方法会得到io.EOF返回值

2.3.2 服务注册并启动

func main() {
   
	// 创建socket监听器
	listener, _ := net.Listen("tcp", ":1234")
	// new一个gRPC服务器,用来注册服务
	grpcserver := grpc.NewServer()
	// 注册服务方法
	pb.RegisterQueryServer(grpcserver, new(Query))
	// 开启gRPC服务
	_ = grpcserver.Serve(listener)
}

使用RegisterQueryServer这个方法向gRPC服务器里注册QueryServer

2.4 客户端

在gRPC-BidirectionalStreaming目录下新建Client文件夹,新建main.go文件

2.4.1 先建立无认证的连接,生成Client,然后通过方法GetAge返回对应的流,最后通过流进行message的收发

func main() {
   
	//建立无认证的连接
	conn, _ := grpc.Dial(":1234", grpc.WithTransportCredentials(insecure.NewCredentials()))

	defer conn.Close()
	client := pb.NewQueryClient(conn)

	//返回GetAge方法对应的流
	log.Printf("start of stream")
	queryStream, _ := client.GetAge(context.Background())

	// 创建goroutine用来向stream中发送message
	ch := make(chan string, 2)
	go func() {
   
		names := []string{
   "foo", "bar"}
		for _, v := range names {
   
			log.Printf("send message wtih Name is %s\n", v)
			ch <- v
			_ = queryStream.Send(&pb.UserInfo{
   Name: v})
			time.Sleep(time.Second)
		}
		// 调用指定次数后主动关闭流
		err := queryStream.CloseSend()
		if err != nil {
   
			log.Println(err)
		}
		close(ch)
	}()

	// 从stream中接收message
	for {
   
		name := <-ch
		ageinfoRecv, err := queryStream.Recv()
		if err == io.EOF {
   
			log.Println("end of stream")
			break
		}
		log.Printf("The age of %s is %d\n", name, ageinfoRecv.GetAge())
	}
}
  • 新建一个goroutine用于发送message,注意由于使用的HTTP/2.0协议,发送一个请求message后可以不用等待响应message即可继续发送
  • 客户端通过CloseSend方法主动关闭发送方向的流
  • 主函数里通过for循环接收来自服务端的响应message。当客户端主动关闭流,服务端在返回最后一个响应message后客户端通过Recv方法会得到返回值io.EOF(标志着整个流的结束)

运行结果如下:
【Golang | gRPC】gRPC-Bidirectional Streaming双向流实战

3. 总结

  • 无论是客户端还是服务端Recv方法会一直阻塞直到收到message或者对端关闭stream
  • 当一方关闭stream时,对端会返回io.EOF
w3cjava