gRPC學習之四：實戰四類服務方法

2021 年 8 月 17 日
筆記

歡迎訪問我的GitHub

//github.com/zq2599/blog_demos

內容：所有原創文章分類匯總及配套源碼，涉及Java、Docker、Kubernetes、DevOPS等；

gRPC學習系列文章鏈接

本篇概覽

本文《gRPC學習》系列的第四篇，前文咱們體驗了最簡單的gRPC開發，編寫客戶端調用服務端，但這只是最簡單的一種，在解決實際問題時是遠遠不夠的；
實際上，gRPC允許你定義以下四類服務方法（以下描述來自//doc.oschina.net/grpc）：

單項 RPC，即客戶端發送一個請求給服務端，從服務端獲取一個應答，就像一次普通的函數調用（前一篇文章就是此類）；
服務端流式 RPC，即客戶端發送一個請求給服務端，可獲取一個數據流用來讀取一系列消息。客戶端從返回的數據流里一直讀取直到沒有更多消息為止；
客戶端流式 RPC，即客戶端用提供的一個數據流寫入並發送一系列消息給服務端。一旦客戶端完成消息寫入，就等待服務端讀取這些消息並返回應答；
雙向流式 RPC，即兩邊都可以分別通過一個讀寫數據流來發送一系列消息。這兩個數據流操作是相互獨立的，所以客戶端和服務端能按其希望的任意順序讀寫，例如：服務端可以在寫應答前等待所有的客戶端消息，或者它可以先讀一個消息再寫一個消息，或者是讀寫相結合的其他方式。每個數據流里消息的順序會被保持。

本篇的內容，就是編碼實現上述四類服務方法，並編寫客戶端程式碼調用，整個開發流程如下圖所示：

源碼下載

