Начало работы с gRPC-Python — Потоковая передача

1. Введение

В этом практическом занятии вы будете использовать gRPC-Python для создания клиента и сервера, которые составляют основу приложения для построения маршрутов, написанного на Python.

К концу этого руководства у вас будет клиент, который подключается к удаленному серверу с помощью gRPC для получения информации о характеристиках маршрута клиента, создания сводки маршрута клиента и обмена информацией о маршруте, такой как обновления трафика, с сервером и другими клиентами.

Сервис определяется в файле Protocol Buffers, который будет использоваться для генерации шаблонного кода для клиента и сервера, чтобы они могли взаимодействовать друг с другом, экономя ваше время и усилия на реализации этой функциональности.

Сгенерированный код учитывает не только сложности взаимодействия между сервером и клиентом, но и сериализацию и десериализацию данных.

Что вы узнаете

  • Как использовать Protocol Buffers для определения API сервиса.
  • Как создать клиент и сервер на основе gRPC из определения Protocol Buffers с помощью автоматической генерации кода.
  • Понимание принципов потокового взаимодействия между клиентом и сервером с использованием gRPC.

Данный практический семинар предназначен для разработчиков на Python, которые только начинают работать с gRPC или хотят освежить свои знания gRPC, а также для всех, кто заинтересован в создании распределенных систем. Предварительный опыт работы с gRPC не требуется.

2. Прежде чем начать

Что вам понадобится

  • Python 3.9 или выше. Мы рекомендуем Python 3.13. Инструкции по установке для конкретной платформы см. в разделе «Настройка и использование Python» . В качестве альтернативы можно установить Python, не являющийся системным, используя такие инструменты, как uv или pyenv .
  • Используйте pip для установки пакетов Python.
  • venv для создания виртуальных сред Python.

Пакеты ensurepip и venv являются частью стандартной библиотеки Python и обычно доступны по умолчанию.

Однако некоторые дистрибутивы на основе Debian (включая Ubuntu) исключают их при распространении Python. Для установки пакетов выполните следующую команду:

sudo apt install python3-pip python3-venv

Получите код

Для упрощения обучения в этом практическом занятии предлагается готовый шаблон исходного кода, который поможет вам начать работу. Следующие шаги проведут вас через завершение приложения, включая генерацию кода gRPC с использованием плагина компилятора Protocol Buffer grpc_tools.protoc .

grpc-codelabs

Исходный код для этого практического занятия находится в каталоге codelabs/grpc-python-streaming/start_here . Если вы предпочитаете не реализовывать код самостоятельно, готовый исходный код доступен в каталоге completed .

Сначала создайте рабочую директорию codelab и перейдите в неё с помощью команды cd:

mkdir grpc-python-streaming && cd grpc-python-streaming

Скачайте и распакуйте CodeLab:

curl -sL https://github.com/grpc-ecosystem/grpc-codelabs/archive/refs/heads/v1.tar.gz \
  | tar xvz --strip-components=4 \
  grpc-codelabs-1/codelabs/grpc-python-streaming/start_here

В качестве альтернативы вы можете скачать ZIP-архив, содержащий только папку codelab, и распаковать его вручную.

3. Определите сообщения и сервисы.

Первым шагом является определение gRPC-сервиса приложения, его RPC-метода, а также типов сообщений запроса и ответа с помощью Protocol Buffers . Ваш сервис будет предоставлять:

  • RPC-методы ListFeatures , RecordRoute и RouteChat , которые реализует сервер, а вызывает клиент.
  • Типы сообщений Point , Feature , Rectangle , RouteNote и RouteSummary представляют собой структуры данных, которыми обмениваются клиент и сервер при вызове методов RPC.

Все эти RPC-методы и типы сообщений для них будут определены в файле protos/route_guide.proto предоставленного исходного кода.

Протоколы Protocol Buffers обычно называются protobufs. Для получения дополнительной информации о терминологии gRPC см. раздел «Основные концепции, архитектура и жизненный цикл gRPC».

Определите типы сообщений

В файле protos/route_guide.proto исходного кода сначала определите тип сообщения Point . Point представляет собой пару координат широты и долготы на карте. Для этого практического задания используйте целые числа для координат:

message Point {
  int32 latitude = 1;
  int32 longitude = 2;
}

Цифры 1 и 2 — это уникальные идентификационные номера для каждого поля в структуре message .

Далее определите тип сообщения Feature . В Feature используется string поле для имени или почтового адреса объекта, расположенного в точке Point :

