grpc 基础知识 - Go

该篇文章介绍了golang的grpc编程。

通过下面的例子，你将会学到：
1. 在一个.proto文件里define一个service
2. 用protocol buffer 编译工具生成server和client代码
3. 用go grpc的api写一个简单的server和client

你应该已经读过 概述( Overview),并且对 protocol buffers有所了解，注意教程中的例子用了protocol buffer 的proto3版本：详情见以下两个链接 proto3 language guide , Go generated code guide.

为什么用grpc？

我们的示例是一个简单的路由映射应用程序，它允许客户机获取有关其路由上功能的信息，创建其路由的摘要，并与服务器和其他客户机交换路由信息，如流量更新。
用grpc我们可以在.proto文件中定义我们的服务，并且支持任何语言的server和client..这些server和client可以在从谷歌的内部服务器到你的平板电脑上运行。所有的不同平台的语言的通讯的复杂性都由grpc为你处理。用protocol buffer的优势包括高效的序列化/简单的IDL/高可用的接口更新

实例代码和setup

示例代码在 grpc/grpc-go/examples/route_guide。去下载示例代码并且clone grpc-go通过运行如下命令：

$ go get google.golang.org/grpc

cd到grpc-go/examples/route_guide：

$ cd GOPATH/src/google.golang.org/grpc/examples/route_guide

你应该已经安装了相应的工具来生成server和client的接口代码。如果没有看如下链接：Go quick start guide.

定义service

我们的第一步是通过protoco buffer 定义grpc的service以及response和request的类型.你可以在examples/route_guide/routeguide/route_guide.proto看到完成的.proto文件。
在proto文件中定义service：

service RouteGuide {
   ...
}

然后再service里定义rpc方法，指定他们的request和response方法。grpc允许你定义四种类型的rpc方法。他们在RouteGuide 里都有用到。

• 一个简单的RPC，客户端用stub发送请求到服务端，等待一个回应。就像普通的函数调用。

// Obtains the feature at a given position.
rpc GetFeature(Point) returns (Feature) {}

• 一个server侧streaming RPC，客户端发送一个请求到server，获取一个stream,可以read一系列message。客户端read这个stream直到全部读完。在例子中可以看到，你指定server侧streaming方法，通过在response类型前面放置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) {}

• 一个客户端侧streaming RPC，客户端通过一个提供的stream,写入一系列的messages,并且发送到server。一旦客户端全部写完，他等待server全部读完，并且返回一个response。你指定客户端侧streaming方法，通过在request类型前面放置stream关键字。

// Accepts a stream of Points on a route being traversed, returning a
// RouteSummary when traversal is completed.
rpc RecordRoute(stream Point) returns (RouteSummary) {}

• 一个双向的streaming RPC,两侧都通过read-write stream发送一串message。这两个steam的操作是独立的。因此client和server可以以任何进行读/写。例如server可以等message读完了之后回一个response。也可以读完之前回复response。也可以读一个回复一个，或者其他的读写组合。每个stream内message的顺序是保存好的。stream关键字在request/response前面。

// 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 message类型（request/response的参数）的定义。例如下面是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;
}

生产client/server代码

下面我们要通过.proto文件生成grpc client/server的接口。我们通过一个带有go插件的编译工具protoc。这类似于我们在quickstart guide所作的。
在route_guide实例目录执行：

protoc -I routeguide/ routeguide/route_guide.proto --go_out=plugins=grpc:routeguide

执行这个命令，在routeguide目录下生产文件：

route_guide.pb.go

文件包含：

• 用于填充、序列化和检索请求和响应消息类型的所有的protocol buffer代码
• 被客户端调用的一个接口类型(或 stub),包含在RouteGuide service中定义的一个方法。
• 被服务端实现的一个接口类型(或 stub),包含在RouteGuide service中定义的一个方法。

创建server

首先我们来看看如何生产RouteGuide server。如果你只对生产rpc client 感兴趣，请跳过该章节。跳到Creating the client

• 实现由service定义生成的service interface.
• 运行gRPC服务器来侦听来自客户机的请求并将其发送到正确的服务实现。
  RouteGuide server的实现在 grpc-go/examples/route_guide/server/server.go。让我们进一步看一下，他们是如何工作的：

实现RouteGuide

如下，我们的server有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 实现所有的services方法。先看一下最简单的GetFeature，它从客户端获取Point然后return一个对应的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
}

这个方法传递一个context对象和client的Pointprotocol buffer 请求。它返回一个包含response信息的Featureprotocol buffer 对象和error。