本篇實戰中的源碼可在GitHub下載到，地址和鏈接資訊如下表所示(//github.com/zq2599/blog_demos)：

名稱	鏈接	備註
項目主頁	//github.com/zq2599/blog_demos	該項目在GitHub上的主頁
git倉庫地址(https)	//github.com/zq2599/blog_demos.git	該項目源碼的倉庫地址，https協議
git倉庫地址(ssh)	[email protected]:zq2599/blog_demos.git	該項目源碼的倉庫地址，ssh協議

這個git項目中有多個文件夾，本章的應用在go-source文件夾下，如下圖紅框所示：

go-source裡面有多個子文件夾，本篇的源碼在grpcstream中，如下圖紅框：

提前說明文件和目錄

本次實戰在$GOPATH/src目錄下新增文件夾grpcstream，裡面總共有以下內容：

[golang@centos7 src]$ tree grpcstream/
grpcstream/
├── client
│   └── client.go
├── grpcstream.pb.go
├── grpcstream.proto
└── server
    └── server.go

準備工作完成，接下來正式開始開發；

編寫proto文件

proto文件用來描述遠程服務相關的資訊，如方法簽名、數據結構等，本篇的proto文件名為grpcstream.proto，位置是$GOPATH/src/grpcstream，內容如下(稍後會指出幾處要注意的地方)：

// 協議類型
syntax = "proto3";

// 包名
package grpcstream;

// 服務端請求的數據結構
message SingleRequest {
  int32 id = 1;
}

// 服務端響應的數據結構
message SingleResponse {
  int32 id = 1;
  string name = 2;
}

// 定義的服務名
service IGrpcStremService {
  // 單項RPC ：單個請求，單個響應
  rpc SingleReqSingleResp (SingleRequest) returns (SingleResponse);

  // 服務端流式 ：單個請求，集合響應
  rpc SingleReqMultiResp (SingleRequest) returns (stream SingleResponse);

  // 客戶端流式 ：集合請求，單個響應
  rpc MultiReqSingleResp (stream SingleRequest) returns (SingleResponse);

  // 雙向流式 ：集合請求，集合響應
  rpc MultiReqMultiResp (stream SingleRequest) returns (stream SingleResponse);
}

這個grpcstream.proto文件有以下幾處要注意的地方：

方法SingleReqSingleResp非常簡單，和上一篇文章中的demo一樣，入參是一個數據結構，服務端返回的也是一個數據結構；
方法SingleReqSingleResp是服務端流式類型，特徵是返回值用stream修飾；
方法MultiReqSingleResp是客戶端流式類型，特徵是入參用stream修飾；
方法MultiReqMultiResp是雙向類型，特徵是入參和返回值都用stream修飾；

似乎有規律可循：客戶端如果想和服務端建立通道傳輸持續的數據，就在通道位置用stream修飾，一共有兩個位置，第一個是進入服務端的入參，第二個是從服務端出來的返回值；

根據proto生成go源碼

在grpcstream.proto所在的目錄，執行以下命令：

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

如果grpcstream.proto沒有語法錯誤，會在當前目錄生成文件grpcstream.pb.go，這裡面是工具protoc-gen-go自動生成的程式碼，裡面生成的程式碼在開發服務端和客戶端時都會用到；
對服務端來說，grpcstream.pb.go中最重要的是IGrpcStremServiceServer介面，服務端需要實現該介面所有的方法作為業務邏輯，介面定義如下：

type IGrpcStremServiceServer interface {
	// 單項流式 ：單個請求，單個響應
	SingleReqSingleResp(context.Context, *SingleRequest) (*SingleResponse, error)
	// 服務端流式 ：單個請求，集合響應
	SingleReqMultiResp(*SingleRequest, IGrpcStremService_SingleReqMultiRespServer) error
	// 客戶端流式 ：集合請求，單個響應
	MultiReqSingleResp(IGrpcStremService_MultiReqSingleRespServer) error
	// 雙向流式 ：集合請求，集合響應
	MultiReqMultiResp(IGrpcStremService_MultiReqMultiRespServer) error
}

對客戶端來說，grpcstream.pb.go中最重要的是IGrpcStremServiceClient介面，如下所示，這意味這客戶端可以發起哪些遠程調用：

type IGrpcStremServiceClient interface {
	// 單項流式 ：單個請求，單個響應
	SingleReqSingleResp(ctx context.Context, in *SingleRequest, opts ...grpc.CallOption) (*SingleResponse, error)
	// 服務端流式 ：單個請求，集合響應
	SingleReqMultiResp(ctx context.Context, in *SingleRequest, opts ...grpc.CallOption) (IGrpcStremService_SingleReqMultiRespClient, error)
	// 客戶端流式 ：集合請求，單個響應
	MultiReqSingleResp(ctx context.Context, opts ...grpc.CallOption) (IGrpcStremService_MultiReqSingleRespClient, error)
	// 雙向流式 ：集合請求，集合響應
	MultiReqMultiResp(ctx context.Context, opts ...grpc.CallOption) (IGrpcStremService_MultiReqMultiRespClient, error)
}

編寫服務端程式碼server.go並啟動

在$GOPATH/src/grpcstream目錄下新建文件夾server，在此文件夾下新建server.go，內容如下(稍後會指出幾處要注意的地方)：

package main

import (
	"context"
	"google.golang.org/grpc"
	pb "grpcstream"
	"io"
	"log"
	"net"
	"strconv"
)

// 常量：監聽埠
const (
	port = ":50051"
)

// 定義結構體，在調用註冊api的時候作為入參，
// 該結構體會帶上proto中定義的方法，裡面是業務程式碼
// 這樣遠程調用時就執行了業務程式碼了
type server struct {
	// pb.go中自動生成的，是個空結構體
	pb.UnimplementedIGrpcStremServiceServer
}

// 單項流式 ：單個請求，單個響應
func (s *server) SingleReqSingleResp(ctx context.Context, req *pb.SingleRequest) (*pb.SingleResponse, error) {
	id := req.GetId()

	// 列印請求參數
	log.Println("1. 收到請求:", id)
	// 實例化結構體SingleResponse，作為返回值
	return &pb.SingleResponse{Id: id, Name: "1. name-" + strconv.Itoa(int(id))}, nil
}

// 服務端流式 ：單個請求，集合響應
func (s *server) SingleReqMultiResp(req *pb.SingleRequest, stream pb.IGrpcStremService_SingleReqMultiRespServer) error {
	// 取得請求參數
	id := req.GetId()

	// 列印請求參數
	log.Println("2. 收到請求:", id)

	// 返回多條記錄
	for i := 0; i < 10; i++ {
		stream.Send(&pb.SingleResponse{Id: int32(i), Name: "2. name-" + strconv.Itoa(i)})
	}

	return nil
}

// 客戶端流式 ：集合請求，單個響應
func (s *server) MultiReqSingleResp(reqStream pb.IGrpcStremService_MultiReqSingleRespServer) error {
	var addVal int32 = 0

	// 在for循環中接收流式請求
	for {
		// 一次接受一條記錄
		singleRequest, err := reqStream.Recv()

		// 不等於io.EOF表示這是條有效記錄
		if err == io.EOF {
			log.Println("3. 客戶端發送完畢")
			break
		} else if err != nil {
			log.Fatalln("3. 接收時發生異常", err)
			break
		} else {
			log.Println("3. 收到請求:", singleRequest.GetId())
			// 收完之後，執行SendAndClose返回數據並結束本次調用
			addVal += singleRequest.GetId()
		}
	}

	return reqStream.SendAndClose(&pb.SingleResponse{Id: addVal, Name: "3. name-" + strconv.Itoa(int(addVal))})
}

// 雙向流式 ：集合請求，集合響應
func (s *server) MultiReqMultiResp(reqStream pb.IGrpcStremService_MultiReqMultiRespServer) error {
	// 簡單處理，對於收到的每一條記錄都返回一個響應
	for {
		singleRequest, err := reqStream.Recv()

		// 不等於io.EOS表示這是條有效記錄
		if err == io.EOF {
			log.Println("4. 接收完畢")
			return nil
		} else if err != nil {
			log.Fatalln("4. 接收時發生異常", err)
			return err
		} else {
			log.Println("4. 接收到數據", singleRequest.GetId())

			id := singleRequest.GetId()

			if sendErr := reqStream.Send(&pb.SingleResponse{Id: id, Name: "4. name-" + strconv.Itoa(int(id))}); sendErr != nil {
				log.Println("4. 返回數據異常數據", sendErr)
				return sendErr
			}
		}
	}
}

func main() {
	// 要監聽的協議和埠
	lis, err := net.Listen("tcp", port)
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}

	// 實例化gRPC server結構體
	s := grpc.NewServer()

	// 服務註冊
	pb.RegisterIGrpcStremServiceServer(s, &server{})

	log.Println("開始監聽，等待遠程調用...")

	if err := s.Serve(lis); err != nil {
		log.Fatalf("failed to serve: %v", err)
	}
}

