Pierwsze kroki z gRPC-Python – przesyłanie strumieniowe

1. Wprowadzenie

W tym module nauczysz się tworzyć za pomocą gRPC-Python klienta i serwera, które stanowią podstawę aplikacji do mapowania tras napisanej w Pythonie.

Po ukończeniu tego samouczka będziesz mieć klienta, który łączy się z serwerem zdalnym za pomocą gRPC, aby uzyskiwać informacje o funkcjach na trasie klienta, tworzyć podsumowanie trasy klienta i wymieniać informacje o trasie, takie jak aktualizacje ruchu, z serwerem i innymi klientami.

Usługa jest zdefiniowana w pliku Protocol Buffers, który będzie używany do generowania kodu szablonowego dla klienta i serwera, aby mogły się ze sobą komunikować. Dzięki temu zaoszczędzisz czas i wysiłek potrzebny na wdrożenie tej funkcji.

Wygenerowany kod obsługuje nie tylko złożoność komunikacji między serwerem a klientem, ale także serializację i deserializację danych.

Czego się nauczysz

  • Jak używać buforów protokołu do definiowania interfejsu API usługi.
  • Jak utworzyć klienta i serwer oparte na gRPC na podstawie definicji Protocol Buffers za pomocą automatycznego generowania kodu.
  • znajomość komunikacji strumieniowej klient-serwer za pomocą gRPC;

Te warsztaty są przeznaczone dla programistów Pythona, którzy dopiero zaczynają korzystać z gRPC lub chcą sobie przypomnieć podstawy tej technologii, a także dla wszystkich innych osób zainteresowanych tworzeniem systemów rozproszonych. Nie musisz mieć wcześniejszego doświadczenia z gRPC.

2. Zanim zaczniesz

Czego potrzebujesz

  • Python w wersji 3.9 lub nowszej. Zalecamy Pythona 3.13. Instrukcje instalacji na poszczególnych platformach znajdziesz w artykule Python Setup and Usage (Konfigurowanie i używanie Pythona). Możesz też zainstalować Pythona, który nie jest częścią systemu, za pomocą narzędzi takich jak uv lub pyenv.
  • pip do instalowania pakietów Pythona.
  • venv do tworzenia wirtualnych środowisk Pythona.

Pakiety ensurepip i venv są częścią standardowej biblioteki Pythona i zazwyczaj są dostępne domyślnie.

Niektóre dystrybucje oparte na Debianie (w tym Ubuntu) decydują się jednak na wykluczenie ich podczas redystrybucji Pythona. Aby zainstalować pakiety, uruchom to polecenie:

sudo apt install python3-pip python3-venv

Pobierz kod

Aby ułatwić Ci naukę, w tym ćwiczeniu w Codelabs znajdziesz gotowy szkielet kodu źródłowego, który pomoże Ci zacząć. Poniższe kroki pomogą Ci wypełnić aplikację, w tym wygenerować kod gRPC za pomocą wtyczki kompilatora grpc_tools.protoc Protocol Buffer.

grpc-codelabs

Kod źródłowy szkieletu tego modułu jest dostępny w katalogu codelabs/grpc-python-streaming/start_here. Jeśli nie chcesz samodzielnie implementować kodu, gotowy kod źródłowy znajdziesz w katalogu completed.

Najpierw utwórz katalog roboczy z ćwiczeniami i przejdź do niego:

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

Pobierz i rozpakuj ćwiczenia z programowania:

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

Możesz też pobrać plik ZIP zawierający tylko katalog z instrukcjami i rozpakować go ręcznie.

3. Określanie komunikatów i usług

Pierwszym krokiem jest zdefiniowanie usługi gRPC aplikacji, jej metody RPC oraz typów wiadomości żądania i odpowiedzi za pomocą buforów protokołu. Twoja usługa będzie zapewniać:

  • Metody RPC wywoływane przez serwer i klienta: ListFeatures, RecordRoute i RouteChat.
  • Typy wiadomości Point, Feature, Rectangle, RouteNote i RouteSummary, czyli struktury danych wymieniane między klientem a serwerem podczas wywoływania metod RPC.

Te metody RPC i ich typy wiadomości będą zdefiniowane w pliku protos/route_guide.proto podanego kodu źródłowego.

Protokoły buforów są powszechnie znane jako protobufy. Więcej informacji o terminologii gRPC znajdziesz w artykule Podstawowe koncepcje, architektura i cykl życia.

Określanie typów wiadomości

W protos/route_guide.proto pliku kodu źródłowego najpierw zdefiniuj Point typ wiadomości. Symbol Point reprezentuje parę współrzędnych szerokości i długości geograficznej na mapie. W tym ćwiczeniu używaj liczb całkowitych jako współrzędnych:

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

Numery 1 i 2 to unikalne identyfikatory poszczególnych pól w strukturze message.

Następnie określ Featuretyp wiadomości. Feature używa pola string na nazwę lub adres pocztowy czegoś w lokalizacji określonej przez Point:

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

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

Aby można było przesyłać strumieniowo do klienta wiele punktów w danym obszarze, potrzebujesz wiadomości Rectangle, która reprezentuje prostokąt o określonych współrzędnych geograficznych, przedstawiony jako 2 przeciwległe punkty po przekątnej lo i hi:

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

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

Oprócz tego wiadomość RouteNote, która reprezentuje wiadomość wysłaną w danym momencie:

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

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

Na koniec potrzebujesz wiadomości RouteSummary. Ta wiadomość jest odbierana w odpowiedzi na wywołanie RPC RecordRoute, które opisujemy w następnej sekcji. Zawiera liczbę otrzymanych punktów, liczbę wykrytych cech i całkowity przebyty dystans jako sumę odległości między poszczególnymi punktami.

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

Określanie metod usługi

Aby zdefiniować usługę, w pliku .proto podaj jej nazwę. Plik route_guide.proto ma strukturę service o nazwie RouteGuide, która definiuje co najmniej 1 metodę udostępnianą przez usługę aplikacji.

Gdy w definicji usługi określasz RPC metody, musisz podać typy żądań i odpowiedzi. W tej części ćwiczenia zdefiniujemy:

ListFeatures

Pobiera obiekty Feature dostępne w ramach danego Rectangle. Wyniki są przesyłane strumieniowo, a nie zwracane od razu, ponieważ prostokąt może obejmować duży obszar i zawierać ogromną liczbę obiektów.

W tej aplikacji użyjesz strumieniowego wywołania RPC po stronie serwera: klient wysyła żądanie do serwera i otrzymuje strumień, z którego może odczytywać sekwencję wiadomości. Klient odczytuje zwrócony strumień, dopóki nie będzie już żadnych wiadomości. Jak widać w naszym przykładzie, metodę przesyłania strumieniowego po stronie serwera określa się, umieszczając słowo kluczowe stream przed typem odpowiedzi.

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

RecordRoute

Akceptuje strumień punktów na pokonywanej trasie i zwraca RouteSummary po zakończeniu pokonywania trasy.

W tym przypadku odpowiednie jest wywołanie RPC strumieniowania po stronie klienta: klient zapisuje sekwencję wiadomości i wysyła je na serwer, ponownie używając udostępnionego strumienia. Gdy klient skończy pisać wiadomości, czeka, aż serwer je wszystkie odczyta i zwróci odpowiedź. Metodę przesyłania strumieniowego po stronie klienta określa się, umieszczając słowo kluczowe stream przed typem żądania.

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

RouteChat

Akceptuje strumień RouteNotes wysyłanych podczas pokonywania trasy, a także inne RouteNotes (np. od innych użytkowników).

To jest właśnie przypadek użycia strumieniowania dwukierunkowego. Dwukierunkowe strumieniowe wywołanie RPC, w którym obie strony wysyłają sekwencję wiadomości za pomocą strumienia odczytu i zapisu. Oba strumienie działają niezależnie, więc klienci i serwery mogą odczytywać i zapisywać dane w dowolnej kolejności. Na przykład serwer może czekać na otrzymanie wszystkich wiadomości od klienta przed zapisaniem odpowiedzi lub może odczytać wiadomość, a następnie zapisać wiadomość albo użyć innej kombinacji odczytów i zapisów. Kolejność wiadomości w każdym strumieniu jest zachowana. Ten typ metody określa się, umieszczając słowo kluczowe stream przed żądaniem i odpowiedzią.

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

