Go語言Grpc?Stream的實現(xiàn)

更新時間：2022年06月20日 09:40:16 作者：范閑

本文主要介紹了Go語言Grpc?Stream的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

Stream Grpc

在我們單次投遞的數(shù)據(jù)量很大的時候，比如傳輸一個二進制文件的時候，數(shù)據(jù)包過大，會造成瞬時傳輸壓力。或者接收方接收到數(shù)據(jù)后，需要對數(shù)據(jù)做一系列的處理工作，

比如：數(shù)據(jù)過濾 -> 數(shù)據(jù)格式轉(zhuǎn)換 -> 數(shù)據(jù)求和，這種場景非常適合使用stream grpc，

Stream Grpc演示

syntax = "proto3";

package book_stream;

option go_package = "/book_stream";

service HelloStreamService {
  rpc BookListStream(BookListStreamRequest) returns (stream BookListStreamResponse){};
  rpc CreateBookStream(stream CreateBookStreamRequest) returns (CreateBookStreamResponse){}
  rpc FindBookByIdStream(stream FindBookByIdStreamRequest) returns (stream FindBookByIdStreamResponse){}
}

message BookListStreamRequest{
}

message BookListStreamResponse{
  BookPoint book = 1;
}

message CreateBookStreamRequest{
  BookPoint book = 1;
}

message CreateBookStreamResponse{
  repeated BookIdPoint idx = 1;
}

message FindBookByIdStreamRequest{
  BookIdPoint idx = 1;
}
message FindBookByIdStreamResponse{
  BookPoint book = 1;
}

message BookIdPoint{
  uint64 idx = 1;
}

message BookPoint{
  uint64 idx = 1;
  string name = 2;
  float price = 3;
  string author = 4;
}

運行protoc --go_out=plugins=grpc:. *.proto生成腳手架文件

BookListStream服務(wù)端流式RPC
CreateBookStream客戶端流式RPC
FindBookByIdStream雙向流式RPC

注意，這里只是用作方便演示使用，演示方法都不是線程安全的

服務(wù)端server

var port = 8888

func main() {
   server := grpc.NewServer()
   book_stream.RegisterHelloStreamServiceServer(server, new(HelloStreamServiceImpl))
   lis, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
   if err != nil {
      panic(err)
   }
   if err := server.Serve(lis); err != nil {
      panic(err)
   }
}

客戶端

func main() {
   var port = 8888
   conn, err := grpc.Dial(fmt.Sprintf(":%d", port), grpc.WithInsecure())
   if err != nil {
      panic(err)
   }
   defer conn.Close()
   client := book_stream.NewHelloStreamServiceClient(conn)

   ctx := context.Background()
   if err := createBookStream(ctx, client); err != nil {
      panic(err)
   }
   if err := printBookList(ctx, client); err != nil {
      panic(err)
   }
   if err := getBookListById(ctx, client); err != nil {
      panic(err)
   }
}

BookListStream

服務(wù)器端流式 RPC，顯然是單向流，并代指 Server 為 Stream 而 Client 為普通 RPC 請求

簡單來講就是客戶端發(fā)起一次普通的 RPC 請求，服務(wù)端通過流式響應(yīng)多次發(fā)送數(shù)據(jù)集，客戶端 Recv 接收數(shù)據(jù)集。

server端實現(xiàn)

var bookStore = map[uint64]book_stream.BookPoint{
   1: {
      Idx:    1,
      Author: "程子",
      Price:  9.9,
      Name:   "游戲思維",
   },
   2: {
      Idx:    2,
      Author: "丁銳",
      Price:  9.9,
      Name:   "活出必要的鋒芒",
   },
}


type HelloStreamServiceImpl struct{}

func (HelloStreamServiceImpl) BookListStream(_ *book_stream.BookListStreamRequest, streamServer book_stream.HelloStreamService_BookListStreamServer) error {
   for idx, bookPoint := range bookStore {
      err := streamServer.Send(&book_stream.BookListStreamResponse{Book: &book_stream.BookPoint{
         Idx:    idx,
         Name:   bookPoint.Name,
         Price:  bookPoint.GetPrice(),
         Author: bookPoint.Author,
      }})
      if err != nil {
         return err
      }
   }
   return nil
}