server侧streaming RPC

我们来看一个streaming RPCs。 ListFeatures是一个server侧的streaming rpc，因此我们需要发送多个Fearute到client。

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
}

函数有一个请求参数（client想要通过Rectangle 来查找Features)和一个RouteGuide_ListFeaturesServer对象去write responses.
函数中，我们填充并return所有的Feature，通过Send()把他们写到RouteGuide_ListFeaturesServer 。

client侧streaming RPC

现在来看一下复杂一点的：客户端侧streaming方法RecordRoute。我们从客户端收到一个stream（内有多个Point），并且返回单个RouteSummary 。正如你所看到的，这次没有request参数，只有一个RouteGuide_RecordRouteServer stream，server可以用来读写message，通过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
    }
}

通过Recv()持续读，直到io.EOF。

双向streaming RPC

最后，我们来看一下双向streaming 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
            }
        }
    }
}

这里读和写的语法和我们的client-streaming 的方法特别像，不同点是server用stream的Send()方法，而不是SendAndClose()，因为需要写入多个response。

启动server

我们完成所有的方法后，需要启动gRPC server，client才能发送请求。下面的片段展示了我们如何启动RouteGuide service的。

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)

新建并且启动一个server：

1. 指定用来监听request的的端口，使用lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
2. 用grpc.NewServer()创建gRPC server的实例。
3. 在GRPC服务器上注册我们的服务实现。
4. 调用Serve()阻塞等待，直到进程被kill或者Stop()被调用。

创建一个client

这一部分我们会看一下如何为RouteGuideservice创建GO client。你可以看到完整的client code在 grpc-go/examples/route_guide/client/client.go

创建一个stub

为了调用服务的method,我们首先需要创建通讯的管道。把server的地址和port传到grpc.Dial(),如下：

conn, err := grpc.Dial(*serverAddr)
if err != nil {
    ...
}
defer conn.Close()

如果service要求的话，你可以在grpc.Dial中用DialOptions设置授权证书。
我们通过.proto文件生成的pb package中的NewRouteGuideClient 方法获取client stub。

client := pb.NewRouteGuideClient(conn)

调用service的方法

现在让我们看一下如何调用service方法。在gRPC-Go，PRCs操作是在阻塞/同步模式，PRC操作需要等待server 的response，要么返回response要么返回error。

简单的RPC

调用简单的RPC GetFeature就像调用本地的函数一样简单。

feature, err := client.GetFeature(context.Background(), &pb.Point{409146138, -746188906})
if err != nil {
        ...
}

函数的参数为protocol buffer对象(point)和context.Context对象，通过context参数我们可以改变PRC的行为，比如超时/取消。如果调用没有返回错误，我们可以从第一个return值中读到response信息。

log.Println(feature)

server侧streaming RPC

这里我们调用server侧streaming的方法ListFeatures,该方法返回一串Feature。

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)
}

这里返回值是一个RouteGuide_ListFeaturesClient实例，客户端可以通过RouteGuide_ListFeaturesClient读取server的一系列返回值。
我们使用routeguide_ListFeaturesClient的recv()方法重复读取服务器的响应到protocol buffer对象，直到不再有消息：客户端需要检查每次调用后recv()返回的错误。如果为nil，可以继续读取；如果为io.eof，则消息流已结束；否则，必须存在一个通过err传递的rpc错误。

client侧streaming RPC

客户端流方法recordroute与服务器端方法类似，只是我们只传递context参数获取routeguide_recordroute client stream，我们可以使用它来写入和读取消息。

// 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返回的错误中获取RPC状态。如果状态为nil，那么closeAndrecv的第一个返回值将是有效的服务器响应。

双向streaming RPC

最后，让我们看看双向流式RPC RouteChat()。就像在recordroute中一样，我们只传递方法一个context对象，并返回一个可以用来写和读消息的流。但是，这一次我们通过自己的方法的stream返回response，而服务器仍在继续向其消息流写入消息。

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()方法。尽管每一方总是按照对方的消息被写入的顺序来获取对方的消息，但是客户机和服务器都可以以任何顺序读写——流完全独立地运行。

Try it out!

编译和运行服务器，假设您在文件夹中$GOPATH/src/google.golang.org/grpc/examples/route_guide:

$ go run server/server.go

同样，要运行客户机：

$ go run client/client.go

参考文档：https://grpc.io/docs/tutorials/basic/go.html#try-it-out