這個server.go文件有以下幾處要注意的地方：

SingleReqMultiResp方法的作用是收到客戶端一個請求參數，然後向客戶端發送多個響應，可見多次調用stream.Send方法即可多次發送數據到客戶端；
MultiReqSingleResp方法可以從客戶端收到多條數據，可見是在for循環中重複調用reqStream.Recv()方法，直到收到客戶端的io.EOF為止，這就要就客戶端在發送完數據後再給一個io.EOF過來，稍後的客戶端程式碼會展示如何做；
MultiReqMultiResp方法持續接受客戶端數據，並且持續發送數據給客戶端，一定要把順序問題考慮清楚，否則會陷入異常(例如一方死循環)，我這裡的邏輯是收到客戶端的io.EOF為止，這就要就客戶端在發送完數據後再給一個io.EOF過來，如果客戶端也在用for循環一直等數據，那就是雙方都在等數據了，無法終止程式，稍後的客戶端程式碼會展示如何做；

在server.go所在目錄執行go run server.go，控制台提示如下：

[golang@centos7 server]$ go run server.go 
2020/12/13 21:29:19 開始監聽，等待遠程調用...

此時gRPC的服務端已經啟動，可以響應遠程調用，接下來開發客戶端程式碼；

編寫客戶端程式碼client.go