客戶端實現(xiàn)

func printBookList(ctx context.Context, client book_stream.HelloStreamServiceClient) error {
   req := &book_stream.BookListStreamRequest{}
   listStream, err := client.BookListStream(ctx, req)
   if err != nil {
      return err
   }
   for true {
      resp, err := listStream.Recv()
      if err != nil {
         if err == io.EOF {
            return nil
         }
         return err
      }
      fmt.Printf("%v\n", *resp.Book)
   }
   return nil
}

CreateBookStream

客戶端流式 RPC，單向流，客戶端通過流式發(fā)起多次 RPC 請求給服務(wù)端，服務(wù)端發(fā)起一次響應(yīng)給客戶端

server端實現(xiàn)

func (HelloStreamServiceImpl) CreateBookStream(server book_stream.HelloStreamService_CreateBookStreamServer) error {
   var resList []*book_stream.BookIdPoint
   for {
      resp, err := server.Recv()
      if err == io.EOF {
         return server.SendAndClose(&book_stream.CreateBookStreamResponse{Idx: resList})
      }
      if err != nil {
         return err
      }
      bookStore[resp.Book.Idx] = *resp.Book
      resList = append(resList, &book_stream.BookIdPoint{Idx: resp.Book.Idx})
   }
}

客戶端實現(xiàn)

var newBookStore = map[uint64]book_stream.BookPoint{
   3: {
      Idx:    3,
      Author: "程子1",
      Price:  9.9,
      Name:   "游戲思維1",
   },
   4: {
      Idx:    4,
      Author: "丁銳1",
      Price:  9.9,
      Name:   "活出必要的鋒芒1",
   },
}

func createBookStream(ctx context.Context, client book_stream.HelloStreamServiceClient) error {
   stream, err := client.CreateBookStream(ctx)
   if err != nil {
      return err
   }
   for _, bookPoint := range newBookStore {
      if err := stream.Send(&book_stream.CreateBookStreamRequest{
         Book: &bookPoint,
      }); err != nil {
         return err
      }
   }
   recv, err := stream.CloseAndRecv()
   if err != nil {
      return err
   }
   fmt.Println(recv.Idx)
   return nil
}

stream.SendAndClose，它是做什么用的呢？

在這段程序中，我們對每一個 Recv 都進行了處理，當(dāng)發(fā)現(xiàn) io.EOF (流關(guān)閉) 后，需要將最終的響應(yīng)結(jié)果發(fā)送給客戶端，同時關(guān)閉正在另外一側(cè)等待的 Recv

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

FindBookByIdStream

服務(wù)端實現(xiàn)

func (HelloStreamServiceImpl) FindBookByIdStream(streamServer book_stream.HelloStreamService_FindBookByIdStreamServer) error {
   for {
      resp, err := streamServer.Recv()
      if err == io.EOF {
         return nil
      }
      if err != nil {
         return err
      }
      if book, ok := bookStore[resp.Idx.Idx]; ok {
         if err := streamServer.Send(&book_stream.FindBookByIdStreamResponse{Book: &book}); err != nil {
            return err
         }
      }
   }
}

客戶端實現(xiàn)

func getBookListById(ctx context.Context, client book_stream.HelloStreamServiceClient) error {
   stream, err := client.FindBookByIdStream(ctx)
   if err != nil {
      return err
   }
   var findList = []uint64{1, 2}
   for _, idx := range findList {
      err := stream.Send(&book_stream.FindBookByIdStreamRequest{Idx: &book_stream.BookIdPoint{Idx: idx}})
      if err != nil {
         return err
      }
      recv, err := stream.Recv()
      if err != nil {
         return err
      }
      fmt.Printf("%v\n", recv.Book)
   }
   if err := stream.CloseSend(); err != nil {
      return err
   }
   return nil
}

到此這篇關(guān)于Go語言Grpc Stream的實現(xiàn)的文章就介紹到這了,更多相關(guān)Go語言Grpc Stream 內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: