Python gRPC 基础教程

Python 中 gRPC 的基本教程介绍。

本教程提供了 Python 程序员使用 gRPC 的基本介绍。

通过浏览此示例，您将学习如何：

在 .proto 文件中定义服务。
使用协议缓冲区编译器生成服务器和客户端代码。
使用 Python gRPC API 为您的服务编写一个简单的客户端和服务器。
它假设您已经阅读了gRPC 简介并且熟悉协议缓冲区. 您可以在proto3 语言指南中找到更多信息和Python 生成的代码指南.

为什么要使用 gRPC？

我们的示例是一个简单的路线映射应用程序，它允许客户端获取有关其路线上的特征的信息，创建路线摘要，并与服务器和其他客户端交换路线信息，例如交通更新。

使用 gRPC，我们可以在一个.proto文件中定义我们的服务，并以 gRPC 支持的任何语言生成客户端和服务器，而这些客户端和服务器又可以在从大型数据中心内的服务器到您自己的平板电脑的环境中运行——所有通信的复杂性gRPC 为您处理不同的语言和环境。我们还获得了使用协议缓冲区的所有优势，包括高效的序列化、简单的 IDL 和轻松的接口更新。

示例代码和设置

本教程的示例代码位于 grpc/grpc/examples/python/route_guide. grpc要下载示例，请通过运行以下命令克隆存储库：

$ git clone -b v1.46.3 --depth 1 --shallow-submodules https://github.com/grpc/grpc
1

然后将当前目录更改examples/python/route_guide为存储库中：

$ cd grpc/examples/python/route_guide
1

您还应该安装相关工具来生成服务器和客户端接口代码 - 如果您还没有，请按照 快速入门中的设置说明进行操作。

定义服务

您的第一步（正如您从gRPC 简介中了解的那样）是使用协议缓冲区定义 gRPC服务以及方法请求和响应类型 . 您可以在中看到完整的 .proto 文件 examples/protos/route_guide.proto.

要定义服务，请service在 .proto 文件中指定一个命名：

service RouteGuide {
   // (Method definitions not shown)
}
1
2
3

然后rpc在服务定义中定义方法，指定它们的请求和响应类型。gRPC 让您定义四种服务方法，所有这些方法都在RouteGuide服务中使用：

一个简单的 RPC，其中客户端使用存根向服务器发送请求并等待响应返回，就像正常的函数调用一样。

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

一种响应流式 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) {}
1
2
3
4
5

一种请求流式 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) {}
1
2
3

双向流式 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) {}
1
2
3

您的.proto文件还包含我们服务方法中使用的所有请求和响应类型的协议缓冲区消息类型定义 - 例如，这是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;
}
1
2
3
4
5
6
7
8

生成客户端和服务器代码

接下来，您需要从 .proto 服务定义中生成 gRPC 客户端和服务器接口。

首先，安装 grpcio-tools 包：

$ pip install grpcio-tools
1

使用以下命令生成 Python 代码：

$ python -m grpc_tools.protoc -I../../protos --python_out=. --grpc_python_out=. ../../protos/route_guide.proto
1

请注意，由于我们已经在示例目录中提供了生成代码的一个版本，因此运行此命令会重新生成适当的文件，而不是创建一个新文件。生成的代码文件被调用 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
笔记
pb22中的 表示生成的代码遵循 Protocol Buffers Python API 版本 2。版本 1 已过时。它与协议缓冲区语言版本无关，协议缓冲区语言版本是由 .proto 文件指示syntax = "proto3"或syntax = "proto2"在 .proto 文件中指示的版本。

创建服务器

首先让我们看看如何创建RouteGuide服务器。如果您只对创建 gRPC 客户端感兴趣，您可以跳过本节并直接进入创建客户端（尽管您可能会觉得它很有趣！）。

创建和运行RouteGuide服务器分为两个工作项：

