本文最后更新于：1 年前

前言

这章的内容需要安装好插件和protoc，建议阅读我的上一篇grpc使用篇
示例代码已经上传到github：点击跳转
gRPC官方文档：点击跳转

一、gRPC的请求模型

gRPC 有两种类型的请求模型：

一元 - 直接的请求响应映射在 HTTP/2 请求响应之上。
- 简单来说一元就是一个简单的 RPC，其中客户端使用存根向服务器发送请求并等待响应返回，就像正常的函数调用一样。

1	`rpc SayHi(Request) returns (Response);`

流式传输——多个请求和响应通过长寿命 HTTP/2 流进行交换，可以是单向或双向的。
- 其中许多进程可以通过 HTTP/2 的多路复用能力（通过单个 TCP 连接一起发送多个响应或接收多个请求）在单个请求中发生。
- Server-side streaming RPC—— 客户端向服务器发送单个请求并接收回数据序列流（读回一系列消息）。客户端从返回的流中读取，直到没有更多消息为止。
- Client-side streaming RPC—— 客户端向服务器发送数据序列流（写入一系列消息），一旦客户端完成了消息的写入，它会等待服务器读取所有消息并返回其响应结果。
- Bidirectional streaming RPC—— 它是双向流式传输，客户端和服务器使用读写流发送一系列消息。两个流独立运行；因此，因此客户端和服务器可以按照他们喜欢的任何顺序读取和写入。保留每个流中消息的顺序。例如，服务器可以在写入响应之前等待接收所有客户端消息，或者它可以交替读取消息然后写入消息，或其他一些读取和写入的组合。

Server-side streaming RPC：服务器端流式 RPC
Client-side streaming RPC：客户端流式 RPC
Bidirectional streaming RPC：双向流式 RPC
stream可以通过将关键字放在请求类型之前来指定流式处理方法。

二、HTTP/2

gRPC 是基于HTTP/2开发的，该协议于 2015 年发布，以克服 HTTP/1.1 的限制。在兼容 HTTP/1.1 的同时，我们来了解一下HTTP/2 带来了许多高级功能，例如：

二进制分帧层 —— 与 HTTP/1.1 不同，HTTP/2 请求/响应分为小消息并以二进制格式分帧，使消息传输高效。通过二进制帧，HTTP/2 协议使请求/响应多路复用成为可能，而不会阻塞网络资源。
流式传输 —— 客户端可以请求并且服务器可以同时响应的双向全双工流式传输。
流控制 —— HTTP/2 中使用流控制机制，可以对用于缓冲动态消息的内存进行详细控制。
标头压缩 —— HTTP/2 中的所有内容，包括标头，都在发送前进行编码，显着提高了整体性能。使用 HPACK 压缩方式，HTTP/2 只共享与之前的 HTTP 头包不同的值。
处理 —— 使用 HTTP/2，gRPC 支持同步和异步处理，可用于执行不同类型的交互和流式 RPC。

HTTP/2 的所有这些特性使 gRPC 能够使用更少的资源，从而减少在云中运行的应用程序和服务之间的响应时间，并延长运行移动设备的客户端的电池寿命。

三、gRPC Streaming, Client and Server

1、为什么我们要用流式传输，简单的一元RPC不行么？

流式为什么要存在呢？我们在使用一元请求的时候可能会遇到以下问题：

数据包过大会造成的瞬时压力。
接收数据包时，需要所有数据包都接受成功且正确后，才能够回调响应，进行业务处理（无法客户端边发送，服务端边处理）

而流式传输却可以：

HTTP2 通过长期 TCP 连接多路复用流，因此新请求没有 TCP 连接开销。HTTP2 成帧允许在单个 TCP 数据包中发送多个 gRPC 消息。
对于长期连接，流式请求应该在每条消息的基础上具有最佳性能。一元请求需要为每个请求建立一个新的 HTTP2 流，包括通过网络发送的附加标头帧。一旦建立，通过流式请求发送的每条新消息只需要通过连接发送消息的数据帧。

2、目录结构

