Go gRPC教程-服務端流式RPC（三）
前言
上一篇介紹了簡單模式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:./ ./simple.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，接著當前的結果繼續獲取訊息。
這能完美解決客戶端暫停->繼續獲取資料的操作。
總結
本篇介紹了服務端流式RPC的簡單實用，客戶端發起一個請求，服務端不停返回資料，直到服務端停止傳送資料或客戶端主動停止接收資料為止。下篇將介紹客戶端流式RPC。
教程原始碼地址：https://github.com/Bingjian-Zhu/go-grpc-example
參考：gRPC官方文件中文