一文看懂gRPC

gRPC 是 RPC 的一种，它使用protocol buffer(简称 Protobuf) 来对数据进行结构化，protocol buffer是来自 google 的序列化框架，比 JSON 更加轻便高效，基于 HTTP/2 标准设计，带来双向流、流控、头部压缩、单 TCP 连接上的多复用请求等特性。这些特性使得其在移动设备上表现更好，更省电和节省空间占用。

使用 gRPC，我们只需要在.proto文件中完成服务的定义，然后便可以在 gRPC 支持的任何语言中生成客户端和服务器的代码，这些客户端和服务器可以在从大型数据中心的服务器到自己的平板电脑等不同的环境中运行，gRPC 会自动处理不同语言和环境之间通信的复杂性。

gRPC核心概念

定义服务

gRPC 基于定义服务（service）的理念，需要在.proto扩展名的普通文本文件指定可以远程调用的方法及其参数和返回类型。服务可以包括多个 rpc 方法，其参数和返回类型均为protocol buffer消息（message）。 如下所示的 SayHello 方法使用 HelloRequest 作为参数，使用 HelloReply 作为返回值。

// The greeter service definition.
service Greeter {
    // Sends a greeting
    rpc SayHello (HelloRequest) returns (HelloReply) {}
}

// The request message containing the user's name.
message HelloRequest {
    string name = 1;
}

// The response message containing the greetings
message HelloReply {
    string message = 1;
}

gRPC中可以定义4种服务方法：

1. 单向 RPC，客户端向服务器发送单个请求并获得单个响应，就像普通的函数调用一样。

rpc SayHello(HelloRequest) returns (HelloResponse);

1. 服务器流式 RPC，客户端向服务器发送请求并获得一个流来读取一系列响应消息。客户端从返回的流中读取，直到没有更多消息。gRPC 保证单个 RPC 调用内的消息顺序。

rpc LotsOfReplies(HelloRequest) returns (stream HelloResponse);

1. 客户端流式 RPC，客户端写入一系列消息并将其发送到服务器，同样使用提供的流。客户端完成写入消息后，等待服务器读取消息并返回其响应。同样，gRPC 保证单个 RPC 调用内的消息顺序。

rpc LotsOfGreetings(stream HelloRequest) returns (HelloResponse);

1. 双向流式 RPC，双方使用读写流发送一系列消息。这两个流独立运行，因此客户端和服务器可以按照任何顺序读取和写入。例如，服务器可以在写入响应之前等待接收所有客户端消息，或者它可以交替读取消息然后写入消息，或者读取和写入的其他组合。每个流中的消息顺序保持不变。

rpc BidiHello(stream HelloRequest) returns (stream HelloResponse);

定义完服务之后，就需要定义和服务相关的消息（message），gRPC中的消息为protocol buffer数据结构，每个消息都是一个小型的逻辑信息记录，包含一系列称为字段的名称-值对。

message Person {
    string name = 1;
    int32 id = 2;
    bool has_ponycopter = 3;
}

生成代码

从.proto文件中的服务定义开始，gRPC 提供protocol buffer编译器，生成客户端和服务器端代码。gRPC 用户通常在客户端调用这些 API，并在服务器端实现相应的 API。

• 在服务器端，服务器实现服务中声明的方法，并运行一个 gRPC 服务器来处理来自客户端的调用请求。gRPC 会对来自客户端的调用请求进行解码、执行服务的方法，并对服务器的响应进行编码。
• 在客户端，客户端有一个称为存根（stub/client）的本地对象，和服务器端类似，它也实现了服务中声明的方法。客户端只需在本地对象上调用这些方法，这些方法会将调用的参数包装在适当的protocol buffer消息类型中，然后发送调用请求到服务器，并返回服务器的protocol buffer响应。

例如，使用protocol buffer编译器 protoc 从 proto 定义中生成指定的首选语言中的数据访问类。这些类提供了每个字段的简单访问器，例如name()和set_name()，以及将整个结构序列化为原始字节或者从原始字节解析出整个结构的方法。

