Go gRPC 系列三：流式客戶端和服務端

時間  2019-11-06
標籤 git github golang bash 服務器 tcp ide google spa 欄目 Git 简体版
原文   原文鏈接

前言

你們好，我是煎魚，本章節將介紹 gRPC 的流式，分爲三種類型：git

  • Server-side streaming RPC：服務器端流式 RPC
  • Client-side streaming RPC：客戶端流式 RPC
  • Bidirectional streaming RPC：雙向流式 RPC

流

任何技術，由於有痛點，因此纔有了存在的必要性。若是您想要了解 gRPC 的流式調用，請繼續github

圖

image

gRPC Streaming 是基於 HTTP/2 的，後續章節再進行詳細講解golang

爲何不用 Simple RPC

流式爲何要存在呢，是 Simple RPC 有什麼問題嗎？經過模擬業務場景，可得知在使用 Simple RPC 時，有以下問題：bash

  • 數據包過大形成的瞬時壓力
  • 接收數據包時，須要全部數據包都接受成功且正確後，纔可以回調響應，進行業務處理（沒法客戶端邊發送，服務端邊處理）

爲何用 Streaming RPC

  • 大規模數據包
  • 實時場景

模擬場景

天天早上 6 點，都有一批百萬級別的數據集要同從 A 同步到 B，在同步的時候，會作一系列操做（歸檔、數據分析、畫像、日誌等）。這一次性涉及的數據量確實大服務器

在同步完成後，也有人立刻會去查閱數據，爲了新的一天籌備。也符合實時性。tcp

二者相較下，這個場景下更適合使用 Streaming RPCide

gRPC

在講解具體的 gRPC 流式代碼時，會着重在第一節講解，由於三種模式實際上是不一樣的組合。但願你可以注重理解，觸類旁通，其實都是同樣的知識點 👍ui

目錄結構

$ tree go-grpc-example 
go-grpc-example
├── client
│   ├── simple_client
│   │   └── client.go
│   └── stream_client
│       └── client.go
├── proto
│   ├── search.proto
│   └── stream.proto
└── server
    ├── simple_server
    │   └── server.go
    └── stream_server
        └── server.go
複製代碼

增長 stream_server、stream_client 存放服務端和客戶端文件，proto/stream.proto 用於編寫 IDLgoogle

IDL

在 proto 文件夾下的 stream.proto 文件中，寫入以下內容：spa

syntax = "proto3";

package proto;

service StreamService {
    rpc List(StreamRequest) returns (stream StreamResponse) {};

    rpc Record(stream StreamRequest) returns (StreamResponse) {};

    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

基礎模板 + 空定義

Server

package main

import (
	"log"
	"net"

	"google.golang.org/grpc"

	pb "github.com/EDDYCJY/go-grpc-example/proto"
	
)

type StreamService struct{}

const (
	PORT = "9002"
)

func main() {
	server := grpc.NewServer()
	pb.RegisterStreamServiceServer(server, &StreamService{})

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

	server.Serve(lis)
}

func (s *StreamService) List(r *pb.StreamRequest, stream pb.StreamService_ListServer) error {
	return nil
}

func (s *StreamService) Record(stream pb.StreamService_RecordServer) error {
	return nil
}

func (s *StreamService) Route(stream pb.StreamService_RouteServer) error {
	return nil
}
複製代碼

寫代碼前，建議先將 gRPC Server 的基礎模板和接口給空定義出來。如有不清楚可參見上一章節的知識點

Client

package main

import (
    "log"
    
	"google.golang.org/grpc"

	pb "github.com/EDDYCJY/go-grpc-example/proto"
)

const (
	PORT = "9002"
)

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: 2018}})
	if err != nil {
		log.Fatalf("printLists.err: %v", err)
	}

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

	err = printRoute(client, &pb.StreamRequest{Pt: &pb.StreamPoint{Name: "gRPC Stream Client: Route", Value: 2018}})
	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、Server-side streaming RPC：服務器端流式 RPC

服務器端流式 RPC，顯然是單向流，並代指 Server 爲 Stream 而 Client 爲普通 RPC 請求

簡單來說就是客戶端發起一次普通的 RPC 請求，服務端經過流式響應屢次發送數據集，客戶端 Recv 接收數據集。大體如圖：

image

Server

