Go教程 · grpc

> From: http://doc.oschina.net/grpc?t=60133 ## 為什么使用 gRPC? 我們的例子是一個簡單的路由映射的應用，它允許客戶端獲取路由特性的信息，生成路由的總結，以及交互路由信息，如服務器和其他客戶端的流量更新。有了 gRPC，我們可以一次性的在一個 .proto 文件中定義服務并使用任何支持它的語言去實現客戶端和服務器，反過來，它們可以在各種環境中，從Google的服務器到你自己的平板電腦—— gRPC 幫你解決了不同語言及環境間通信的復雜性.使用 protocol buffers 還能獲得其他好處，包括高效的序列號，簡單的 IDL 以及容易進行接口更新。 ## 例子的代碼和設置教程的代碼在這里 [grpc/grpc-go/examples/cpp/route\_guide](https://github.com/grpc/grpc-go/tree/master/examples/route_guide)。要下載例子，通過運行下面的命令去克隆`grpc-go`代碼庫： ~~~ $ go get google.golang.org/grpc ~~~ 然后改變當前的目錄到 `grpc-go/examples/route_guide`: ~~~ $ cd $GOPATH/src/google.golang.org/grpc/examples/route_guide ~~~ 你還需要安裝生成服務器和客戶端的接口代碼相關工具-如果你還沒有安裝的話，請查看下面的設置指南 [Go快速開始指南](http://doc.oschina.net/docs/installation/go.html)。 ## 定義服務我們的第一步(可以從[概覽](http://doc.oschina.net/docs/index.html)中得知)是使用 [protocol buffers](https://developers.google.com/protocol-buffers/docs/overview)去定義 gRPC *service* 和方法 *request* 以及 *response* 的類型。你可以在[`examples/protos/route_guide.proto`](https://github.com/grpc/grpc/blob/{{ site.data.config.grpc_release_branch }}/examples/protos/route_guide.proto)看到完整的 .proto 文件。要定義一個服務，你必須在你的 .proto 文件中指定 `service`： ~~~ service RouteGuide { ... } ~~~ 然后在你的服務中定義 `rpc` 方法，指定請求的和響應類型。gRPC 允許你定義4種類型的 service 方法，這些都在 `RouteGuide` 服務中使用： * 一個 *簡單 RPC* ，客戶端使用存根發送請求到服務器并等待響應返回，就像平常的函數調用一樣。 ~~~ // Obtains the feature at a given position. rpc GetFeature(Point) returns (Feature) {} ~~~ * 一個 *服務器端流式 RPC* ，客戶端發送請求到服務器，拿到一個流去讀取返回的消息序列。客戶端讀取返回的流，直到里面沒有任何消息。從例子中可以看出，通過在 *響應* 類型前插入 `stream` 關鍵字，可以指定一個服務器端的流方法。 ~~~ // Obtains the Features available within the given Rectangle. Results are // streamed rather than returned at once (e.g. in a response message with a // repeated field), as the rectangle may cover a large area and contain a // huge number of features. rpc ListFeatures(Rectangle) returns (stream Feature) {} ~~~ * 一個 *客戶端流式 RPC* ，客戶端寫入一個消息序列并將其發送到服務器，同樣也是使用流。一旦客戶端完成寫入消息，它等待服務器完成讀取返回它的響應。通過在 *請求* 類型前指定 `stream` 關鍵字來指定一個客戶端的流方法。 ~~~ // Accepts a stream of Points on a route being traversed, returning a // RouteSummary when traversal is completed. rpc RecordRoute(stream Point) returns (RouteSummary) {} ~~~ * 一個 *雙向流式 RPC* 是雙方使用讀寫流去發送一個消息序列。兩個流獨立操作，因此客戶端和服務器可以以任意喜歡的順序讀寫：比如，服務器可以在寫入響應前等待接收所有的客戶端消息，或者可以交替的讀取和寫入消息，或者其他讀寫的組合。每個流中的消息順序被預留。你可以通過在請求和響應前加 `stream` 關鍵字去制定方法的類型。 ~~~ // Accepts a stream of RouteNotes sent while a route is being traversed, // while receiving other RouteNotes (e.g. from other users). rpc RouteChat(stream RouteNote) returns (stream RouteNote) {} ~~~ 我們的 .proto 文件也包含了所有請求的 protocol buffer 消息類型定義以及在服務方法中使用的響應類型——比如，下面的`Point`消息類型： ~~~ // Points are represented as latitude-longitude pairs in the E7 representation // (degrees multiplied by 10**7 and rounded to the nearest integer). // Latitudes should be in the range +/- 90 degrees and longitude should be in // the range +/- 180 degrees (inclusive). message Point { int32 latitude = 1; int32 longitude = 2; } ~~~ ## 生成客戶端和服務器端代碼接下來我們需要從 .proto 的服務定義中生成 gRPC 客戶端和服務器端的接口。我們通過 protocol buffer 的編譯器 `protoc` 以及一個特殊的 gRPC Go 插件來完成。簡單起見，我們提供一個 [bash 腳本](https://github.com/grpc/grpc-go/blob/master/codegen.sh) 幫你用合適的插件，輸入，輸出去運行 `protoc`(如果你想自己去運行，確保你已經安裝了 protoc，并且請遵循下面的 gRPC-Go [安裝指南](https://github.com/grpc/grpc-go/blob/master/README.md))來操作： ~~~ $ codegen.sh route_guide.proto ~~~ 實際上運行的是： ~~~ $ protoc --go_out=plugins=grpc:. route_guide.proto ~~~ 運行這個命令可以在當前目錄中生成下面的文件： * `route_guide.pb.go` 這些包括： * 所有用于填充，序列化和獲取我們請求和響應消息類型的 protocol buffer 代碼 * 一個為客戶端調用定義在`RouteGuide`服務的方法的接口類型（或者 *存根* ） * 一個為服務器使用定義在`RouteGuide`服務的方法去實現的接口類型（或者 *存根* ） ## 創建服務器首先來看看我們如何創建一個 `RouteGuide` 服務器。如果你只對創建 gRPC 客戶端感興趣，你可以跳過這個部分，直接到[創建客戶端](http://doc.oschina.net/grpc?t=60133#client) (當然你也可能發現它也很有意思)。讓 `RouteGuide` 服務工作有兩個部分： * 實現我們服務定義的生成的服務接口：做我們的服務的實際的“工作”。 * 運行一個 gRPC 服務器，監聽來自客戶端的請求并返回服務的響應。你可以從[grpc-go/examples/route\_guide/server/server.go](https://github.com/grpc/grpc-go/tree/master/examples/route_guide/server/server.go)看到我們的 `RouteGuide` 服務器的實現代碼。現在讓我們近距離研究它是如何工作的。 ### 實現RouteGuide 我們可以看出，服務器有一個實現了生成的 `RouteGuideServer` 接口的 `routeGuideServer` 結構類型： ~~~ type routeGuideServer struct { ... } ... func (s *routeGuideServer) GetFeature(ctx context.Context, point *pb.Point) (*pb.Feature, error) { ... } ... func (s *routeGuideServer) ListFeatures(rect *pb.Rectangle, stream pb.RouteGuide_ListFeaturesServer) error { ... } ... func (s *routeGuideServer) RecordRoute(stream pb.RouteGuide_RecordRouteServer) error { ... } ... func (s *routeGuideServer) RouteChat(stream pb.RouteGuide_RouteChatServer) error { ... } ... ~~~ #### 簡單 RPC `routeGuideServer` 實現了我們所有的服務方法。首先讓我們看看最簡單的類型 `GetFeature`，它從客戶端拿到一個 `Point` 對象，然后從返回包含從數據庫拿到的feature信息的 `Feature`. ~~~ func (s *routeGuideServer) GetFeature(ctx context.Context, point *pb.Point) (*pb.Feature, error) { for _, feature := range s.savedFeatures { if proto.Equal(feature.Location, point) { return feature, nil } } // No feature was found, return an unnamed feature return &pb.Feature{"", point}, nil } ~~~ 該方法傳入了 RPC 的上下文對象，以及客戶端的 `Point` protocol buffer請求。它返回了一個包含響應信息和`error` 的 `Feature` protocol buffer對象。在方法中我們用適當的信息填充 `Feature`，然后將其和一個`nil`錯誤一起返回，告訴 gRPC 我們完成了對 RPC 的處理，并且 `Feature` 可以返回給客戶端。 #### 服務器端流式 RPC 現在讓我們來看看我們的一種流式 RPC。 `ListFeatures` 是一個服務器端的流式 RPC，所以我們需要將多個 `Feature` 發回給客戶端。 ~~~ func (s *routeGuideServer) ListFeatures(rect *pb.Rectangle, stream pb.RouteGuide_ListFeaturesServer) error { for _, feature := range s.savedFeatures { if inRange(feature.Location, rect) { if err := stream.Send(feature); err != nil { return err } } } return nil } ~~~ 如你所見，這里的請求對象是一個 `Rectangle`，客戶端期望從中找到 `Feature`，這次我們得到了一個請求對象和一個特殊的`RouteGuide_ListFeaturesServer`來寫入我們的響應，而不是得到方法參數中的簡單請求和響應對象。在這個方法中，我們填充了盡可能多的 `Feature` 對象去返回，用它們的 `Send()` 方法把它們寫入 `RouteGuide_ListFeaturesServer`。最后，在我們的簡單 RPC中，我們返回了一個 `nil` 錯誤告訴 gRPC 響應的寫入已經完成。如果在調用過程中發生任何錯誤，我們會返回一個非 `nil` 的錯誤；gRPC 層會將其轉化為合適的 RPC 狀態通過線路發送。 #### 客戶端流式 RPC 現在讓我們看看稍微復雜點的東西：客戶端流方法 `RecordRoute`，我們通過它可以從客戶端拿到一個 `Point` 的流，其中包括它們路徑的信息。如你所見，這次這個方法沒有請求參數。相反的，它拿到了一個 `RouteGuide_RecordRouteServer` 流，服務器可以用它來同時讀 *和* 寫消息——它可以用自己的 `Recv()` 方法接收客戶端消息并且用 `SendAndClose()` 方法返回它的單個響應。 ~~~ func (s *routeGuideServer) RecordRoute(stream pb.RouteGuide_RecordRouteServer) error { var pointCount, featureCount, distance int32 var lastPoint *pb.Point startTime := time.Now() for { point, err := stream.Recv() if err == io.EOF { endTime := time.Now() return stream.SendAndClose(&pb.RouteSummary{ PointCount: pointCount, FeatureCount: featureCount, Distance: distance, ElapsedTime: int32(endTime.Sub(startTime).Seconds()), }) } if err != nil { return err } pointCount++ for _, feature := range s.savedFeatures { if proto.Equal(feature.Location, point) { featureCount++ } } if lastPoint != nil { distance += calcDistance(lastPoint, point) } lastPoint = point } } ~~~ 在方法體中，我們使用 `RouteGuide_RecordRouteServer` 的 `Recv()` 方法去反復讀取客戶端的請求到一個請求對象（在這個場景下是 `Point`），直到沒有更多的消息：服務器需要在每次調用后檢查 `Read()` 返回的錯誤。如果返回值為 `nil`，流依然完好，可以繼續讀取；如果返回值為 `io.EOF`，消息流結束，服務器可以返回它的 `RouteSummary`。如果它還有其它值，我們原樣返回錯誤，gRPC 層會把它轉換為 RPC 狀態。 #### 雙向流式 RPC 最后，讓我們看看雙向流式 RPC `RouteChat()`。 ~~~ func (s *routeGuideServer) RouteChat(stream pb.RouteGuide_RouteChatServer) error { for { in, err := stream.Recv() if err == io.EOF { return nil } if err != nil { return err } key := serialize(in.Location) ... // look for notes to be sent to client for _, note := range s.routeNotes[key] { if err := stream.Send(note); err != nil { return err } } } } ~~~ 這次我們得到了一個 `RouteGuide_RouteChatServer` 流，和我們的客戶端流的例子一樣，它可以用來讀寫消息。但是，這次當客戶端還在往 *它們* 的消息流中寫入消息時，我們通過方法的流返回值。這里讀寫的語法和客戶端流方法相似，除了服務器會使用流的 `Send()` 方法而不是 `SendAndClose()`，因為它需要寫多個響應。雖然客戶端和服務器端總是會拿到對方寫入時順序的消息，它們可以以任意順序讀寫——流的操作是完全獨立的。 ### 啟動服務器一旦我們實現了所有的方法，我們還需要啟動一個gRPC服務器，這樣客戶端才可以使用服務。下面這段代碼展示了在我們`RouteGuide`服務中實現的過程： ~~~ flag.Parse() lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port)) if err != nil { log.Fatalf("failed to listen: %v", err) } grpcServer := grpc.NewServer() pb.RegisterRouteGuideServer(grpcServer, &routeGuideServer{}) ... // determine whether to use TLS grpcServer.Serve(lis) ~~~ 為了構建和啟動服務器，我們需要： 1. 使用 `lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))` 指定我們期望客戶端請求的監聽端口。 2. 使用`grpc.NewServer()`創建 gRPC 服務器的一個實例。 3. 在 gRPC 服務器注冊我們的服務實現。 4. 用服務器 `Serve()` 方法以及我們的端口信息區實現阻塞等待，直到進程被殺死或者 `Stop()` 被調用。 ## 創建客戶端在這部分，我們將嘗試為 `RouteGuide` 服務創建一個 Go 的客戶端。你可以從[grpc-go/examples/route\_guide/client/client.go](https://github.com/grpc/grpc-go/tree/master/examples/route_guide/client/client.go)看到我們完整的客戶端例子代碼. ### 創建存根為了調用服務方法，我們首先創建一個 gRPC *channel* 和服務器交互。我們通過給 `grpc.Dial()` 傳入服務器地址和端口號做到這點，如下： ~~~ conn, err := grpc.Dial(*serverAddr) if err != nil { ... } defer conn.Close() ~~~ 你可以使用 `DialOptions` 在 `grpc.Dial` 中設置授權認證（如， TLS，GCE認證，JWT認證），如果服務有這樣的要求的話 —— 但是對于 `RouteGuide` 服務，我們不用這么做。一旦 gRPC *channel* 建立起來，我們需要一個客戶端 *存根* 去執行 RPC。我們通過 .proto 生成的 `pb` 包提供的 `NewRouteGuideClient` 方法來完成。 ~~~ client := pb.NewRouteGuideClient(conn) ~~~ ### 調用服務方法現在讓我們看看如何調用服務方法。注意，在 gRPC-Go 中，RPC以阻塞/同步模式操作，這意味著 RPC 調用等待服務器響應，同時要么返回響應，要么返回錯誤。 #### 簡單 RPC 調用簡單 RPC `GetFeature` 幾乎是和調用一個本地方法一樣直觀。 ~~~ feature, err := client.GetFeature(context.Background(), &pb.Point{409146138, -746188906}) if err != nil { ... } ~~~ 如你所見，我們調用了前面創建的存根上的方法。在我們的方法參數中，我們創建并且填充了一個請求的 protocol buffer 對象（例子中為 `Point`）。我們同時傳入了一個 `context.Context` ，在有需要時可以讓我們改變 RPC 的行為，比如超時/取消一個正在運行的 RPC。如果調用沒有返回錯誤，那么我們就可以從服務器返回的第一個返回值中讀到響應信息。 ~~~ log.Println(feature) ~~~ #### 服務器端流式 RPC `ListFeatures` 就是我們說的服務器端流方法，它會返回地理的`Feature` 流。如果你已經讀過[創建服務器](http://doc.oschina.net/grpc?t=60133#server)，本節的一些內容也許看上去會很熟悉——流式 RPC 是在客戶端和服務器兩端以一種類似的方式實現的。 ~~~ rect := &pb.Rectangle{ ... } // initialize a pb.Rectangle stream, err := client.ListFeatures(context.Background(), rect) if err != nil { ... } for { feature, err := stream.Recv() if err == io.EOF { break } if err != nil { log.Fatalf("%v.ListFeatures(_) = _, %v", client, err) } log.Println(feature) } ~~~ 在簡單 RPC 的例子中，我們給方法傳入一個上下文和請求。然而，我們得到返回的是一個 `RouteGuide_ListFeaturesClient` 實例，而不是一個應答對象。客戶端可以使用 `RouteGuide_ListFeaturesClient` 流去讀取服務器的響應。我們使用 `RouteGuide_ListFeaturesClient` 的 `Recv()` 方法去反復讀取服務器的響應到一個響應 protocol buffer 對象（在這個場景下是`Feature`）直到消息讀取完畢：每次調用完成時，客戶端都要檢查從 `Recv()` 返回的錯誤 `err`。如果返回為 `nil`，流依然完好并且可以繼續讀取；如果返回為 `io.EOF`，則說明消息流已經結束；否則就一定是一個通過 `err` 傳過來的 RPC 錯誤。 #### 客戶端流式 RPC 除了我們需要給方法傳入一個上下文而后返回 `RouteGuide_RecordRouteClient` 流以外，客戶端流方法 `RecordRoute` 和服務器端方法類似，它可以用來讀 *和* 寫消息。 ~~~ // Create a random number of random points r := rand.New(rand.NewSource(time.Now().UnixNano())) pointCount := int(r.Int31n(100)) + 2 // Traverse at least two points var points []*pb.Point for i := 0; i < pointCount; i++ { points = append(points, randomPoint(r)) } log.Printf("Traversing %d points.", len(points)) stream, err := client.RecordRoute(context.Background()) if err != nil { log.Fatalf("%v.RecordRoute(_) = _, %v", client, err) } for _, point := range points { if err := stream.Send(point); err != nil { log.Fatalf("%v.Send(%v) = %v", stream, point, err) } } reply, err := stream.CloseAndRecv() if err != nil { log.Fatalf("%v.CloseAndRecv() got error %v, want %v", stream, err, nil) } log.Printf("Route summary: %v", reply) ~~~ `RouteGuide_RecordRouteClient` 有一個 `Send()` 方法，我們可以用它來給服務器發送請求。一旦我們完成使用 `Send()` 方法將客戶端請求寫入流，就需要調用流的 `CloseAndRecv()`方法，讓 gRPC 知道我們已經完成了寫入同時期待返回應答。我們從 `CloseAndRecv()` 返回的 `err` 中獲得 RPC 的狀態。如果狀態為`nil`，那么`CloseAndRecv()`的第一個返回值將會是合法的服務器應答。 #### 雙向流式 RPC 最后，讓我們看看雙向流式 RPC `RouteChat()`。和 `RecordRoute` 的場景類似，我們只給函數傳入一個上下文對象，拿到可以用來讀寫的流。但是，當服務器依然在往 *他們* 的消息流寫入消息時，我們通過方法流返回值。 ~~~ stream, err := client.RouteChat(context.Background()) waitc := make(chan struct{}) go func() { for { in, err := stream.Recv() if err == io.EOF { // read done. close(waitc) return } if err != nil { log.Fatalf("Failed to receive a note : %v", err) } log.Printf("Got message %s at point(%d, %d)", in.Message, in.Location.Latitude, in.Location.Longitude) } }() for _, note := range notes { if err := stream.Send(note); err != nil { log.Fatalf("Failed to send a note: %v", err) } } stream.CloseSend() <-waitc ~~~ 這里讀寫的語法和我們的客戶端流方法很像，除了在完成調用時，我們會使用流的 `CloseSend()` 方法。雖然每一端獲取對方信息的順序和信息被寫入的順序一致，客戶端和服務器都可以以任意順序讀寫——流的操作是完全獨立的。 ## 來試試吧！假設你在 `$GOPATH/src/google.golang.org/grpc/examples/route_guide` 目錄，要編譯和運行服務器，只需要運行： ~~~ $ go run server/server.go ~~~ 同樣的，運行客戶端: ~~~ $ go run client/client.go ~~~