gRPC的Python开发实践

开发目标

创建一个简单的地图应用，允许客户端获取关于路线的特征（Feature），路线的摘要（RouteSummary），并和服务器或其他客户端交换路线信息（交通状况等）。

定义服务

使用protocol buffer定义 gRPC 服务以及方法请求和响应类型：

service RouteGuide {
  
  // A simple RPC.
  // Obtains the feature at a given position.
  rpc GetFeature(Point) returns (Feature) {}
  
  // A server-to-client streaming RPC.
  // 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) {}

  // A client-to-server streaming RPC.
  // Accepts a stream of Points on a route being traversed, returning a
  // RouteSummary when traversal is completed.
  rpc RecordRoute(stream Point) returns (RouteSummary) {}

  // A Bidirectional streaming RPC.
  // 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消息类型定义：

// Points are represented as latitude-longitude pairs in the E7 representation
// 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;
}

// A latitude-longitude rectangle, represented as two diagonally opposite
// points "lo" and "hi".
message Rectangle {
  // One corner of the rectangle.
  Point lo = 1;

  // The other corner of the rectangle.
  Point hi = 2;
}

// A feature names something at a given point.
message Feature {
  // The name of the feature.
  string name = 1;

  // The point where the feature is detected.
  Point location = 2;
}

// A RouteSummary is received in response to a RecordRoute rpc.
// It contains the number of individual points received, the number of
// detected features, and the total distance covered as the cumulative sum of
// the distance between each point.
message RouteSummary {
  // The number of points received.
  int32 point_count = 1;

  // The number of known features passed while traversing the route.
  int32 feature_count = 2;

  // The distance covered in metres.
  int32 distance = 3;

  // The duration of the traversal in seconds.
  int32 elapsed_time = 4;
}

// A RouteNote is a message sent while at a given point.
message RouteNote {
  // The location from which the message is sent.
  Point location = 1;

  // The message to be sent.
  string message = 2;
}

生成代码

从.proto服务定义中生成 gRPC 客户端和服务器接口。

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. your_service.proto

生成的代码文件被称为 route_guide_pb2.py 和 route_guide_pb2_grpc.py，它们包含

• route_guide.proto 中定义的消息的类
• route_guide.proto 中定义的服务的类
  • RouteGuideStub，客户端可以使用它来调用 RouteGuide RPC
  • RouteGuideServicer，它定义了 RouteGuide 服务实现的接口
• route_guide.proto 中定义的服务的函数
  • add_RouteGuideServicer_to_server，它将 RouteGuideServicer 添加到 grpc.Server

实现服务器功能

创建和运行 RouteGuide 服务器可以分解为两部分内容：

1. 使用从服务定义生成的服务器接口，实现执行服务实际“工作”的函数。
2. 运行一个 gRPC 服务器来监听来自客户端的请求并传输响应。

route_guide_server.py 包含一个 RouteGuideServicer 类，它继承自生成的类 route_guide_pb2_grpc.RouteGuideServicer，用于实现所有的 RouteGuide 服务方法。

简单的单向 RPC

首先看看最简单的类型GetFeature，它从客户端获取一个 Point，并从其数据库中以 Feature 的形式返回相应的特征信息。 该方法接收 RPC 的 route_guide_pb2.Point 请求，以及一个提供 RPC 特定信息（例如超时限制）的 grpc.ServicerContext 对象，返回一个 route_guide_pb2.Feature 响应。

def get_feature(feature_db, point):
    """Returns Feature at given location or None."""
    for feature in feature_db:
        if feature.location == point:
            return feature
    return None

def GetFeature(self, request, context):
    feature = get_feature(self.db, request)
    if feature is None:
        return route_guide_pb2.Feature(name="", location=request)
    else:
        return feature

服务器流式 RPC