再打開一個控制台；
在$GOPATH/src/grpcstream目錄下新建文件夾client，在此文件夾下新建client.go，內容如下(稍後會指出幾處要注意的地方)：

package main

import (
	"context"
	"google.golang.org/grpc"
	"io"
	"log"
	"time"

	pb "grpcstream"
)

const (
	address     = "localhost:50051"
	defaultId = "666"
)

func main() {
	// 遠程連接服務端
	conn, err := grpc.Dial(address, grpc.WithInsecure(), grpc.WithBlock())
	if err != nil {
		log.Fatalf("did not connect: %v", err)
	}

	// main方法執行完畢後關閉遠程連接
	defer conn.Close()

	// 實例化數據結構
	client := pb.NewIGrpcStremServiceClient(conn)

	// 超時設置
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)

	defer cancel()

	log.Println("測試單一請求應答，一對一")
	singleReqSingleResp(ctx, client)

	log.Println("測試服務端流式應答，一對多")
	singleReqMultiResp(ctx, client)

	log.Println("測試客戶端流式請求，多對一")
	multiReqSingleResp(ctx, client)

	log.Println("測試雙向流式請求應答，多對多")
	multiReqMultiResp(ctx, client)

	log.Println("測試完成")
}


func singleReqSingleResp(ctx context.Context, client pb.IGrpcStremServiceClient) error {
	// 遠程調用
	r, err := client.SingleReqSingleResp(ctx, &pb.SingleRequest{Id: 101})

	if err != nil {
		log.Fatalf("1. 遠程調用異常 : %v", err)
		return err
	}

	// 將服務端的返回資訊列印出來
	log.Printf("response, id : %d, name : %s", r.GetId(), r.GetName())

	return nil
}


func singleReqMultiResp(ctx context.Context, client pb.IGrpcStremServiceClient) error {
	// 遠程調用
	recvStream, err := client.SingleReqMultiResp(ctx, &pb.SingleRequest{Id: 201})

	if err != nil {
		log.Fatalf("2. 遠程調用異常 : %v", err)
		return err
	}

	for {
		singleResponse, err := recvStream.Recv()
		if err == io.EOF {
			log.Printf("2. 獲取數據完畢")
			break
		}

		log.Printf("2. 收到服務端響應, id : %d, name : %s", singleResponse.GetId(), singleResponse.GetName())
	}

	return nil
}

func multiReqSingleResp(ctx context.Context, client pb.IGrpcStremServiceClient) error {
	// 遠程調用
	sendStream, err := client.MultiReqSingleResp(ctx)

	if err != nil {
		log.Fatalf("3. 遠程調用異常 : %v", err)
		return err
	}

	// 發送多條記錄到服務端
	for i:=0; i<10; i++ {
		if err = sendStream.Send(&pb.SingleRequest{Id: int32(300+i)}); err!=nil {
			log.Fatalf("3. 通過流發送數據異常 : %v", err)
			return err
		}
	}

	singleResponse, err := sendStream.CloseAndRecv()

	if err != nil {
		log.Fatalf("3. 服務端響應異常 : %v", err)
		return err
	}

	// 將服務端的返回資訊列印出來
	log.Printf("response, id : %d, name : %s", singleResponse.GetId(), singleResponse.GetName())

	return nil
}

