Go?gRPC服務(wù)客戶端流式RPC教程

更新時(shí)間：2022年06月16日 08:52:24 作者：煙花易冷人憔悴

這篇文章主要為大家介紹了Go?gRPC服務(wù)客戶端流式RPC教程示例，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

前言

上一篇介紹了服務(wù)端流式RPC，客戶端發(fā)送請(qǐng)求到服務(wù)器，拿到一個(gè)流去讀取返回的消息序列。客戶端讀取返回的流的數(shù)據(jù)。本篇將介紹客戶端流式RPC。

客戶端流式RPC：與服務(wù)端流式RPC相反，客戶端不斷的向服務(wù)端發(fā)送數(shù)據(jù)流，而在發(fā)送結(jié)束后，由服務(wù)端返回一個(gè)響應(yīng)。

情景模擬：客戶端大量數(shù)據(jù)上傳到服務(wù)端

新建proto文件

新建client_stream.proto文件

1.定義發(fā)送信息

// 定義流式請(qǐng)求信息
message StreamRequest{
    //流式請(qǐng)求參數(shù)
    string stream_data = 1;
}

2.定義接收信息

// 定義響應(yīng)信息
message SimpleResponse{
    //響應(yīng)碼
    int32 code = 1;
    //響應(yīng)值
    string value = 2;
}

3.定義服務(wù)方法RouteList

客戶端流式rpc，只要在請(qǐng)求的參數(shù)前添加stream即可

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

4.編譯proto文件

進(jìn)入client_stream.proto所在目錄，運(yùn)行指令:

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

創(chuàng)建Server端

1.定義我們的服務(wù)，并實(shí)現(xiàn)RouteList方法

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

2.啟動(dòng)gRPC服務(wù)器

const (
	// Address 監(jiān)聽地址
	Address string = ":8000"
	// Network 網(wǎng)絡(luò)通信協(xié)議
	Network string = "tcp"
)

func main() {
	// 監(jiān)聽本地端口
	listener, err := net.Listen(Network, Address)
	if err != nil {
		log.Fatalf("net.Listen err: %v", err)
	}
	log.Println(Address + " net.Listing...")
	// 新建gRPC服務(wù)器實(shí)例
	grpcServer := grpc.NewServer()
	// 在gRPC服務(wù)器注冊(cè)我們的服務(wù)
	pb.RegisterStreamClientServer(grpcServer, &SimpleService{})

	//用服務(wù)器 Serve() 方法以及我們的端口信息區(qū)實(shí)現(xiàn)阻塞等待，直到進(jìn)程被殺死或者 Stop() 被調(diào)用
	err = grpcServer.Serve(listener)
	if err != nil {
		log.Fatalf("grpcServer.Serve err: %v", err)
	}
}

運(yùn)行服務(wù)端

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

創(chuàng)建Client端

1.創(chuàng)建調(diào)用服務(wù)端RouteList方法

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

2.啟動(dòng)gRPC客戶端

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

var streamClient pb.StreamClientClient

func main() {
	// 連接服務(wù)器
	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()
}

運(yùn)行客戶端

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

服務(wù)端不斷從客戶端獲取到數(shù)據(jù)

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

思考

服務(wù)端在沒有接受完消息時(shí)候能主動(dòng)停止接收數(shù)據(jù)嗎（很少有這種場(chǎng)景）？

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

1.我們把服務(wù)端的RouteList方法實(shí)現(xiàn)稍微修改，當(dāng)接收到一條數(shù)據(jù)后馬上調(diào)用SendAndClose()關(guān)閉stream.

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

2.再把客戶端調(diào)用RouteList方法的實(shí)現(xiàn)稍作修改

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

客戶端Send()需要檢測(cè)err是否為EOF，因?yàn)楫?dāng)服務(wù)端在消息沒接收完前主動(dòng)調(diào)用SendAndClose()關(guān)閉stream，若此時(shí)客戶端繼續(xù)執(zhí)行Send()，則會(huì)返回EOF錯(cuò)誤。