gRPC入门指南 — 客户端流式RPC（三）

前言

前一篇文章我们学习了服务端流式RPC，客户端发送一次请求，通过流的方式多次从服务端收到信息。这一节我们来学习下客户端流式RPC，该模式与服务端流式RPC正好相反，客户端不断向服务端发送数据，结束之后，服务端返回一个响应，如下：

新建并编译 proto 文件

新建 client_stream.proto 文件：

syntax = "proto3";

package proto;

// 定义流式请求信息
message StreamRequest{
  // 参数类型 参数名称 标识号
  string data = 1;
}

// 定义响应信息
message SimpleResponse{
  int32 code = 1;
  string value = 2;
}

// 定义我们的服务(可以定义多个服务，每个服务可以定义多个接口)
service StreamService{
  // 客户端流式RPC，需要在请求数据前加stream
  rpc Record(stream StreamRequest) returns (SimpleResponse){};
}

客户端流式 RPC，定义方法时需要在请求值之前加上 stream。

进入 client_stream.proto 所在的目录，使用如下命令编译文件

protoc --go_out=plugins=grpc:. client_stream.proto

执行完成之后会生成 client_stream.pb.go 文件。

创建server端

package main

import (
 pb "go-grpc-example/3-client_stream_rpc/proto"
 "google.golang.org/grpc"
 "io"
 "log"
 "net"
)

const (
 Address string = ":8000"
 Network string = "tcp"
)

// 定义我们的服务
type StreamService struct{}

// 实现 Record 方法
func (s *StreamService) Record(srv pb.StreamService_RecordServer) error {
 for {
  // 从流中获取消息
  req, err := srv.Recv()
  if err == io.EOF {
   // 发送数据并关闭
   return srv.SendAndClose(&pb.SimpleResponse{
    Code:  1,
    Value: "ok",
   })
  }
  if err != nil {
   return err
  }
  log.Printf("get from client:%v", req.Data)
 }
}

func main() {
 // 1.监听端口
 listener, err := net.Listen(Network, Address)
 if err != nil {
  log.Fatalf("listener err: %v", err)
 }
 log.Println(Address + " net.Listing...")

 // 2.创建gRPC服务端实例
 grpcServer := grpc.NewServer()

 // 3.注册我们实现的服务 StreamService
 pb.RegisterStreamServiceServer(grpcServer, &StreamService{})

 // 4.启动gRPC服务端
 err = grpcServer.Serve(listener)
 if err != nil {
  log.Fatalf("grpc server err: %v", err)
 }
}

在实现的 Record() 方法中，可以看到 server 端在 for 循环中不断从客户端接收消息，知道接收完毕，服务端返回一个响应。

运行服务端：

go run server.go

输出：
:8000  net listening...

创建client端

package main

import (
 "context"
 pb "go-grpc-example/3-client_stream_rpc/proto"
 "google.golang.org/grpc"
 "log"
 "strconv"
 "time"
)

const Address string = ":8000"

func main() {
 // 1.连接服务端
 conn, err := grpc.Dial(Address, grpc.WithInsecure())
 if err != nil {
  log.Fatalf("grpc conn err: %v", err)
 }
 defer conn.Close()

 // 2.建立gRPC连接
 streamClient := pb.NewStreamServiceClient(conn)

 // 3.调用record，获取流
 stream, err := streamClient.Record(context.Background())
 if err != nil {
  log.Fatalf("call record err: %v", err)
 }

 for i := 0; i < 5; i++ {
  // 4.向流中发送数据
  err := stream.Send(&pb.StreamRequest{Data: strconv.Itoa(i)})
  if err != nil {
   log.Fatalf("stream request err: %v", err)
  }
  time.Sleep(1 * time.Second)
 }
 // 5.关闭流并获取返回的消息
 resp, err := stream.CloseAndRecv()
 if err != nil {
  log.Fatalf("client stream close err: %v", err)
 }
 log.Printf("get from server,code:%v,value:%v", resp.GetCode(), resp.GetValue())
}

客户端代码，在 for 循环里面向服务端发送了 5 次消息，接着调用 CloseAndRecv() 关闭流并接收服务端返回的数据。

运行客户端：

go run client.go

服务端输出：

get from client:0
get from client:1
get from client:2
get from client:3
get from client:4

服务端输出之后，客户端输出：

get from server,code:1,value:ok

总结

这篇文章主要介绍了客户端流式 RPC的简单使用，该模式下客户端可以多次向服务端发送数据，数据发送完毕之后，服务端会返回一次响应。下篇文章我们会介绍双向流式 RPC。