func multiReqMultiResp(ctx context.Context, client pb.IGrpcStremServiceClient) error {
	// 遠程調用
	intOutStream, err := client.MultiReqMultiResp(ctx)

	if err != nil {
		log.Fatalf("4. 遠程調用異常 : %v", err)
		return err
	}

	// 發送多條記錄到服務端
	for i:=0; i<10; i++ {
		if err = intOutStream.Send(&pb.SingleRequest{Id: int32(400+i)}); err!=nil {
			log.Fatalf("4. 通過流發送數據異常 : %v", err)
			return err
		}
	}

	// 服務端一直在接收，直到收到io.EOF為止
	// 因此，這裡必須發送io.EOF到服務端，讓服務端知道發送已經結束(很重要)
	intOutStream.CloseSend()

	// 接收服務端發來的數據
	for {
		singleResponse, err := intOutStream.Recv()
		if err == io.EOF {
			log.Printf("4. 獲取數據完畢")
			break
		} else if err != nil {
			log.Fatalf("4. 接收服務端數據異常 : %v", err)
			break
		}

		log.Printf("4. 收到服務端響應, id : %d, name : %s", singleResponse.GetId(), singleResponse.GetName())
	}

	return nil
}

這個client.go文件有以下幾處要注意的地方：

singleReqMultiResp方法會接收服務端的多條記錄，在for循環中調用recvStream.Recv即可收到所有數據；
multiReqSingleResp方法會向服務端發送多條數據，由於服務端在等待io.EOF作為結束標誌，因此調用sendStream.CloseAndRecv即可發送io.EOF，並得到服務端的返回值；
multiReqMultiResp方法在持續向服務端發送數據，並且也在持續獲取服務端發來的數據，在發送數據完成後，必須調用intOutStream.CloseSend方法，即可發送io.EOF，讓服務端不再接收數據，避免前面提到的死循環；
在main方法中，依次發起四類服務方法的調用；

執行客戶端

編碼完成後，在client.go所在目錄執行go run client.go，會立即向服務端發起遠程調用，控制台提示如下，可見四類服務方法測試全部成功，響應的數據都符合預期：

[golang@centos7 client]$ go run client.go 
2020/12/13 21:39:35 測試單一請求應答，一對一
2020/12/13 21:39:35 response, id : 101, name : 1. name-101
2020/12/13 21:39:35 測試服務端流式應答，一對多
2020/12/13 21:39:35 2. 收到服務端響應, id : 0, name : 2. name-0
2020/12/13 21:39:35 2. 收到服務端響應, id : 1, name : 2. name-1
2020/12/13 21:39:35 2. 收到服務端響應, id : 2, name : 2. name-2
2020/12/13 21:39:35 2. 收到服務端響應, id : 3, name : 2. name-3
2020/12/13 21:39:35 2. 收到服務端響應, id : 4, name : 2. name-4
2020/12/13 21:39:35 2. 收到服務端響應, id : 5, name : 2. name-5
2020/12/13 21:39:35 2. 收到服務端響應, id : 6, name : 2. name-6
2020/12/13 21:39:35 2. 收到服務端響應, id : 7, name : 2. name-7
2020/12/13 21:39:35 2. 收到服務端響應, id : 8, name : 2. name-8
2020/12/13 21:39:35 2. 收到服務端響應, id : 9, name : 2. name-9
2020/12/13 21:39:35 2. 獲取數據完畢
2020/12/13 21:39:35 測試客戶端流式請求，多對一
2020/12/13 21:39:35 response, id : 3045, name : 3. name-3045
2020/12/13 21:39:35 測試雙向流式請求應答，多對多
2020/12/13 21:39:35 4. 收到服務端響應, id : 400, name : 4. name-400
2020/12/13 21:39:35 4. 收到服務端響應, id : 401, name : 4. name-401
2020/12/13 21:39:35 4. 收到服務端響應, id : 402, name : 4. name-402
2020/12/13 21:39:35 4. 收到服務端響應, id : 403, name : 4. name-403
2020/12/13 21:39:35 4. 收到服務端響應, id : 404, name : 4. name-404
2020/12/13 21:39:35 4. 收到服務端響應, id : 405, name : 4. name-405
2020/12/13 21:39:35 4. 收到服務端響應, id : 406, name : 4. name-406
2020/12/13 21:39:35 4. 收到服務端響應, id : 407, name : 4. name-407
2020/12/13 21:39:35 4. 收到服務端響應, id : 408, name : 4. name-408
2020/12/13 21:39:35 4. 收到服務端響應, id : 409, name : 4. name-409
2020/12/13 21:39:35 4. 獲取數據完畢
2020/12/13 21:39:35 測試完成

