Go gRPC教程-雙向流式RPC

一、前言

上一篇介紹了客戶端流式RPC，客戶端不斷的向服務端傳送資料流，在傳送結束或流關閉後，由服務端返回一個響應。本篇將介紹雙向流式RPC。

雙向流式RPC：客戶端和服務端雙方使用讀寫流去傳送一個訊息序列，兩個流獨立操作，雙方可以同時傳送和同時接收。

情景模擬：雙方對話（可以一問一答、一問多答、多問一答，形式靈活）。

原始碼

二、新建proto檔案

新建bothStream.proto檔案

1.定義傳送資訊

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

2.定義接收資訊

// 定義流式響應資訊
message StreamResponse{
    //流響應資料
    string answer = 1;
}

3.定義服務方法Conversations

雙向流式rpc，只要在請求的引數前和響應引數前都新增stream即可

service Stream{
    // 雙向流式rpc，同時在請求引數前和響應引數前加上stream
    rpc Conversations(stream StreamRequest) returns(stream StreamResponse){};
}

4.編譯proto檔案

syntax = "proto3";// 協議為proto3

package proto;
option go_package = "./;proto";

//  生成pb.go命令:  protoc -I ./ --go_out=plugins=grpc:.\05bothStream\proto\  .\05bothStream\proto\bothStream.proto

// 定義我們的服務（可定義多個服務,每個服務可定義多個介面）
service Stream{
  // 雙向流式rpc，同時在請求引數前和響應引數前加上stream
  rpc Conversations(stream StreamRequest) returns(stream StreamResponse){};
}

// 定義傳送請求資訊
message StreamRequest{
  //流請求引數
  string question = 1;
}

// 定義流式響應資訊
message StreamResponse{
  //流響應資料
  string answer = 1;
}

 

進入bothStream.proto所在目錄，執行指令:

go-grpc-example> protoc -I ./ --go_out=plugins=grpc:.\05bothStream\proto\  .\05bothStream\proto\bothStream.proto

三、建立Server端

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

這裡簡單實現對話中一問一答的形式

// StreamService 定義我們的服務
type StreamService struct{}
// Conversations 實現Conversations方法
func (s *StreamService) Conversations(srv pb.Stream_ConversationsServer) error {
	n := 1
	for {
		req, err := srv.Recv()
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return err
		}
		err = srv.Send(&pb.StreamResponse{
			Answer: "from stream server answer: the " + strconv.Itoa(n) + " question is " + req.Question,
		})
		if err != nil {
			return err
		}
		n++
		log.Printf("from stream client question: %s", req.Question)
	}
}
 

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.RegisterStreamServer(grpcServer, &StreamService{})

	//用伺服器 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.建立呼叫服務端Conversations方法

// conversations 呼叫服務端的Conversations方法
func conversations() {
	//呼叫服務端的Conversations方法，獲取流
	stream, err := streamClient.Conversations(context.Background())
	if err != nil {
		log.Fatalf("get conversations stream err: %v", err)
	}
	for n := 0; n < 5; n++ {
		err := stream.Send(&pb.StreamRequest{Question: "stream client rpc " + strconv.Itoa(n)})
		if err != nil {
			log.Fatalf("stream request err: %v", err)
		}
		res, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatalf("Conversations get stream err: %v", err)
		}
		// 列印返回值
		log.Println(res.Answer)
	}
	//最後關閉流
	err = stream.CloseSend()
	if err != nil {
		log.Fatalf("Conversations close stream err: %v", err)
	}
}

2.啟動gRPC客戶端

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

var streamClient pb.StreamClient

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.NewStreamClient(conn)
	conversations()
}

服務端獲取到來自客戶端的提問

2022/03/28 11:28:00 :8000 net.Listing...
2022/03/28 11:28:11 from stream client question: stream client rpc 0
2022/03/28 11:28:11 from stream client question: stream client rpc 1
2022/03/28 11:28:11 from stream client question: stream client rpc 2
2022/03/28 11:28:11 from stream client question: stream client rpc 3
2022/03/28 11:28:11 from stream client question: stream client rpc 4

執行客戶端，獲取到服務端的應答

2022/03/28 11:28:11 from stream server answer: the 1question is stream client rpc 0
2022/03/28 11:28:11 from stream server answer: the 2question is stream client rpc 1
2022/03/28 11:28:11 from stream server answer: the 3question is stream client rpc 2
2022/03/28 11:28:11 from stream server answer: the 4question is stream client rpc 3
2022/03/28 11:28:11 from stream server answer: the 5question is stream client rpc 4

五、總結

本篇介紹了雙向流式RPC的簡單使用。
參考：gRPC官方文件中文版