Go gRPC教程-客戶端流式RPC（四）

2020 年 4 月 14 日
筆記
Go, gRPC

前言

上一篇介紹了服務端流式RPC，客戶端發送請求到伺服器，拿到一個流去讀取返回的消息序列。客戶端讀取返回的流的數據。本篇將介紹客戶端流式RPC。

客戶端流式RPC：與服務端流式RPC相反，客戶端不斷的向服務端發送數據流，而在發送結束後，由服務端返回一個響應。

情景模擬：客戶端大量數據上傳到服務端。

新建proto文件

新建client_stream.proto文件

1.定義發送資訊

// 定義流式請求資訊
message StreamRequest{
    //流式請求參數
    string stream_data = 1;
}

2.定義接收資訊

// 定義響應資訊
message SimpleResponse{
    //響應碼
    int32 code = 1;
    //響應值
    string value = 2;
}

3.定義服務方法RouteList

客戶端流式rpc，只要在請求的參數前添加stream即可

service StreamClient{
    // 客戶端流式rpc，在請求的參數前添加stream
    rpc RouteList (stream StreamRequest) returns (SimpleResponse){};
}

4.編譯proto文件

進入client_stream.proto所在目錄，運行指令:

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

創建Server端

1.定義我們的服務，並實現RouteList方法

// SimpleService 定義我們的服務
type SimpleService struct{}
// RouteList 實現RouteList方法
func (s *SimpleService) RouteList(srv pb.StreamClient_RouteListServer) error {
	for {
		//從流中獲取消息
		res, err := srv.Recv()
		if err == io.EOF {
			//發送結果，並關閉
			return srv.SendAndClose(&pb.SimpleResponse{Value: "ok"})
		}
		if err != nil {
			return err
		}
		log.Println(res.StreamData)
	}
}

2.啟動gRPC伺服器

const (
	// Address 監聽地址
	Address string = ":8000"
	// Network 網路通訊協議
	Network string = "tcp"
)

func main() {
	// 監聽本地埠
	listener, err := net.Listen(Network, Address)
	if err != nil {
		log.Fatalf("net.Listen err: %v", err)
	}
	log.Println(Address + " net.Listing...")
	// 新建gRPC伺服器實例
	grpcServer := grpc.NewServer()
	// 在gRPC伺服器註冊我們的服務
	pb.RegisterStreamClientServer(grpcServer, &SimpleService{})

	//用伺服器 Serve() 方法以及我們的埠資訊區實現阻塞等待，直到進程被殺死或者 Stop() 被調用
	err = grpcServer.Serve(listener)
	if err != nil {
		log.Fatalf("grpcServer.Serve err: %v", err)
	}
}

運行服務端

go run server.go
:8000 net.Listing...

創建Client端

1.創建調用服務端RouteList方法

// routeList 調用服務端RouteList方法
func routeList() {
	//調用服務端RouteList方法，獲流
	stream, err := streamClient.RouteList(context.Background())
	if err != nil {
		log.Fatalf("Upload list err: %v", err)
	}
	for n := 0; n < 5; n++ {
		//向流中發送消息
		err := stream.Send(&pb.StreamRequest{StreamData: "stream client rpc " + strconv.Itoa(n)})
		if err != nil {
			log.Fatalf("stream request err: %v", err)
		}
	}
	//關閉流並獲取返回的消息
	res, err := stream.CloseAndRecv()
	if err != nil {
		log.Fatalf("RouteList get response err: %v", err)
	}
	log.Println(res)
}

2.啟動gRPC客戶端

// Address 連接地址
const Address string = ":8000"

var streamClient pb.StreamClientClient

func main() {
	// 連接伺服器
	conn, err := grpc.Dial(Address, grpc.WithInsecure())
	if err != nil {
		log.Fatalf("net.Connect err: %v", err)
	}
	defer conn.Close()

	// 建立gRPC連接
	streamClient = pb.NewStreamClientClient(conn)
	routeList()
}

運行客戶端

go run client.go
code:200 value:"hello grpc"
value:"ok"

服務端不斷從客戶端獲取到數據

stream client rpc 0
stream client rpc 1
stream client rpc 2
stream client rpc 3
stream client rpc 4

思考

服務端在沒有接受完消息時候能主動停止接收數據嗎（很少有這種場景）？

答案：可以的，但是客戶端程式碼需要注意EOF判斷

1.我們把服務端的RouteList方法實現稍微修改，當接收到一條數據後馬上調用SendAndClose()關閉stream.

// RouteList 實現RouteList方法
func (s *SimpleService) RouteList(srv pb.StreamClient_RouteListServer) error {
	for {
		//從流中獲取消息
		res, err := srv.Recv()
		if err == io.EOF {
			//發送結果，並關閉
			return srv.SendAndClose(&pb.SimpleResponse{Value: "ok"})
		}
		if err != nil {
			return err
		}
		log.Println(res.StreamData)
		return srv.SendAndClose(&pb.SimpleResponse{Value: "ok"})
	}
}

2.再把客戶端調用RouteList方法的實現稍作修改

// routeList 調用服務端RouteList方法
func routeList() {
	//調用服務端RouteList方法，獲流
	stream, err := streamClient.RouteList(context.Background())
	if err != nil {
		log.Fatalf("Upload list err: %v", err)
	}
	for n := 0; n < 5; n++ {
		//向流中發送消息
		err := stream.Send(&pb.StreamRequest{StreamData: "stream client rpc " + strconv.Itoa(n)})
		//發送也要檢測EOF，當服務端在消息沒接收完前主動調用SendAndClose()關閉stream，此時客戶端還執行Send()，則會返回EOF錯誤，所以這裡需要加上io.EOF判斷
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatalf("stream request err: %v", err)
		}
	}
	//關閉流並獲取返回的消息
	res, err := stream.CloseAndRecv()
	if err != nil {
		log.Fatalf("RouteList get response err: %v", err)
	}
	log.Println(res)
}

客戶端Send()需要檢測err是否為EOF，因為當服務端在消息沒接收完前主動調用SendAndClose()關閉stream，若此時客戶端繼續執行Send()，則會返回EOF錯誤。

總結

本篇介紹了客戶端流式RPC的簡單使用，下篇將介紹雙向流式RPC。

教程源碼地址：//github.com/Bingjian-Zhu/go-grpc-example
參考：gRPC官方文檔中文版

Tags: Go gRPC