再去服務端的控制台看一下，通過日誌發現業務程式碼被執行，收到了遠程請求的參數：

[golang@centos7 server]$ go run server.go 
2020/12/13 21:29:19 開始監聽，等待遠程調用...
2020/12/13 21:39:35 1. 收到請求: 101
2020/12/13 21:39:35 2. 收到請求: 201
2020/12/13 21:39:35 3. 收到請求: 300
2020/12/13 21:39:35 3. 收到請求: 301
2020/12/13 21:39:35 3. 收到請求: 302
2020/12/13 21:39:35 3. 收到請求: 303
2020/12/13 21:39:35 3. 收到請求: 304
2020/12/13 21:39:35 3. 收到請求: 305
2020/12/13 21:39:35 3. 收到請求: 306
2020/12/13 21:39:35 3. 收到請求: 307
2020/12/13 21:39:35 3. 收到請求: 308
2020/12/13 21:39:35 3. 收到請求: 309
2020/12/13 21:39:35 3. 客戶端發送完畢
2020/12/13 21:39:35 4. 接收到數據 400
2020/12/13 21:39:35 4. 接收到數據 401
2020/12/13 21:39:35 4. 接收到數據 402
2020/12/13 21:39:35 4. 接收到數據 403
2020/12/13 21:39:35 4. 接收到數據 404
2020/12/13 21:39:35 4. 接收到數據 405
2020/12/13 21:39:35 4. 接收到數據 406
2020/12/13 21:39:35 4. 接收到數據 407
2020/12/13 21:39:35 4. 接收到數據 408
2020/12/13 21:39:35 4. 接收到數據 409
2020/12/13 21:39:35 4. 接收完畢

至此，gRPC的四類服務方法的服務端、客戶端開發咱們都嘗試過了，這四類方法已經可以覆蓋了大多數業務場景需求，希望本文能給您一些參考，接下來的文章會繼續學習gRPC豐富的功能；

你不孤單，欣宸原創一路相伴

歡迎關注公眾號：程式設計師欣宸

微信搜索「程式設計師欣宸」，我是欣宸，期待與您一同暢遊Java世界…
//github.com/zq2599/blog_demos

gRPC學習之四：實戰四類服務方法

歡迎訪問我的GitHub

gRPC學習系列文章鏈接

本篇概覽

源碼下載

提前說明文件和目錄

編寫proto文件

根據proto生成go源碼

編寫服務端程式碼server.go並啟動

編寫客戶端程式碼client.go

執行客戶端

你不孤單，欣宸原創一路相伴

歡迎關注公眾號：程式設計師欣宸

VirMach 便宜 VPS

QNews

gRPC學習之四：實戰四類服務方法

歡迎訪問我的GitHub

gRPC學習系列文章鏈接

本篇概覽

源碼下載

提前說明文件和目錄

編寫proto文件

根據proto生成go源碼

編寫服務端程式碼server.go並啟動

編寫客戶端程式碼client.go

執行客戶端

你不孤單，欣宸原創一路相伴

歡迎關注公眾號：程式設計師欣宸

分享此文：

Related Posts

Spring Security 架構簡介

《Head First 設計模式》：外觀模式

吳亦凡被批捕 官方定調公開發聲：徹底杯具！

一個免費APP 是如何成為網民的老婆生成器的？

VirMach 便宜 VPS

QNews

熱門搜尋

吳亦凡被批捕官方定調公開發聲：徹底杯具！