ListFeatures 是一个响应流式 RPC，它向客户端发送多个 Feature。它的请求消息是一个 route_guide_pb2.Rectangle，客户端希望在其中查找 Feature。该方法不返回单个响应，而是生成零个或多个响应。

def ListFeatures(self, request, context):
    left = min(request.lo.longitude, request.hi.longitude)
    right = max(request.lo.longitude, request.hi.longitude)
    top = max(request.lo.latitude, request.hi.latitude)
    bottom = min(request.lo.latitude, request.hi.latitude)
    for feature in self.db:
        if (
            feature.location.longitude >= left
            and feature.location.longitude <= right
            and feature.location.latitude >= bottom
            and feature.location.latitude <= top
        ):
            yield feature

客户端流式 RPC

请求流式方法 RecordRoute 使用请求值的迭代器Point，并返回单个响应值RouteSummary。

def RecordRoute(self, request_iterator, context):
    point_count = 0
    feature_count = 0
    distance = 0.0
    prev_point = None

    start_time = time.time()
    for point in request_iterator:
        point_count += 1
        if get_feature(self.db, point):
            feature_count += 1
        if prev_point:
            distance += get_distance(prev_point, point)
        prev_point = point

    elapsed_time = time.time() - start_time
    return route_guide_pb2.RouteSummary(
        point_count=point_count,
        feature_count=feature_count,
        distance=int(distance),
        elapsed_time=int(elapsed_time),
    )

双向流式 RPC

RouteChat 是请求流方法和响应流方法的组合。它接收一个请求值的迭代器RouteNote，返回一个响应值的迭代器RouteNote。

def RouteChat(self, request_iterator, context):
    prev_notes = []
    for new_note in request_iterator:
        for prev_note in prev_notes:
            if prev_note.location == new_note.location:
                yield prev_note
        prev_notes.append(new_note)

启动和运行服务器

在实现完所有 RouteGuide 方法后，下一步是启动一个 gRPC 服务器，以便客户端可以使用服务器提供的服务。

服务器start()方法是非阻塞的，一个新线程将会被实例化并用来处理请求，这个线程通常在此期间没有其他工作要做。在这个例子中，可以通过调用server.wait_for_termination()来阻塞该线程，直到服务器终止。

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    route_guide_pb2_grpc.add_RouteGuideServicer_to_server(RouteGuideServicer(), server)
    server.add_insecure_port("[::]:50051")
    server.start()
    server.wait_for_termination()

if __name__ == "__main__":
    logging.basicConfig()
    serve()

实现客户端功能

创建stub

实际调用服务器的方法时，我们首先需要创建一个stub。实例化 route_guide_pb2_grpc 模块的 RouteGuideStub 类。

with grpc.insecure_channel("localhost:50051") as channel:
    stub = route_guide_pb2_grpc.RouteGuideStub(channel)

调用服务器的方法

对简单 RPC GetFeature 的同步调用与调用本地方法一样简单。RPC 调用将等待服务器响应，并返回正常响应值或抛出异常值。

对 GetFeature 的异步调用和在线程池中异步调用本地方法类似。

# 同步
feature = stub.GetFeature(point)
# 异步
feature_future = stub.GetFeature.future(point)
feature = feature_future.result()

调用响应流 ListFeatures 类似于使用序列类型。

for feature in stub.ListFeatures(rectangle):
    print("Feature: %s" % feature)

调用请求流 RecordRoute 类似于将迭代器传递给本地方法。像上面也返回单个响应的简单 RPC 一样，它可以同步或异步调用。

# 同步
route_summary = stub.RecordRoute(point_iterator)
# 异步
route_summary_future = stub.RecordRoute.future(point_iterator)
route_summary = route_summary_future.result()

调用双向流 RouteChat 具有（与服务端一样）请求流和响应流语义的组合。

for received_route_note in stub.RouteChat(sent_route_note_iterator):
    print("Note: %s" % received_route_note)