go-grpc-example
├── client
│   └──hello_client
│   │   └── client.go
│   └── stream_client
│       └── client.go
├── proto
│   └──hello
│   │   └── hello.proto
│   └──stream
│   │   └── stream.proto
├── server
│   └──hello_server
│   │   └── server.go
│   └──stream_server
│   │   └── server.go
├── Makefile

增加 stream_server、stream_client 存放服务端和客户端文件，proto/stream/stream.proto 用于编写 IDL

3、编写IDL

在 proto/stream 文件夹下的 stream.proto 文件中，写入如下内容：

syntax = "proto3";

option go_package="./proto/stream;stream";
package proto;

service StreamService {
  //List：服务器端流式 RPC
  rpc List(StreamRequest) returns (stream StreamResponse) {};
  //Record：客户端流式 RPC
  rpc Record(stream StreamRequest) returns (StreamResponse) {};
  //Route：双向流式 RPC
  rpc Route(stream StreamRequest) returns (stream StreamResponse) {};
}


message StreamPoint {
  string name = 1;
  int32 value = 2;
}

message StreamRequest {
  StreamPoint pt = 1;
}

message StreamResponse {
  StreamPoint pt = 1;
}

注意关键字 stream，声明其为一个流方法。这里共涉及三个方法，对应关系为

List：服务器端流式 RPC
Record：客户端流式 RPC
Route：双向流式 RPC

4、Makefile

这是我拖了很久的关于Makefile的用法，感觉Makefile更适合在项目使用中穿插讲解一下。

有一篇很不错的Makefile文档：点击跳转

作用：Makefile 用于帮助决定大型程序的哪些部分需要重新编译。

这里我们用make gen指令代替proto插件从我们的.proto 服务定义中生成 gRPC 客户端和服务器接口。

在Makefile文件中写入：

1 2	`gen: protoc --go_out=. --go-grpc_out=. ./proto/stream/*.proto`

用make gen指令生成Go代码：

1 2	`➜ make gen protoc --go_out=. --go-grpc_out=. ./proto/stream/*.proto`

注意使用Makefile生成的时候，要注意.proto文件 go_package 指定生成的位置。

5、写出基础模板和空定义

我们先把基础的模板和空定义写出来在进行完善，不太懂的看我上一篇文章

1）server.go

type StreamService struct {
	pb.UnimplementedStreamServiceServer
}

const PORT = "8888"

func main() {
	server := grpc.NewServer() //创建 gRPC Server 对象
	pb.RegisterStreamServiceServer(server, &StreamService{})

	lis, err := net.Listen("tcp", ":"+PORT)
	if err != nil {
		log.Fatalf("net.Listen err: %v", err)
	}

	server.Serve(lis)
}

//服务端流式RPC，Server是Stream，Client为普通RPC请求
//客户端发送一次普通的RPC请求，服务端通过流式响应多次发送数据集
func (s *StreamService) List(r *pb.StreamRequest, stream pb.StreamService_ListServer) error {
	return nil
}

//客户端流式RPC，单向流
//客户端通过流式多次发送RPC请求给服务端，服务端发送一次普通的RPC请求给客户端
func (s *StreamService) Record(stream pb.StreamService_RecordServer) error {
	return nil
}

//双向流，由客户端发起流式的RPC方法请求，服务端以同样的流式RPC方法响应请求
//首个请求一定是client发起，具体交互方法（谁先谁后，一次发多少，响应多少，什么时候关闭）根据程序编写方式来确定（可以结合协程）
func (s *StreamService) Route(stream pb.StreamService_RouteServer) error {
	return nil
}

2）client.go

const PORT = "8888"

