Go gRPC教程-客戶端流式RPC(四)
阿新 • • 發佈:2020-04-14
前言
上一篇介紹了服務端流式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
。
教程原始碼地址:https://github.com/Bingjian-Zhu/go-grpc-example
參考:gRPC官方文件中文