前言

前一篇文章我們學習了客戶端流式 RPC，客戶端多次向服務端發(fā)送數(shù)據(jù)，發(fā)送結(jié)束之后，由服務端返回一個響應。與服務端流式 RPC類似，都只支持單項連續(xù)發(fā)送數(shù)據(jù)，今天我們要來學習雙向流式 RPC 支持通信雙方同時多次發(fā)送或接收數(shù)據(jù)。如下如所示：

新建并編譯proto文件

新建 bidirectional_stream.proto 文件：

syntax = "proto3";

package proto;

// 定義流式請求信息
message StreamRequest{
  // 參數(shù)類型 參數(shù)名稱 標識號
  string data = 1;
}

// 定義流響應信息
message StreamResponse{
  int32 code = 1;
  string value = 2;
}

// 定義我們的服務(可以定義多個服務，每個服務可以定義多個接口)
service StreamService{
  // 雙向流RPC，需要在請求、響應數(shù)據(jù)前加stream
  rpc Record(stream StreamRequest) returns (stream StreamResponse){};
}

雙向流式 RPC，定義方法時需要在請求值和返回值之前加上 stream。

進入 bidirectional_stream.proto 所在的目錄，使用如下命令編譯文件

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

執(zhí)行完成之后會生成 bidirectional_stream.pb.go 文件。

創(chuàng)建server端

package main

import (
 pb "go-grpc-example/4-bidirectional_stream_rpc/proto"
 "google.golang.org/grpc"
 "io"
 "log"
 "net"
 "strconv"
 "time"
)

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

// 定義我們的服務
type StreamService struct{}

// 實現(xiàn) Record() 方法
func (s *StreamService) Record(srv pb.StreamService_RecordServer) error {
 n := 1
 for {
  // 接收數(shù)據(jù)
  req, err := srv.Recv()
  if err == io.EOF {
   return nil
  }
  if err != nil {
   log.Fatalf("stream get from client err: %v", err)
   return err
  }
  // 發(fā)送數(shù)據(jù)
  err = srv.Send(&pb.StreamResponse{
   Code:  int32(n),
   Value: "This is the " + strconv.Itoa(n) + " message",
  })
  if err != nil {
   log.Fatalf("stream send to client err: %v", err)
   return err
  }
  n++
  log.Println("stream get from client: ", req.Data)
  time.Sleep(1 * time.Second)
 }
 return nil
}

func main() {
 // 1.監(jiān)聽端口
 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.注冊我們的服務
 pb.RegisterStreamServiceServer(grpcServer, &StreamService{})

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

在實現(xiàn)的 Record() 方法中，for() 循環(huán)里面讀取客戶端發(fā)送的消息并返回一個響應數(shù)據(jù)。

運行服務端：

go run server.go

輸出：
:8000  net listening...

創(chuàng)建client端

package main

import (
 "context"
 pb "go-grpc-example/4-bidirectional_stream_rpc/proto"
 "google.golang.org/grpc"
 "io"
 "log"
 "strconv"
 "time"
)

const Address = ":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.創(chuàng)建gRPC客戶端
 grpcClient := pb.NewStreamServiceClient(conn)

 // 3.調(diào)用 Record() 方法獲取流
 stream, err := grpcClient.Record(context.Background())
 if err != nil {
  log.Fatalf("call record err: %v", err)
 }

 for i := 0; i < 5; i++ {
  // 4.發(fā)送數(shù)據(jù)
  err := stream.Send(&pb.StreamRequest{
   Data: strconv.Itoa(i),
  })
  if err != nil {
   log.Fatalf("stream send to server err: %v", err)
  }
  // 5.接收服務端發(fā)送過來的數(shù)據(jù)
  resp, err := stream.Recv()
  if err == io.EOF {
   break
  }
  if err != nil {
   log.Fatalf("stream get from server err: %v", err)
  }
  log.Printf("stream get from server,code:%v,value:%v", resp.GetCode(), resp.Value)
  time.Sleep(1 * time.Second)
 }
 // 6.關(guān)閉流
 err = stream.CloseSend()
 if err != nil {
  log.Fatalf("close stream err:%v", err)
 }
}

客戶端代碼，在 for() 循環(huán)里面向服務端發(fā)送了 5 次消息，并接收服務端返回的數(shù)據(jù)，5次數(shù)據(jù)交互之后調(diào)用 CloseSend() 關(guān)閉流。

運行客戶端：

go run client.go

客戶端輸出：

stream get from server,code:1,value:This is the 1 message
stream get from server,code:2,value:This is the 2 message
stream get from server,code:3,value:This is the 3 message
stream get from server,code:4,value:This is the 4 message
stream get from server,code:5,value:This is the 5 message

服務端輸出：

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

觀察仔細的同學會注意到，客戶端和服務端是交替輸出的。

總結(jié)

這篇文章我們簡單介紹了 gRPC 的雙向流式 RPC，支持通信雙方同時多次發(fā)送或接收數(shù)據(jù)。