func main() {
	conn, err := grpc.Dial(":"+PORT, grpc.WithInsecure())
	if err != nil {
		log.Fatalf("grpc.Dial err: %v", err)
	}
	defer conn.Close()

	client := pb.NewStreamServiceClient(conn)

	err = printLists(client, &pb.StreamRequest{Pt: &pb.StreamPoint{Name: "gRPC Stream Client: List", Value: 1234}})
	if err != nil {
		log.Fatalf("printLists.err: %v", err)
	}

	err = printRecord(client, &pb.StreamRequest{Pt: &pb.StreamPoint{Name: "gRPC Stream Client: Record", Value: 9999}})
	if err != nil {
		log.Fatalf("printRecord.err: %v", err)
	}

	err = printRoute(client, &pb.StreamRequest{Pt: &pb.StreamPoint{Name: "gRPC Stream Client: Route", Value: 1111}})
	if err != nil {
		log.Fatalf("printRoute.err: %v", err)
	}
}

func printLists(client pb.StreamServiceClient, r *pb.StreamRequest) error {
	return nil
}

func printRecord(client pb.StreamServiceClient, r *pb.StreamRequest) error {
	return nil
}

func printRoute(client pb.StreamServiceClient, r *pb.StreamRequest) error {
	return nil
}

6、Server-side streaming RPC：服务器端流式 RPC

服务端流式RPC，Server是Stream，Client为普通RPC请求，客户端发送一次普通的RPC请求，服务端通过流式响应多次发送数据集。

1）server

/*
1. 建立连接 获取client
2. 通过 client 获取stream
3. for循环中通过stream.Recv()依次获取服务端推送的消息
4. err==io.EOF则表示服务端关闭stream了
*/
func (s *StreamService) List(r *pb.StreamRequest, stream pb.StreamService_ListServer) error {
	// 具体返回多少个response根据业务逻辑调整
	for n := 0; n <= 6; n++ {
		// 通过 send 方法不断推送数据
		err := stream.Send(&pb.StreamResponse{
			Pt: &pb.StreamPoint{
				Name:  r.Pt.Name,
				Value: r.Pt.Value + int32(n),
			},
		})
		if err != nil {
			return err
		}
		time.Sleep(time.Second)
	}
	// 返回nil表示已经完成响应
	return nil
}

在 Server，主要留意 stream.Send 方法。它看上去能发送 N 次？有没有大小限制？

type StreamService_ListServer interface {
	Send(*StreamResponse) error
	grpc.ServerStream
}

func (x *streamServiceListServer) Send(m *StreamResponse) error {
	return x.ServerStream.SendMsg(m)
}

通过阅读源码，可得知是 protoc 在生成时，根据定义生成了各式各样符合标准的接口方法。最终再统一调度内部的 SendMsg 方法，该方法涉及以下过程:

消息体（对象）序列化
压缩序列化后的消息体
对正在传输的消息体增加 5 个字节的 header
判断压缩+序列化后的消息体总字节长度是否大于预设的 maxSendMessageSize（预设值为 math.MaxInt32），若超出则提示错误
写入给流的数据集

2）client

/*
1. 建立连接 获取client
2. 通过 client 获取stream
3. for循环中通过stream.Recv()依次获取服务端推送的消息
4. err==io.EOF则表示服务端关闭stream了
*/
func printLists(client pb.StreamServiceClient, r *pb.StreamRequest) error {
	// 调用获取stream
	stream, err := client.List(context.Background(), r)
	if err != nil {
		return err
	}
	// for循环获取服务端推送的消息
	for {
		// 通过 Recv() 不断获取服务端send()推送的消息
		resp, err := stream.Recv()
		// err==io.EOF则表示服务端关闭stream了
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
		}
		log.Printf("resp: pj.name: %s, pt.value: %d", resp.Pt.Name, resp.Pt.Value)
	}
	return nil
}

在 Client，主要留意 stream.Recv() 方法。什么情况下 io.EOF ？什么情况下存在错误信息呢?

type StreamService_ListClient interface {
	Recv() (*StreamResponse, error)
	grpc.ClientStream
}

func (x *streamServiceListClient) Recv() (*StreamResponse, error) {
	m := new(StreamResponse)
	if err := x.ClientStream.RecvMsg(m); err != nil {
		return nil, err
	}
	return m, nil
}

RecvMsg 会从流中读取完整的 gRPC 消息体，另外通过阅读源码可得知：

（1）RecvMsg 是阻塞等待的

（2）RecvMsg 当流成功/结束（调用了 Close）时，会返回 io.EOF

