【golang】10、grpc server 和 client、proto

本文包括如下部分：

1. 定义proto文件
2. 编译出各编程语言代码
3. go的grpc程序

用idl定义的协议, 可用protoc工具生成不同编程语言实现的此协议

协议包括, 从proto buffers反序列化到struct, 和将struct序列化到proto buffers

client可以像在local一样调用remote的server

我们定义service, 及其methods, 在client有grpc的stub实现了定义的methods

grpc的优势

我们的示例是一个简单的路径映射应用程序，它允许客户端获取有关其路径上的要素的信息，创建其路径的摘要，并与服务器和其他客户端交换路径信息，如交通更新。
方便: 可跨端,跨语言通信, grpc会处理底层的一切
高效: protobuf序列化格式

下文以如下示例code讲解

git clone -b v1.46.0 --depth 1 https://github.com/grpc/grpc-go
cd grpc-go/examples/route_guide
1
2

定义proto文件

service RouteGuide {
    
}
1
2
3

然后在其中定义rpc方法

• 如下是client向server的同步调用

rpc GetFeature(Point) returns (Feature) {}
1

• 单向stream调用: 如下是client向server发请求, 获取到stream. client迭代stream直到无消息. 其server端定义如下

// 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) {}
1
2
3
4
5

• 双向stream调用: 双端各自都发送1个read-write stream, 2个流独立工作

// 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) {}
1
2
3

proto文件同时可定义protobuf数据结构

// 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;
}
1
2
3
4
5
6
7
8

生成client和server code

protoc -I=./ --go_out=plugins=grpc:./ ./*.proto 即可生成go语言代码，详见 proto 文档

protoc --go_out=. --go_opt=paths=source_relative \
    --go-grpc_out=. --go-grpc_opt=paths=source_relative \
    routeguide/route_guide.proto
1
2
3

即可在route_guide.pb.go中定义message, 在route_guide_grpc.pb.go 中定义 client需实现的interface, server需实现的interface

创建server

实现server的interface

创建一个实现service的interface的server, 并启动server使listen到client.

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 {
        ...
}
...
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

最简单的同步rpc方法

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{Location: point}, nil
}
1
2
3
4
5
6
7
8
9

server的stream rpc方法

server端将数据通过Send()发送

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
}
1
2
3
4
5
6
7
8
9
10

client的stream rpc方法

下例中, 客户端的流方法Recordroute，其中我们从客户端获得一个点流，并返回一个包含他们行程信息的RouteSummary.
该stream参数可Recv(),可SendAndClose()
若收到io.EOF则说明server已传输完毕

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
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

双向rpc

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
      }
    }
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

启动server

实现完方法后, 启动一个grpc server

flag.Parse()
lis, err := net.Listen("tcp", fmt.Sprintf("localhost:%d", *port))
if err != nil {
  log.Fatalf("failed to listen: %v", err)
}
var opts []grpc.ServerOption
...
grpcServer := grpc.NewServer(opts...)
pb.RegisterRouteGuideServer(grpcServer, newServer())
grpcServer.Serve(lis)
1
2
3
4
5
6
7
8
9
10

指定端口号, 创建grpc的server, 注册我们的server实现,调serve()方法阻塞等待

创建client

创建一个grpc的channel, 用于和server连接, 其中dialOption(可选)可设置鉴权参数(如TLS,GCE证书,JWT证书等)

var opts []grpc.DialOption
...
conn, err := grpc.Dial(*serverAddr, opts...)
if err != nil {
  ...
}
defer conn.Close()
1
2
3
4
5
6
7

再建立stub来执行rpc

client := pb.NewRouteGuideClient(conn)
1

在GRPC-GO中，RPC以阻塞/同步模式运行，这意味着RPC调用等待服务器响应，并将返回响应或错误。

最简单的rpc方法

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

client调用server端的stream 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)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

client端的stream rpc

// 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)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

client端的双向stream rpc

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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

参考

go grpc的基本概念

server 和 client 示例