使用执行服务实际“工作”的功能实现从我们的服务定义生成的服务程序接口。
运行 gRPC 服务器以侦听来自客户端的请求并传输响应。
您可以在examples/python/route_guide/route_guide_server.pyRouteGuide中找到示例服务器 .

实施路线指南

route_guide_server.py有一个RouteGuideServicer子类生成的类route_guide_pb2_grpc.RouteGuideServicer：

# RouteGuideServicer provides an implementation of the methods of the RouteGuide service.
class RouteGuideServicer(route_guide_pb2_grpc.RouteGuideServicer):
1
2

RouteGuideServicer实现所有RouteGuide服务方法。

简单的 RPC

我们先来看最简单的类型GetFeature，它只是Point 从客户端获取 a ，并在 a 中从其数据库中返回相应的特征信息Feature。

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
1
2
3
4
5
6

该方法被传递一个route_guide_pb2.Point对 RPC 的请求，以及一个 grpc.ServicerContext提供 RPC 特定信息（例如超时限制）的对象。它返回一个route_guide_pb2.Feature响应。

响应流式 RPC
现在让我们看看下一个方法。ListFeatures是一个向客户端发送多个Features 的响应流式 RPC。

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

这里的请求消息是route_guide_pb2.Rectangle客户端想要在其中找到Features 的消息。该方法不是返回单个响应，而是产生零个或多个响应。

请求流式 RPC
request-streaming 方法RecordRoute使用 迭代器请求值并返回单个响应值。

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

双向流式 RPC
最后让我们看看双向流的方法RouteChat。

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)
1
2
3
4
5
6
7

该方法的语义是请求流方法和响应流方法的组合。它被传递一个请求值的迭代器，它本身就是一个响应值的迭代器。

启动服务器
实现所有RouteGuide方法后，下一步是启动 gRPC 服务器，以便客户端可以实际使用您的服务：

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()
1
2
3
4
5
6
7

服务器start()方法是非阻塞的。将实例化一个新线程来处理请求。在此期间，线程调用server.start()通常不会有任何其他工作要做。在这种情况下，您可以调用 server.wait_for_termination()干净地阻塞调用线程，直到服务器终止。

创建客户端
您可以在 examples/python/route_guide/route_guide_client.py中看到完整的示例客户端代码.

创建存根
要调用服务方法，我们首先需要创建一个存根。

我们实例化 从 .proto 生成RouteGuideStub的模块类。route_guide_pb2_grpc

channel = grpc.insecure_channel('localhost:50051')
stub = route_guide_pb2_grpc.RouteGuideStub(channel)
1
2

调用服务方法
对于返回单个响应的 RPC 方法（“响应一元”方法），gRPC Python 支持同步（阻塞）和异步（非阻塞）控制流语义。对于响应流 RPC 方法，调用会立即返回响应值的迭代器。调用该迭代器的next()方法阻塞，直到从迭代器产生的响应变为可用。

简单的 RPC
对简单 RPC 的同步调用GetFeature几乎与调用本地方法一样简单。RPC 调用等待服务器响应，并将返回响应或引发异常：

feature = stub.GetFeature(point)
1

异步调用GetFeature类似，但类似于在线程池中异步调用本地方法：

feature_future = stub.GetFeature.future(point)
feature = feature_future.result()
1
2

响应流式 RPC
调用 response-streamingListFeatures类似于使用序列类型：

for feature in stub.ListFeatures(rectangle):
1

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

route_summary = stub.RecordRoute(point_iterator)
route_summary_future = stub.RecordRoute.future(point_iterator)
route_summary = route_summary_future.result()
1
2
3

双向流式 RPC
调用双向流式传输RouteChat具有（就像服务端的情况一样）请求流式传输和响应流式传输语义的组合：

for received_route_note in stub.RouteChat(sent_route_note_iterator):
1

试试看！
运行服务器：

$ python route_guide_server.py
1

从不同的终端，运行客户端：

$ python route_guide_client.py
1