（3）RecvMsg 当流出现任何错误时，流会被中止，错误信息会包含 RPC 错误码。而在 RecvMsg 中可能出现如下错误：

io.EOF
io.ErrUnexpectedEOF
transport.ConnectionError
google.golang.org/grpc/codes

3）启动 & 请求

# 启动服务端
$ go run server.go
API server listening at: 127.0.0.1:55149

# 启动客户端
$ go run client.go 
API server listening at: 127.0.0.1:55158
2022/11/03 09:35:03 resp: pj.name: gRPC Stream Client: List, pt.value: 1234
2022/11/03 09:35:04 resp: pj.name: gRPC Stream Client: List, pt.value: 1235
2022/11/03 09:35:05 resp: pj.name: gRPC Stream Client: List, pt.value: 1236
2022/11/03 09:35:06 resp: pj.name: gRPC Stream Client: List, pt.value: 1237
2022/11/03 09:35:07 resp: pj.name: gRPC Stream Client: List, pt.value: 1238
2022/11/03 09:35:08 resp: pj.name: gRPC Stream Client: List, pt.value: 1239
2022/11/03 09:35:09 resp: pj.name: gRPC Stream Client: List, pt.value: 1240

服务器流式 RPC 类似于一元 RPC，除了服务器返回消息流以响应客户端的请求。发送所有消息后，服务器的状态详细信息（状态代码和可选状态消息）和可选尾随元数据将发送到客户端。这样就完成了服务器端的处理。客户端在拥有服务器的所有消息后完成。

7、Client-side streaming RPC：客户端流式 RPC

客户端通过流式多次发送RPC请求给服务端，服务端发送一次响应给客户端。

1）server

/*
1. for循环中通过stream.Recv()不断接收client传来的数据
2. err == io.EOF表示客户端已经发送完毕关闭连接了,此时在等待服务端处理完并返回消息
3. stream.SendAndClose() 发送消息并关闭连接(虽然在客户端流里服务器这边并不需要关闭 但是方法还是叫的这个名字，内部也只会调用Send())
*/
func (s *StreamService) Record(stream pb.StreamService_RecordServer) error {
	// for循环接收客户端发送的消息
	for {
		// 通过 Recv() 不断获取客户端 send()推送的消息
		r, err := stream.Recv()
		// err == io.EOF表示已经获取全部数据
		if err == io.EOF {
			// SendAndClose 返回并关闭连接
			// 在客户端发送完毕后服务端即可返回响应
			return stream.SendAndClose(&pb.StreamResponse{Pt: &pb.StreamPoint{Name: "gRPC Stream Server: Record", Value: 1}})
		}
		if err != nil {
			return err
		}
		log.Printf("stream.Recv pt.name: %s, pt.value: %d", r.Pt.Name, r.Pt.Value)
		time.Sleep(time.Second)
	}
	return nil
}

stream.SendAndClose：我们对每一个 Recv 都进行了处理，当发现 io.EOF (流关闭) 后，需要将最终的响应结果发送给客户端，同时关闭正在另外一侧等待的 Recv

2）client

/*
1. 建立连接并获取client
2. 获取 stream 并通过 Send 方法不断推送数据到服务端
3. 发送完成后通过stream.CloseAndRecv() 关闭stream并接收服务端返回结果
*/
func printRecord(client pb.StreamServiceClient, r *pb.StreamRequest) error {
	// 获取 stream
	stream, err := client.Record(context.Background())
	if err != nil {
		return err
	}

	for i := 0; i <= 6; i++ {
		// 通过 Send 方法不断推送数据到服务端
		err := stream.Send(r)
		if err != nil {
			return err
		}
	}

	// 发送完成后通过stream.CloseAndRecv() 关闭stream并接收服务端返回结果
	// (服务端则根据err==io.EOF来判断client是否关闭stream)
	resp, err := stream.CloseAndRecv()
	if err != nil {
		return err
	}
	log.Printf("resp: pj.name: %s, pt.value: %d", resp.Pt.Name, resp.Pt.Value)
	return nil
}

