Go gRPC伺服器端流式RPC教學範例

2022-06-16 10:02:58

前言

上一篇介紹了簡單模式RPC，當資料量大或者需要不斷傳輸資料時候，我們應該使用流式RPC，它允許我們邊處理邊傳輸資料。本篇先介紹伺服器端流式RPC。

伺服器端流式RPC：使用者端傳送請求到伺服器，拿到一個流去讀取返回的訊息序列。使用者端讀取返回的流，直到裡面沒有任何訊息。

情景模擬：實時獲取股票走勢

1.使用者端要獲取某原油股的實時走勢，使用者端傳送一個請求

2.伺服器端實時返回該股票的走勢

新建proto檔案

新建server_stream.proto檔案

1.定義傳送資訊

// 定義傳送請求資訊
message SimpleRequest{
    // 定義傳送的引數，採用駝峰命名方式，小寫加下劃線，如：student_name
    // 請求引數
    string data = 1;
}

2.定義接收資訊

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

3.定義服務方法ListValue

伺服器端流式rpc，只要在響應資料前新增stream即可

// 定義我們的服務（可定義多個服務,每個服務可定義多個介面）
service StreamServer{
    // 伺服器端流式rpc，在響應資料前新增stream
    rpc ListValue(SimpleRequest)returns(stream StreamResponse){};
}

4.編譯proto檔案

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

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

建立Server端

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

// SimpleService 定義我們的服務
type StreamService struct{}
// ListValue 實現ListValue方法
func (s *StreamService) ListValue(req *pb.SimpleRequest, srv pb.StreamServer_ListValueServer) error {
	for n := 0; n < 5; n++ {
		// 向流中傳送訊息， 預設每次send送訊息最大長度為`math.MaxInt32`bytes
		err := srv.Send(&pb.StreamResponse{
			StreamValue: req.Data + strconv.Itoa(n),
		})
		if err != nil {
			return err
		}
	}
	return nil
}

初學者可能覺得比較迷惑，ListValue的引數和返回值是怎樣確定的。其實這些都是編譯proto時生成的.pb.go檔案中有定義，我們只需要實現就可以了。

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伺服器範例
	// 預設單次接收最大訊息長度為`1024*1024*4`bytes(4M)，單次傳送訊息最大長度為`math.MaxInt32`bytes
	// grpcServer := grpc.NewServer(grpc.MaxRecvMsgSize(1024*1024*4), grpc.MaxSendMsgSize(math.MaxInt32))
	grpcServer := grpc.NewServer()
	// 在gRPC伺服器註冊我們的服務
	pb.RegisterStreamServerServer(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.建立呼叫伺服器端ListValue方法

// listValue 呼叫伺服器端的ListValue方法
func listValue() {
	// 建立傳送結構體
	req := pb.SimpleRequest{
		Data: "stream server grpc ",
	}
	// 呼叫我們的服務(ListValue方法)
	stream, err := grpcClient.ListValue(context.Background(), &req)
	if err != nil {
		log.Fatalf("Call ListStr err: %v", err)
	}
	for {
		//Recv() 方法接收伺服器端訊息，預設每次Recv()最大訊息長度為`1024*1024*4`bytes(4M)
		res, err := stream.Recv()
		// 判斷訊息流是否已經結束
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatalf("ListStr get stream err: %v", err)
		}
		// 列印返回值
		log.Println(res.StreamValue)
	}
}

2.啟動gRPC使用者端

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

var grpcClient pb.StreamServerClient

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

	// 建立gRPC連線
	grpcClient = pb.NewStreamServerClient(conn)
	route()
	listValue()
}

執行使用者端

go run client.go
stream server grpc 0
stream server grpc 1
stream server grpc 2
stream server grpc 3
stream server grpc 4

使用者端不斷從伺服器端獲取資料

思考

假如伺服器端不停傳送資料，類似獲取股票走勢實時資料，使用者端能自己停止獲取資料嗎？

答案：可以的

1.我們把伺服器端的ListValue方法稍微修改

// ListValue 實現ListValue方法
func (s *StreamService) ListValue(req *pb.SimpleRequest, srv pb.StreamServer_ListValueServer) error {
	for n := 0; n < 15; n++ {
		// 向流中傳送訊息， 預設每次send送訊息最大長度為`math.MaxInt32`bytes
		err := srv.Send(&pb.StreamResponse{
			StreamValue: req.Data + strconv.Itoa(n),
		})
		if err != nil {
			return err
		}
		log.Println(n)
		time.Sleep(1 * time.Second)
	}
	return nil
}

2.再把使用者端呼叫ListValue方法的實現稍作修改，就可以得到結果了

// listValue 呼叫伺服器端的ListValue方法
func listValue() {
	// 建立傳送結構體
	req := pb.SimpleRequest{
		Data: "stream server grpc ",
	}
	// 呼叫我們的服務(Route方法)
	// 同時傳入了一個 context.Context ，在有需要時可以讓我們改變RPC的行為，比如超時/取消一個正在執行的RPC
	stream, err := grpcClient.ListValue(context.Background(), &req)
	if err != nil {
		log.Fatalf("Call ListStr err: %v", err)
	}
	for {
		//Recv() 方法接收伺服器端訊息，預設每次Recv()最大訊息長度為`1024*1024*4`bytes(4M)
		res, err := stream.Recv()
		// 判斷訊息流是否已經結束
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatalf("ListStr get stream err: %v", err)
		}
		// 列印返回值
		log.Println(res.StreamValue)
		break
	}
	//可以使用CloseSend()關閉stream，這樣伺服器端就不會繼續產生流訊息
	//呼叫CloseSend()後，若繼續呼叫Recv()，會重新啟用stream，接著之前結果獲取訊息
	stream.CloseSend()
}

只需要呼叫CloseSend()方法，就可以關閉伺服器端的stream，讓它停止傳送資料。值得注意的是，呼叫CloseSend()後，若繼續呼叫Recv()，會重新啟用stream，接著當前的結果繼續獲取訊息。

這能完美解決使用者端暫停->繼續獲取資料的操作。