4. Generowanie kodu klienta i serwera

Następnie wygeneruj kod gRPC dla klienta i serwera z pliku .proto za pomocą kompilatora bufora protokołu.

Do generowania kodu Pythona gRPC stworzyliśmy narzędzie grpcio-tools. Usługa ta obejmuje:

  1. Zwykły kompilator protoc, który generuje kod Pythona z definicji message.
  2. wtyczka gRPC protobuf, która generuje kod Pythona (namiastki klienta i serwera) z definicji service;

Zainstalujemy pakiet Pythona grpcio-tools za pomocą narzędzia pip. Utwórz nowe wirtualne środowisko Pythona (venv), aby odizolować zależności projektu od pakietów systemowych:

python3 -m venv --upgrade-deps .venv

Aby aktywować środowisko wirtualne w powłoce bash/zsh:

source .venv/bin/activate

W przypadku systemu Windows i niestandardowych powłok zobacz tabelę na stronie https://docs.python.org/3/library/venv.html#how-venvs-work.

Następnie zainstaluj grpcio-tools (spowoduje to również zainstalowanie pakietu grpcio):

pip install grpcio-tools

Aby wygenerować kod początkowy w Pythonie, użyj tego polecenia:

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

Spowoduje to wygenerowanie tych plików dla interfejsów zdefiniowanych w route_guide.proto:

  1. route_guide_pb2.py zawiera kod dynamicznie tworzący klasy wygenerowane na podstawie definicji message.
  2. route_guide_pb2.pyi to „plik stub” lub „plik wskazówek dotyczących typu” wygenerowany na podstawie definicji message. Zawiera tylko sygnatury bez implementacji. Pliki stub mogą być używane przez środowiska IDE, aby zapewnić lepsze autouzupełnianie i wykrywanie błędów.
  3. route_guide_pb2_grpc.py jest generowany na podstawie definicji service i zawiera klasy oraz funkcje specyficzne dla gRPC.

Kod gRPC zawiera:

  1. RouteGuideStub, z którego klient gRPC może korzystać do wywoływania wywołań RPC RouteGuide.
  2. RouteGuideServicer, który definiuje interfejs implementacji usługi RouteGuide.
  3. add_RouteGuideServicer_to_server, która służy do rejestrowania RouteGuideServicer na serwerze gRPC.

5. Tworzenie serwera

Najpierw zobaczmy, jak utworzyć RouteGuideserwerRouteGuide. Tworzenie i uruchamianie serwera RouteGuide dzieli się na 2 etapy:

  • Wdrożenie interfejsu usługi wygenerowanego na podstawie definicji usługi za pomocą funkcji, które wykonują rzeczywistą „pracę” usługi.
  • Uruchomienie serwera gRPC, który będzie nasłuchiwać żądań od klientów i przesyłać odpowiedzi.

Przyjrzyjmy się route_guide_server.py.

Implementacja RouteGuide

Klasa route_guide_server.py ma klasę RouteGuideServicer, która jest podklasą wygenerowanej klasy route_guide_pb2_grpc.RouteGuideServicer:

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

RouteGuideServicer implementuje wszystkie metody usługi RouteGuide.

RPC przesyłania strumieniowego po stronie serwera

ListFeatures to RPC przesyłający strumieniowo odpowiedzi, który wysyła do klienta wiele obiektów 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

Wiadomość z żądaniem to route_guide_pb2.Rectangle, w której klient chce znaleźć Feature. Zamiast zwracać jedną odpowiedź, metoda zwraca zero lub więcej odpowiedzi.

RPC przesyłania strumieniowego po stronie klienta

Metoda przesyłania strumieniowego żądań RecordRoute używa iteratora wartości żądań i zwraca pojedynczą wartość odpowiedzi.

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

Dwukierunkowe strumieniowe RPC

Na koniec przyjrzyjmy się dwukierunkowemu strumieniowemu wywołaniu 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)

Semantyka tej metody jest połączeniem semantyki metody strumieniowania żądań i metody strumieniowania odpowiedzi. Przyjmuje iterator wartości żądań i jest iteratorem wartości odpowiedzi.

Uruchamianie serwera

Po zaimplementowaniu wszystkich RouteGuide metod następnym krokiem jest uruchomienie serwera gRPC, aby klienci mogli korzystać z Twojej usługi:

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

Metoda serwera start() nie blokuje wątku. Do obsługi żądań zostanie utworzony nowy wątek. Wątek wywołujący funkcję server.start() często nie ma w tym czasie nic innego do zrobienia. W takim przypadku możesz wywołać funkcję server.wait_for_termination(), aby prawidłowo zablokować wątek wywołujący do momentu zakończenia działania serwera.

6. Tworzenie klienta

Przyjrzyjmy się route_guide_client.py.

Tworzenie stubu

Aby wywoływać metody usługi, musimy najpierw utworzyć stub.

Tworzymy instancję klasy RouteGuideStub modułu route_guide_pb2_grpc wygenerowanego na podstawie .proto. w metodzie run():

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

Zwróć uwagę, że w tym przypadku channel jest używany jako menedżer kontekstu i zostanie automatycznie zamknięty, gdy interpreter opuści blok with.

Wywoływanie metod usługi

W przypadku metod RPC, które zwracają pojedynczą odpowiedź (metody „response-unary”), gRPC Python obsługuje zarówno synchroniczną (blokującą), jak i asynchroniczną (nieblokującą) semantykę przepływu sterowania. W przypadku metod RPC przesyłających strumieniowo odpowiedzi wywołania natychmiast zwracają iterator wartości odpowiedzi. Wywołania metody next() tego iteratora są blokowane, dopóki nie będzie dostępna odpowiedź, która ma zostać zwrócona przez iterator.

RPC przesyłania strumieniowego po stronie serwera

Wywoływanie strumieniowania odpowiedzi ListFeatures jest podobne do pracy z typami sekwencji:

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 przesyłania strumieniowego po stronie klienta

Wywołanie strumieniowego żądania RecordRoute jest podobne do przekazania iteratora do metody lokalnej. Podobnie jak w przypadku prostego wywołania RPC powyżej, które też zwraca jedną odpowiedź, można go wywołać synchronicznie:

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

Dwukierunkowe strumieniowe RPC

Wywołanie dwukierunkowego strumienia RouteChat ma (podobnie jak po stronie usługi) semantykę strumienia żądań i strumienia odpowiedzi.

Wygeneruj wiadomości z prośbą i wysyłaj je pojedynczo za pomocą funkcji 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)

Otrzymywanie i przetwarzanie odpowiedzi serwera:

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

Wywoływanie metod pomocniczych

W funkcji run wykonaj utworzone metody i przekaż do nich stub.

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

7. Wypróbuj

Uruchom serwer:

python route_guide_server.py

W innym terminalu ponownie aktywuj środowisko wirtualne (source .venv/bin/activate)), a następnie uruchom klienta:

python route_guide_client.py

Przyjrzyjmy się wynikowi.

ListFeatures

Najpierw zobaczysz listę funkcji. Każda funkcja jest przesyłana strumieniowo z serwera (RPC przesyłania strumieniowego po stronie serwera), gdy tylko zostanie wykryta w obrębie żądanego prostokąta:

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

Po drugie, RecordRoute pokazuje listę losowo odwiedzanych punktów przesyłanych strumieniowo z klienta na serwer (strumieniowe wywołanie RPC po stronie klienta):

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

Po zakończeniu przesyłania strumieniowego wszystkich odwiedzonych punktów klient otrzyma z serwera odpowiedź bez przesyłania strumieniowego (binarne wywołanie RPC). Odpowiedź będzie zawierać podsumowanie obliczeń wykonanych na podstawie pełnej trasy klienta.

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

RouteChat

Wreszcie dane wyjściowe RouteChat pokazują strumieniowanie dwukierunkowe. Gdy klient „odwiedza” punkty home lub work, zapisuje notatkę dotyczącą tego punktu, wysyłając do serwera obiekt RouteNote. Gdy punkt został już odwiedzony, serwer przesyła strumieniowo wszystkie poprzednie notatki dotyczące tego punktu.

-------------- 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. Co dalej?