func (s *StreamService) List(r *pb.StreamRequest, stream pb.StreamService_ListServer) error {
	for n := 0; n <= 6; n++ {
		err := stream.Send(&pb.StreamResponse{
			Pt: &pb.StreamPoint{
				Name:  r.Pt.Name,
				Value: r.Pt.Value + int32(n),
			},
		})
		if err != nil {
			return err
		}
	}

	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），若超出則提示錯誤
  • 寫入給流的數據集

Client

func printLists(client pb.StreamServiceClient, r *pb.StreamRequest) error {
	stream, err := client.List(context.Background(), r)
	if err != nil {
		return err
	}

	for {
		resp, err := stream.Recv()
		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

同時須要注意，默認的 MaxReceiveMessageSize 值爲 1024 * 1024 * 4，建議不要超出

驗證

運行 stream_server/server.go：

$ go run server.go
複製代碼

運行 stream_client/client.go：

$ go run client.go 
2018/09/24 16:18:25 resp: pj.name: gRPC Stream Client: List, pt.value: 2018
2018/09/24 16:18:25 resp: pj.name: gRPC Stream Client: List, pt.value: 2019
2018/09/24 16:18:25 resp: pj.name: gRPC Stream Client: List, pt.value: 2020
2018/09/24 16:18:25 resp: pj.name: gRPC Stream Client: List, pt.value: 2021
2018/09/24 16:18:25 resp: pj.name: gRPC Stream Client: List, pt.value: 2022
2018/09/24 16:18:25 resp: pj.name: gRPC Stream Client: List, pt.value: 2023
2018/09/24 16:18:25 resp: pj.name: gRPC Stream Client: List, pt.value: 2024
複製代碼

2、Client-side streaming RPC：客戶端流式 RPC

客戶端流式 RPC，單向流，客戶端經過流式發起屢次 RPC 請求給服務端，服務端發起一次響應給客戶端，大體如圖：

image

Server

func (s *StreamService) Record(stream pb.StreamService_RecordServer) error {
	for {
		r, err := stream.Recv()
		if err == io.EOF {
			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)
	}

	return nil
}
複製代碼

多了一個從未見過的方法 stream.SendAndClose，它是作什麼用的呢？

在這段程序中，咱們對每個 Recv 都進行了處理，當發現 io.EOF (流關閉) 後，須要將最終的響應結果發送給客戶端，同時關閉正在另一側等待的 Recv

Client

func printRecord(client pb.StreamServiceClient, r *pb.StreamRequest) error {
	stream, err := client.Record(context.Background())
	if err != nil {
		return err
	}

	for n := 0; n < 6; n++ {
		err := stream.Send(r)
		if err != nil {
			return err
		}
	}

	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 是配套使用的流方法，相信聰明的你已經秒懂它的做用了

驗證

重啓 stream_server/server.go，再次運行 stream_client/client.go：

stream_client：
$ go run client.go
2018/09/24 16:23:03 resp: pj.name: gRPC Stream Server: Record, pt.value: 1
複製代碼
stream_server：
$ go run server.go
2018/09/24 16:23:03 stream.Recv pt.name: gRPC Stream Client: Record, pt.value: 2018
2018/09/24 16:23:03 stream.Recv pt.name: gRPC Stream Client: Record, pt.value: 2018
2018/09/24 16:23:03 stream.Recv pt.name: gRPC Stream Client: Record, pt.value: 2018
2018/09/24 16:23:03 stream.Recv pt.name: gRPC Stream Client: Record, pt.value: 2018
2018/09/24 16:23:03 stream.Recv pt.name: gRPC Stream Client: Record, pt.value: 2018
2018/09/24 16:23:03 stream.Recv pt.name: gRPC Stream Client: Record, pt.value: 2018
複製代碼

3、Bidirectional streaming RPC：雙向流式 RPC

雙向流式 RPC，顧名思義是雙向流。由客戶端以流式的方式發起請求，服務端一樣以流式的方式響應請求

首個請求必定是 Client 發起，但具體交互方式（誰先誰後、一次發多少、響應多少、何時關閉）根據程序編寫的方式來肯定（能夠結合協程）

假設該雙向流是按順序發送的話，大體如圖：

image

仍是要強調，雙向流變化很大，因程序編寫的不一樣而不一樣。雙向流圖示沒法適用不一樣的場景

Server

func (s *StreamService) Route(stream pb.StreamService_RouteServer) error {
	n := 0
	for {
		err := stream.Send(&pb.StreamResponse{
			Pt: &pb.StreamPoint{
				Name:  "gPRC Stream Client: Route",
				Value: int32(n),
			},
		})
		if err != nil {
			return err
		}

		r, err := stream.Recv()
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return err
		}

		n++

		log.Printf("stream.Recv pt.name: %s, pt.value: %d", r.Pt.Name, r.Pt.Value)
	}

	return nil
}
複製代碼

Client

func printRoute(client pb.StreamServiceClient, r *pb.StreamRequest) error {
	stream, err := client.Route(context.Background())
	if err != nil {
		return err
	}

	for n := 0; n <= 6; n++ {
		err = stream.Send(r)
		if err != nil {
			return err
		}

		resp, err := stream.Recv()
		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)
	}

	stream.CloseSend()

	return nil
}
複製代碼

驗證

重啓 stream_server/server.go，再次運行 stream_client/client.go：

stream_server
$ go run server.go
2018/09/24 16:29:43 stream.Recv pt.name: gRPC Stream Client: Route, pt.value: 2018
2018/09/24 16:29:43 stream.Recv pt.name: gRPC Stream Client: Route, pt.value: 2018
2018/09/24 16:29:43 stream.Recv pt.name: gRPC Stream Client: Route, pt.value: 2018
2018/09/24 16:29:43 stream.Recv pt.name: gRPC Stream Client: Route, pt.value: 2018
2018/09/24 16:29:43 stream.Recv pt.name: gRPC Stream Client: Route, pt.value: 2018
2018/09/24 16:29:43 stream.Recv pt.name: gRPC Stream Client: Route, pt.value: 2018
複製代碼
stream_client
$ go run client.go
2018/09/24 16:29:43 resp: pj.name: gPRC Stream Client: Route, pt.value: 0
2018/09/24 16:29:43 resp: pj.name: gPRC Stream Client: Route, pt.value: 1
2018/09/24 16:29:43 resp: pj.name: gPRC Stream Client: Route, pt.value: 2
2018/09/24 16:29:43 resp: pj.name: gPRC Stream Client: Route, pt.value: 3
2018/09/24 16:29:43 resp: pj.name: gPRC Stream Client: Route, pt.value: 4
2018/09/24 16:29:43 resp: pj.name: gPRC Stream Client: Route, pt.value: 5
2018/09/24 16:29:43 resp: pj.name: gPRC Stream Client: Route, pt.value: 6
複製代碼

總結

在本文共介紹了三類流的交互方式，能夠根據實際的業務場景去選擇合適的方式。會事半功倍哦。

？

若是有任何疑問或錯誤，歡迎在 issues 進行提問或給予修正意見，若是喜歡或對你有所幫助，歡迎 Star，對做者是一種鼓勵和推動。

個人公衆號

image

參考

本系列示例代碼

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程