前言

这章的内容需要安装好插件和protoc，建议阅读我的上一篇 grpc使用篇

个人网站：linzyblog.netlify.app/

示例代码已经上传到github：点击跳转

gRPC官方文档：点击跳转

一、gRPC的请求模型

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

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


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文件中写入：

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

用make gen指令生成Go代码：

 
➜ 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来返回响应。

参考文章：

	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

	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;
	}

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

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

	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
	}

	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
	}

	/*
	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
	}

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

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

	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
	}

	# 启动服务端
	$ 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

	/*
	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
	}

	/*
	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
	}

	# 启动服务端
	$ 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

	/*
	// 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
	}

	/*
	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
	}

	# 启动服务端
	$ 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

gRPC（四）基础：gRPC流

前言

一、gRPC的请求模型

二、HTTP/2

三、gRPC Streaming, Client and Server

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

2、目录结构

3、编写IDL

4、Makefile

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

1）server.go

2）client.go

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

1）server

2）client

3）启动 & 请求

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

1）server

2）client

3）启动 & 请求

8、Bidirectional streaming RPC：双向流式 RPC

1）server

2）client

3）启动 & 请求

四、小结