stream.CloseAndRecv 和 stream.SendAndClose 是配套使用的流方法

3）启动 & 请求

# 启动服务端
$ go run server.go
API server listening at: 127.0.0.1:57789
2022/11/03 11:59:31 stream.Recv pt.name: gRPC Stream Client: Record, pt.value: 9999
2022/11/03 11:59:32 stream.Recv pt.name: gRPC Stream Client: Record, pt.value: 9999
2022/11/03 11:59:33 stream.Recv pt.name: gRPC Stream Client: Record, pt.value: 9999
2022/11/03 11:59:34 stream.Recv pt.name: gRPC Stream Client: Record, pt.value: 9999
2022/11/03 11:59:35 stream.Recv pt.name: gRPC Stream Client: Record, pt.value: 9999
2022/11/03 11:59:36 stream.Recv pt.name: gRPC Stream Client: Record, pt.value: 9999
2022/11/03 11:59:37 stream.Recv pt.name: gRPC Stream Client: Record, pt.value: 9999

# 启动客户端
$ go run client.go 
API server listening at: 127.0.0.1:57793
2022/11/03 11:59:38 resp: pj.name: gRPC Stream Server: Record, pt.value: 1

8、Bidirectional streaming RPC：双向流式 RPC

双向流，由客户端发起流式的RPC方法请求，服务端以同样的流式RPC方法响应请求首个请求一定是client发起，具体交互方法（谁先谁后，一次发多少，响应多少，什么时候关闭）根据程序编写方式来确定（可以结合协程）。

1）server

一般是使用两个 Goroutine，一个接收数据，一个推送数据。最后通过 return nil 表示已经完成响应。

/*
// 1. 建立连接 获取client
// 2. 通过client调用方法获取stream
// 3. 开两个goroutine（使用 chan 传递数据） 分别用于Recv()和Send()
// 3.1 一直Recv()到err==io.EOF(即客户端关闭stream)
// 3.2 Send()则自己控制什么时候Close 服务端stream没有close 只要跳出循环就算close了。 具体见https://github.com/grpc/grpc-go/issues/444
*/
func (s *StreamService) Route(stream pb.StreamService_RouteServer) error {
	var (
		wg    sync.WaitGroup //任务编排
		msgCh = make(chan *pb.StreamPoint)
	)
	wg.Add(1)
	go func() {
		n := 0
		defer wg.Done()
		for v := range msgCh {
			err := stream.Send(&pb.StreamResponse{
				Pt: &pb.StreamPoint{
					Name:  v.GetName(),
					Value: int32(n),
				},
			})
			if err != nil {
				fmt.Println("Send error :", err)
				continue
			}
			n++
		}
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			r, err := stream.Recv()
			if err == io.EOF {
				break
			}
			if err != nil {
				log.Fatalf("recv error :%v", err)
			}
			log.Printf("stream.Recv pt.name: %s, pt.value: %d", r.Pt.Name, r.Pt.Value)
			msgCh <- &pb.StreamPoint{
				Name: "gRPC Stream Server: Route",
			}
		}
		close(msgCh)
	}()

	wg.Wait() //等待任务结束

	return nil
}

2）client

和服务端类似，不过客户端推送结束后需要主动调用 stream.CloseSend() 函数来关闭Stream。

/*
1. 建立连接 获取client
2. 通过client获取stream
3. 开两个goroutine 分别用于Recv()和Send()
	3.1 一直Recv()到err==io.EOF(即服务端关闭stream)
	3.2 Send()则由自己控制
4. 发送完毕调用 stream.CloseSend()关闭stream 必须调用关闭 否则Server会一直尝试接收数据 一直报错...
*/
func printRoute(client pb.StreamServiceClient, r *pb.StreamRequest) error {
	var wg sync.WaitGroup
	// 调用方法获取stream
	stream, err := client.Route(context.Background())
	if err != nil {
		return err
	}

	// 开两个goroutine 分别用于Recv()和Send()
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			resp, err := stream.Recv()
			if err == io.EOF {
				fmt.Println("Server Closed")
				break
			}
			if err != nil {
				continue
			}
			log.Printf("resp: pj.name: %s, pt.value: %d", resp.Pt.Name, resp.Pt.Value)
		}
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()

		for n := 0; n <= 6; n++ {
			err := stream.Send(r)
			if err != nil {
				log.Printf("send error:%v\n", err)
			}
			time.Sleep(time.Second)
		}

		// 发送完毕关闭stream
		err = stream.CloseSend()
		if err != nil {
			log.Printf("Send error:%v\n", err)
			return
		}
	}()

	wg.Wait()
	return nil
}