message Feature {
  // The name or address of the feature.
  string name = 1;

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

Для того чтобы можно было передавать данные из нескольких точек в пределах одной области клиенту, вам потребуется сообщение Rectangle , представляющее собой прямоугольник широты и долготы, отображаемый в виде двух точек lo и hi расположенных по диагонали друг от друга:

message Rectangle {
  // One corner of the rectangle.
  Point lo = 1;

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

Также сообщение RouteNote , представляющее собой сообщение, отправленное в заданной точке:

message RouteNote {
  // The location from which the message is sent.
  Point location = 1;

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

Наконец, вам понадобится сообщение RouteSummary . Это сообщение принимается в ответ на RPC-запрос RecordRoute , который описан в следующем разделе. Оно содержит количество полученных отдельных точек, количество обнаруженных объектов и общее пройденное расстояние как сумма расстояний между каждой точкой.

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

Определите методы обслуживания

Для определения сервиса необходимо указать именованный сервис в файле .proto . Файл route_guide.proto содержит структуру service с именем RouteGuide , которая определяет один или несколько методов, предоставляемых сервисом приложения.

При определении методов RPC внутри определения сервиса вы указываете их типы запроса и ответа. В этом разделе практического занятия давайте определим:

ListFeatures

Получает доступные объекты Feature внутри заданного Rectangle . Результаты передаются потоком, а не возвращаются сразу, поскольку прямоугольник может занимать большую площадь и содержать огромное количество объектов.

Для этого приложения вы будете использовать потоковый RPC на стороне сервера : клиент отправляет запрос на сервер и получает поток для чтения последовательности сообщений. Клиент читает из возвращенного потока до тех пор, пока не закончатся сообщения. Как вы можете видеть в нашем примере, вы указываете метод потоковой передачи на стороне сервера, помещая ключевое слово stream перед типом ответа.

rpc ListFeatures(Rectangle) returns (stream Feature) {}

RecordRoute

Принимает поток точек на проходимом маршруте и возвращает RouteSummary после завершения обхода.

В данном случае подходит RPC -запрос с потоковой передачей на стороне клиента : клиент записывает последовательность сообщений и отправляет их на сервер, снова используя предоставленный поток. После завершения записи сообщений клиент ожидает, пока сервер прочтет их все и вернет свой ответ. Метод потоковой передачи на стороне клиента указывается путем добавления ключевого слова stream перед типом запроса. 

rpc RecordRoute(stream Point) returns (RouteSummary) {}

RouteChat

Принимает поток RouteNotes , отправляемых во время прохождения маршрута, одновременно получая другие RouteNotes (например, от других пользователей).

Это именно тот случай, когда может пригодиться двунаправленная потоковая передача . Двунаправленный потоковый RPC-вызов, в котором обе стороны отправляют последовательность сообщений, используя поток чтения и записи. Два потока работают независимо, поэтому клиенты и серверы могут читать и записывать в любом порядке: например, сервер может дождаться получения всех сообщений от клиентов, прежде чем записывать свои ответы, или он может поочередно читать сообщение, а затем записывать сообщение, или использовать какую-либо другую комбинацию чтения и записи. Порядок сообщений в каждом потоке сохраняется. Этот тип метода указывается путем размещения ключевого слова stream перед запросом и ответом. 

rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}

4. Сгенерируйте код клиента и сервера.

Далее, используя компилятор Protocol Buffer, сгенерируйте шаблонный gRPC-код для клиента и сервера из файла .proto .

Для генерации кода gRPC на Python мы создали grpcio-tools . Он включает в себя:

  1. Стандартный компилятор protoc , генерирующий код Python на основе определений message .
  2. Плагин gRPC protobuf, генерирующий код Python (заглушки для клиента и сервера) на основе определений service .

Мы установим пакет grpcio-tools для Python с помощью pip. Давайте создадим новое виртуальное окружение Python (venv), чтобы изолировать зависимости вашего проекта от системных пакетов: 

python3 -m venv --upgrade-deps .venv

Для активации виртуального окружения в оболочке bash/zsh: 

source .venv/bin/activate

Для Windows и нестандартных оболочек см. таблицу по адресу https://docs.python.org/3/library/venv.html#how-venvs-work .

Далее установите grpcio-tools (это также установит пакет grpcio ): 

pip install grpcio-tools

Для генерации шаблонного кода на Python используйте следующую команду: 

python -m grpc_tools.protoc --proto_path=./protos  \
 --python_out=. --pyi_out=. --grpc_python_out=. \
 ./protos/route_guide.proto

В результате будут сгенерированы следующие файлы для интерфейсов, определенных в route_guide.proto :

  1. route_guide_pb2.py содержит код , который динамически создает классы, генерируемые на основе определений message .
  2. route_guide_pb2.pyi — это «файл-заглушка» или «файл подсказок типов», созданный на основе определений message . Он содержит только сигнатуры без какой-либо реализации. Файлы-заглушки могут использоваться IDE для улучшения автозавершения кода и обнаружения ошибок.
  3. route_guide_pb2_grpc.py генерируется на основе определений service и содержит классы и функции, специфичные для gRPC.

Специализированный код gRPC содержит:

  1. RouteGuideStub , который может использоваться gRPC-клиентом для вызова RPC-вызовов RouteGuide.
  2. RouteGuideServicer , который определяет интерфейс для реализаций сервиса RouteGuide .
  3. Функция add_RouteGuideServicer_to_server используется для регистрации RouteGuideServicer на gRPC-сервере .

5. Создайте сервер.

Для начала давайте рассмотрим, как создать сервер RouteGuide . Создание и запуск сервера RouteGuide состоит из двух этапов:

  • Реализация интерфейса сервиса, сгенерированного на основе определения нашего сервиса, с функциями, которые выполняют фактическую «работу» сервиса.
  • Запуск gRPC-сервера для приема запросов от клиентов и передачи ответов.

Давайте посмотрим на route_guide_server.py .

Внедрить RouteGuide

В 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):

RouteGuideServicer реализует все методы сервиса RouteGuide .

Серверная потоковая передача RPC

ListFeatures — это RPC-вызов с потоковой передачей ответов, который отправляет клиенту несколько Feature : 

def ListFeatures(self, request, context):
    """List all features contained within the given Rectangle."""
    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:
        lat, lng = feature.location.latitude, feature.location.longitude
        if left <= lng <= right and bottom <= lat <= top:
            yield feature

В данном случае запрос представляет собой объект route_guide_pb2.Rectangle , внутри которого клиент хочет найти объекты Feature . Вместо возврата одного ответа метод выдает ноль или более ответов.

RPC потоковой передачи на стороне клиента

Метод потоковой передачи запросов RecordRoute использует итератор значений запроса и возвращает одно значение ответа. 

def RecordRoute(self, request_iterator, context):
    """Calculate statistics about the trip composed of Points."""
    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

Наконец, давайте рассмотрим наш двунаправленный потоковый RPC RouteChat() : 

def RouteChat(self, request_iterator, context):
    """
    Receive a stream of message/location pairs, and responds with
    a stream of all previous messages for the given location.
    """
    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-сервера, чтобы клиенты могли фактически использовать ваш сервис: 

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    route_guide_pb2_grpc.add_RouteGuideServicer_to_server(
        RouteGuideServicer(),
        server,
    )
    listen_addr = "localhost:50051"
    server.add_insecure_port(listen_addr)
    print(f"Starting server on {listen_addr}")
    server.start()
    server.wait_for_termination()

Метод start() сервера является неблокирующим. Для обработки запросов будет создан новый поток. Поток, вызывающий server.start() , часто не имеет других задач в это время. В этом случае можно вызвать server.wait_for_termination() , чтобы корректно заблокировать вызывающий поток до завершения работы сервера.

6. Создайте клиента.

Давайте посмотрим на route_guide_client.py .

Создать заглушку

Для вызова методов сервиса нам сначала нужно создать заглушку .

Мы создаём экземпляр класса RouteGuideStub модуля route_guide_pb2_grpc , сгенерированного из нашего .proto. В методе run() : 

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

Обратите внимание, что здесь channel используется в качестве менеджера контекста и будет автоматически закрыт, как только интерпретатор покинет блок with .

Методы вызова сервиса

Для методов RPC, возвращающих один ответ («одномерные» методы), gRPC Python поддерживает как синхронную (блокирующую), так и асинхронную (неблокирующую) семантику управления потоком. Для методов RPC с потоковой передачей ответов вызовы немедленно возвращают итератор значений ответа. Вызовы метода next() этого итератора блокируются до тех пор, пока не станет доступен ответ, который должен быть получен от итератора.

Серверная потоковая передача RPC

Вызов функции ListFeatures для потоковой передачи ответов аналогичен работе с типами последовательностей: 

def guide_list_features(stub):
    _lo = route_guide_pb2.Point(latitude=400000000, longitude=-750000000)
    _hi = route_guide_pb2.Point(latitude=420000000, longitude=-730000000)
    rectangle = route_guide_pb2.Rectangle(
        lo=_lo,
        hi=_hi,
    )
    print("Looking for features between 40, -75 and 42, -73")

    features = stub.ListFeatures(rectangle)
    for feature in features:
        print(
            f"Feature called '{feature.name}'"
            f" at {format_point(feature.location)}"
        )

RPC потоковой передачи на стороне клиента

Вызов функции RecordRoute для потоковой передачи запросов аналогичен передаче итератора локальному методу. Как и простой RPC-вызов, описанный выше, который также возвращает один ответ, его можно вызывать синхронно: 

def guide_record_route(stub):
    feature_list = route_guide_resources.read_route_guide_database()
    route_iterator = generate_route(feature_list)

    route_summary = stub.RecordRoute(route_iterator)
    print(f"Finished trip with {route_summary.point_count} points")
    print(f"Passed {route_summary.feature_count} features")
    print(f"Traveled {route_summary.distance} meters")
    print(f"It took {route_summary.elapsed_time} seconds")

Двунаправленный потоковый RPC

Вызов RouteChat с двунаправленной потоковой передачей данных (как и на стороне сервиса) сочетает в себе семантику потоковой передачи запросов и потоковой передачи ответов.

Сгенерируйте запросы и отправляйте их по одному, используя yield . 

def generate_notes():
    home = route_guide_pb2.Point(latitude=1, longitude=1)
    work = route_guide_pb2.Point(latitude=2, longitude=2)
    notes = [
        make_route_note("Departing from home", home),
        make_route_note("Arrived at work", work),
        make_route_note("Having lunch at work", work),
        make_route_note("Departing from work", work),
        make_route_note("Arrived home", home),
    ]
    for note in notes:
        print(
            f"Sending RouteNote for {format_point(note.location)}:"
            f" {note.message}"
        )
        yield note
        # Sleep to simulate moving from one point to another.
        # Only for demonstrating the order of the messages.
        time.sleep(0.1)

Получение и обработка ответов от сервера: 

def guide_route_chat(stub):
    responses = stub.RouteChat(generate_notes())
    for response in responses:
        print(
            "< Found previous note at"
            f" {format_point(response.location)}: {response.message}"
        )

Вызов вспомогательных методов

В функции run выполните только что созданные методы и передайте им stub . 

print("-------------- ListFeatures --------------")
guide_list_features(stub)
print("-------------- RecordRoute --------------")
guide_record_route(stub)
print("-------------- RouteChat --------------")
guide_route_chat(stub)

7. Попробуйте.

Запустите сервер: 

python route_guide_server.py

С другого терминала снова активируйте виртуальное окружение ( source .venv/bin/activate) , затем запустите клиент: 

python route_guide_client.py

Давайте посмотрим на результат.

ListFeatures

Сначала вы увидите список объектов. Каждый объект передается с сервера ( серверная потоковая передача RPC ) по мере того, как сервер обнаруживает его в пределах запрошенного прямоугольника: 

-------------- ListFeatures --------------
Looking for features between 40, -75 and 42, -73
Feature called 'Patriots Path, Mendham, NJ 07945, USA' at (lat=407838351, lng=-746143763)
Feature called '101 New Jersey 10, Whippany, NJ 07981, USA' at (lat=408122808, lng=-743999179)
Feature called 'U.S. 6, Shohola, PA 18458, USA' at (lat=413628156, lng=-749015468)
Feature called '5 Conners Road, Kingston, NY 12401, USA' at (lat=419999544, lng=-740371136)
...

RecordRoute

Во-вторых, RecordRoute демонстрирует список случайно посещенных точек, передаваемый потоком с клиента на сервер (потоковая RPC-передача на стороне клиента) : 

-------------- RecordRoute --------------
Visiting point (lat=410395868, lng=-744972325)
Visiting point (lat=404310607, lng=-740282632)
Visiting point (lat=403966326, lng=-748519297)
Visiting point (lat=407586880, lng=-741670168)
Visiting point (lat=406589790, lng=-743560121)
Visiting point (lat=410322033, lng=-747871659)
Visiting point (lat=415464475, lng=-747175374)
Visiting point (lat=407586880, lng=-741670168)
Visiting point (lat=402647019, lng=-747071791)
Visiting point (lat=414638017, lng=-745957854)

После того, как клиент завершит потоковую передачу всех посещенных точек, он получит от сервера ответ без потоковой передачи ( односторонний RPC ). Этот ответ будет содержать сводку вычислений, выполненных по всему маршруту клиента. 

Finished trip with 10 points
Passed 10 features
Traveled 654743 meters
It took 0 seconds

RouteChat

Наконец, вывод RouteChat демонстрирует двустороннюю потоковую передачу . Когда клиент «посещает» home или work точку, он записывает заметку для этой точки, отправляя RouteNote на сервер. Если точка уже была посещена, сервер возвращает все предыдущие заметки для этой точки. 

-------------- RouteChat --------------
Sending RouteNote for (lat=1, lng=1): Departing from home
Sending RouteNote for (lat=2, lng=2): Arrived at work
Sending RouteNote for (lat=2, lng=2): Having lunch at work
< Found previous note at (lat=2, lng=2): Arrived at work
Sending RouteNote for (lat=2, lng=2): Departing from work
< Found previous note at (lat=2, lng=2): Arrived at work
< Found previous note at (lat=2, lng=2): Having lunch at work
Sending RouteNote for (lat=1, lng=1): Arrived home
< Found previous note at (lat=1, lng=1): Departing from home

8. Что дальше?