3）启动 & 请求

# 启动服务端
$ go run server.go
API server listening at: 127.0.0.1:55108
2022/11/03 12:29:35 stream.Recv pt.name: gRPC Stream Client: Route, pt.value: 1111
2022/11/03 12:29:36 stream.Recv pt.name: gRPC Stream Client: Route, pt.value: 1111
2022/11/03 12:29:37 stream.Recv pt.name: gRPC Stream Client: Route, pt.value: 1111
2022/11/03 12:29:38 stream.Recv pt.name: gRPC Stream Client: Route, pt.value: 1111
2022/11/03 12:29:39 stream.Recv pt.name: gRPC Stream Client: Route, pt.value: 1111
2022/11/03 12:29:40 stream.Recv pt.name: gRPC Stream Client: Route, pt.value: 1111
2022/11/03 12:29:41 stream.Recv pt.name: gRPC Stream Client: Route, pt.value: 1111

# 启动客户端
$ go run client.go 
API server listening at: 127.0.0.1:55113
2022/11/03 12:29:35 resp: pj.name: gRPC Stream Server: Route, pt.value: 0
2022/11/03 12:29:36 resp: pj.name: gRPC Stream Server: Route, pt.value: 1
2022/11/03 12:29:37 resp: pj.name: gRPC Stream Server: Route, pt.value: 2
2022/11/03 12:29:38 resp: pj.name: gRPC Stream Server: Route, pt.value: 3
2022/11/03 12:29:39 resp: pj.name: gRPC Stream Server: Route, pt.value: 4
2022/11/03 12:29:40 resp: pj.name: gRPC Stream Server: Route, pt.value: 5
2022/11/03 12:29:41 resp: pj.name: gRPC Stream Server: Route, pt.value: 6
Server Closed

四、小结

客户端或者服务端都有对应的推送或者接收对象，我们只要不断循环 Recv()或者 Send() 就能接收或者推送了！

gRPC Stream 和 goroutine 配合简直完美。通过 Stream 我们可以更加灵活的实现自己的业务。如订阅，大数据传输等。

Client发送完成后需要手动调用Close()或者CloseSend()方法关闭stream，Server端则return nil就会自动 Close。

1）ServerStream

服务端处理完成后return nil代表响应完成
客户端通过 err == io.EOF判断服务端是否响应完成

2）ClientStream

客户端发送完毕通过CloseAndRecv关闭stream 并接收服务端响应
服务端通过 err == io.EOF判断客户端是否发送完毕，完毕后使用SendAndClose关闭 stream并返回响应。

3）BidirectionalStream

客户端服务端都通过stream向对方推送数据
客户端推送完成后通过CloseSend关闭流，通过err == io.EOF判断服务端是否响应完成
服务端通过err == io.EOF判断客户端是否响应完成,通过return nil表示已经完成响应
通过err == io.EOF来判定是否把对方推送的数据全部获取到了。

客户端通过CloseAndRecv或者CloseSend关闭 Stream，服务端则通过SendAndClose或者直接 return nil来返回响应。

参考文章：
https://www.lixueduan.com/posts/grpc/03-stream/#4-serverstream

gRPC学习

#学习 #go #微服务 #grpc

gRPC（四）基础：gRPC流

https://gopherlinzy.github.io/2022/11/03/grpc-stream/

作者

孙禄毅

发布于

2022年11月3日

许可协议

gRPC（五）进阶：通过TLS建立安全连接上一篇

gRPC（三）基础：gRPC快速入门下一篇