Mulai Menggunakan gRPC-Python - Streaming

1. Pengantar

Dalam codelab ini, Anda akan menggunakan gRPC-Python untuk membuat klien dan server yang membentuk dasar aplikasi pemetaan rute yang ditulis dalam Python.

Di akhir tutorial, Anda akan memiliki klien yang terhubung ke server jarak jauh menggunakan gRPC untuk mendapatkan informasi tentang fitur di rute klien, membuat ringkasan rute klien, dan bertukar informasi rute seperti info terbaru lalu lintas dengan server dan klien lain.

Layanan ini ditentukan dalam file Protocol Buffers, yang akan digunakan untuk membuat kode boilerplate bagi klien dan server sehingga keduanya dapat berkomunikasi satu sama lain, sehingga menghemat waktu dan upaya Anda dalam menerapkan fungsi tersebut.

Kode yang dihasilkan ini tidak hanya menangani kompleksitas komunikasi antara server dan klien, tetapi juga serialisasi dan deserialisasi data.

Yang akan Anda pelajari

• Cara menggunakan Protocol Buffers untuk menentukan API layanan.
• Cara membangun klien dan server berbasis gRPC dari definisi Protocol Buffers menggunakan pembuatan kode otomatis.
• Pemahaman tentang komunikasi streaming klien-server dengan gRPC.

Codelab ini ditujukan bagi developer Python yang baru menggunakan gRPC atau ingin mempelajari kembali gRPC, atau siapa pun yang tertarik untuk membangun sistem terdistribusi. Tidak diperlukan pengalaman gRPC sebelumnya.

2. Sebelum memulai

Yang Anda butuhkan

• Python 3.9 atau yang lebih tinggi. Sebaiknya gunakan Python 3.13. Untuk mengetahui petunjuk penginstalan khusus platform, lihat Penyiapan dan Penggunaan Python. Atau, instal Python non-sistem menggunakan alat seperti uv atau pyenv.
• pip untuk menginstal paket Python.
• venv untuk membuat lingkungan virtual Python.

Paket ensurepip dan venv adalah bagian dari Python Standard Library dan biasanya tersedia secara default.

Namun, beberapa distribusi berbasis Debian (termasuk Ubuntu) memilih untuk mengecualikannya saat mendistribusikan ulang Python. Untuk menginstal paket, jalankan:

sudo apt install python3-pip python3-venv

Mendapatkan kode

Untuk menyederhanakan pembelajaran Anda, codelab ini menawarkan struktur kode sumber bawaan untuk membantu Anda memulai. Langkah-langkah berikut akan memandu Anda menyelesaikan aplikasi, termasuk pembuatan kode gRPC menggunakan plugin compiler Protocol Buffer grpc_tools.protoc.

grpc-codelabs

Kode sumber scaffold untuk codelab ini tersedia di direktori codelabs/grpc-python-streaming/start_here. Jika Anda memilih untuk tidak menerapkan kode sendiri, kode sumber yang sudah selesai tersedia di direktori completed.

Pertama, buat direktori kerja codelab dan cd ke dalamnya:

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

Download dan ekstrak 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

Atau, Anda dapat mendownload file .zip yang hanya berisi direktori codelab dan mengekstraknya secara manual.

Download ZIP

3. Menentukan pesan dan layanan

Langkah pertama Anda adalah menentukan layanan gRPC aplikasi, metode RPC, serta jenis pesan permintaan dan responsnya menggunakan Protocol Buffers. Layanan Anda akan menyediakan:

• Metode RPC yang disebut ListFeatures, RecordRoute, dan RouteChat yang diimplementasikan server dan dipanggil klien.
• Jenis pesan Point, Feature, Rectangle, RouteNote, dan RouteSummary, yang merupakan struktur data yang dipertukarkan antara klien dan server saat memanggil metode RPC.

Metode RPC ini dan jenis pesannya akan ditentukan dalam file protos/route_guide.proto kode sumber yang diberikan.

Protocol Buffers biasanya dikenal sebagai protobuf. Untuk mengetahui informasi selengkapnya tentang terminologi gRPC, lihat Konsep inti, arsitektur, dan siklus proses gRPC.

Menentukan jenis pesan

Dalam file protos/route_guide.proto kode sumber, tentukan terlebih dahulu jenis pesan Point. Point mewakili pasangan koordinat lintang-bujur di peta. Untuk codelab ini, gunakan bilangan bulat untuk koordinat:

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

Angka 1 dan 2 adalah nomor ID unik untuk setiap kolom dalam struktur message.

Selanjutnya, tentukan jenis pesan Feature. Feature menggunakan kolom string untuk nama atau alamat pos sesuatu di lokasi yang ditentukan oleh Point:

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

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

Agar beberapa titik dalam suatu area dapat di-streaming ke klien, Anda memerlukan pesan Rectangle yang merepresentasikan persegi panjang lintang-bujur, yang direpresentasikan sebagai dua titik yang berlawanan secara diagonal lo dan hi:

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

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

Selain itu, pesan RouteNote yang mewakili pesan yang dikirim saat berada di titik tertentu:

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

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

Terakhir, Anda memerlukan pesan RouteSummary. Pesan ini diterima sebagai respons terhadap RPC RecordRoute, yang dijelaskan di bagian berikutnya. Objek ini berisi jumlah titik individual yang diterima, jumlah fitur yang terdeteksi, dan total jarak yang ditempuh sebagai jumlah kumulatif jarak antara setiap titik.

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

Menentukan metode layanan

Untuk menentukan layanan, Anda menentukan layanan bernama dalam file .proto. File route_guide.proto memiliki struktur service bernama RouteGuide yang menentukan satu atau beberapa metode yang disediakan oleh layanan aplikasi.

Saat menentukan metode RPC di dalam definisi layanan, Anda menentukan jenis permintaan dan responsnya. Di bagian codelab ini, mari kita tentukan:

ListFeatures

Mendapatkan objek Feature yang tersedia dalam Rectangle tertentu. Hasil di-streaming, bukan ditampilkan sekaligus karena persegi panjang dapat mencakup area yang luas dan berisi sejumlah besar fitur.

Untuk aplikasi ini, Anda akan menggunakan RPC streaming sisi server: klien mengirimkan permintaan ke server dan mendapatkan stream untuk membaca kembali urutan pesan. Klien membaca dari stream yang ditampilkan hingga tidak ada lagi pesan. Seperti yang dapat Anda lihat dalam contoh kami, Anda menentukan metode streaming sisi server dengan menempatkan kata kunci stream sebelum jenis respons.

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

RecordRoute

Menerima aliran Poin pada rute yang dilalui, menampilkan RouteSummary saat penelusuran selesai.

RPC streaming sisi klien cocok dalam kasus ini: klien menulis urutan pesan dan mengirimkannya ke server, lagi-lagi menggunakan stream yang disediakan. Setelah klien selesai menulis pesan, klien akan menunggu server membaca semua pesan tersebut dan menampilkan responsnya. Anda menentukan metode streaming sisi klien dengan menempatkan kata kunci stream sebelum jenis permintaan.

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

RouteChat

Menerima aliran RouteNotes yang dikirim saat rute sedang dilalui, sambil menerima RouteNotes lain (misalnya dari pengguna lain).

Inilah kasus penggunaan yang tepat untuk streaming dua arah. RPC streaming dua arah di mana kedua sisi mengirimkan urutan pesan menggunakan stream baca-tulis. Kedua aliran beroperasi secara independen, sehingga klien dan server dapat membaca dan menulis dalam urutan apa pun yang mereka inginkan: misalnya, server dapat menunggu untuk menerima semua pesan klien sebelum menulis responsnya, atau dapat membaca pesan lalu menulis pesan, atau kombinasi pembacaan dan penulisan lainnya. Urutan pesan di setiap aliran akan tetap terjaga. Anda menentukan jenis metode ini dengan menempatkan kata kunci stream sebelum permintaan dan respons.

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

4. Buat kode klien dan server

Selanjutnya, buat kode gRPC boilerplate untuk klien dan server dari file .proto menggunakan compiler protocol buffer.

Untuk pembuatan kode gRPC Python, kami membuat grpcio-tools. Ini mencakup:

1. Compiler protoc reguler yang menghasilkan kode Python dari definisi message.
2. Plugin protobuf gRPC yang menghasilkan kode Python (stub klien dan server) dari definisi service.

Kita akan menginstal paket Python grpcio-tools menggunakan pip. Mari buat lingkungan virtual python (venv) baru untuk mengisolasi dependensi project Anda dari paket sistem:

python3 -m venv --upgrade-deps .venv

Untuk mengaktifkan lingkungan virtual di shell bash/zsh:

source .venv/bin/activate

Untuk Windows dan shell non-standar, lihat tabel di https://docs.python.org/3/library/venv.html#how-venvs-work.

Catatan: Anda harus mengaktifkan lingkungan virtual setiap kali Anda membuka shell baru. Skrip activate mengonfigurasi variabel $PATH untuk menggunakan python dan pip dari venv Anda. Untuk menonaktifkan venv, cukup jalankan deactivate.

Selanjutnya, instal grpcio-tools (tindakan ini juga akan menginstal paket grpcio):

pip install grpcio-tools

Gunakan perintah berikut untuk membuat kode boilerplate Python:

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

Tindakan ini akan menghasilkan file berikut untuk antarmuka yang kita tentukan di route_guide.proto:

1. route_guide_pb2.py berisi kode yang membuat class secara dinamis yang dihasilkan dari definisi message.
2. route_guide_pb2.pyi adalah "file stub" atau "file petunjuk jenis" yang dihasilkan dari definisi message. Hanya berisi tanda tangan tanpa penerapan. File stub dapat digunakan oleh IDE untuk memberikan pelengkapan otomatis dan deteksi error yang lebih baik.
3. route_guide_pb2_grpc.py dihasilkan dari definisi service dan berisi class serta fungsi khusus gRPC.

Fakta: Suffix _pb2 tidak menyiratkan sintaksis proto2. Menunjukkan versi Python Protocol Buffers API. Saat ini, proto2, proto3, dan Editions semuanya menggunakan _pb2.py untuk file yang dihasilkan.

Kode khusus gRPC berisi:

1. RouteGuideStub, yang dapat digunakan oleh klien gRPC untuk memanggil RPC RouteGuide.
2. RouteGuideServicer, yang menentukan antarmuka untuk penerapan layanan RouteGuide.
3. Fungsi add_RouteGuideServicer_to_server yang digunakan untuk mendaftarkan RouteGuideServicer ke server gRPC.

5. Buat server

Pertama, mari kita lihat cara membuat server RouteGuide. Membuat dan menjalankan server RouteGuide dibagi menjadi dua item pekerjaan:

• Mengimplementasikan antarmuka layanan yang dihasilkan dari definisi layanan dengan fungsi yang melakukan "tugas" layanan yang sebenarnya.
• Menjalankan server gRPC untuk memproses permintaan dari klien dan mengirimkan respons.

Mari kita lihat route_guide_server.py.

Mengimplementasikan RouteGuide

route_guide_server.py memiliki class RouteGuideServicer yang merupakan subclass dari class route_guide_pb2_grpc.RouteGuideServicer yang dihasilkan:

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

RouteGuideServicer mengimplementasikan semua metode layanan RouteGuide.

RPC streaming sisi server

ListFeatures adalah RPC streaming respons yang mengirim beberapa Feature ke klien:

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

Di sini, pesan permintaan adalah route_guide_pb2.Rectangle yang di dalamnya klien ingin menemukan Feature. Daripada menampilkan satu respons, metode ini menghasilkan nol atau lebih respons.

RPC streaming sisi klien

Metode streaming permintaan RecordRoute menggunakan iterator nilai permintaan dan menampilkan satu nilai respons.

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 streaming dua arah

Terakhir, mari kita lihat RPC streaming dua arah 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)

Semantik metode ini adalah kombinasi dari semantik metode streaming permintaan dan metode streaming respons. Objek ini meneruskan iterator nilai permintaan dan merupakan iterator nilai respons.

Mulai server

Setelah Anda menerapkan semua metode RouteGuide, langkah berikutnya adalah memulai server gRPC agar klien dapat benar-benar menggunakan layanan Anda:

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

Metode start() server tidak memblokir. Thread baru akan di-instansiasi untuk menangani permintaan. Thread yang memanggil server.start() sering kali tidak memiliki tugas lain yang harus dilakukan pada saat yang sama. Dalam hal ini, Anda dapat memanggil server.wait_for_termination() untuk memblokir thread panggilan secara bersih hingga server berhenti.

6. Buat klien

Mari kita lihat route_guide_client.py.

Membuat stub

Untuk memanggil metode layanan, kita harus membuat stub terlebih dahulu.

Kita membuat instance class RouteGuideStub dari modul route_guide_pb2_grpc, yang dihasilkan dari metode .proto. In run():

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

Perhatikan bahwa di sini channel digunakan sebagai pengelola konteks, dan akan otomatis ditutup setelah interpreter keluar dari blok with.

Panggil metode layanan

Untuk metode RPC yang menampilkan satu respons (metode "response-unary"), gRPC Python mendukung semantik alur kontrol sinkron (pemblokiran) dan asinkron (non-pemblokiran). Untuk metode RPC streaming respons, panggilan akan segera menampilkan iterator nilai respons. Panggilan ke metode next() iterator tersebut akan diblokir hingga respons yang akan dihasilkan dari iterator tersedia.

RPC streaming sisi server

Memanggil streaming respons ListFeatures mirip dengan bekerja dengan jenis urutan:

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 streaming sisi klien

Memanggil RecordRoute streaming permintaan mirip dengan meneruskan iterator ke metode lokal. Seperti RPC sederhana di atas yang juga menampilkan satu respons, RPC ini dapat dipanggil secara serentak:

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 streaming dua arah

Memanggil RouteChat streaming dua arah memiliki (seperti yang terjadi di sisi layanan) kombinasi semantik streaming permintaan dan streaming respons.

Buat pesan permintaan, lalu kirim satu per satu menggunakan 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)

Catatan: Dalam contoh ini, kita akan menunggu sebentar sebelum mengirim setiap pesan untuk menggambarkan urutan pengiriman dan penerimaan pesan. Dalam praktiknya, Anda dapat mendesain protokol tempat operasi ini terjadi secara independen.

Menerima dan memproses respons server:

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

Panggil metode bantuan

Saat dijalankan, eksekusi metode yang baru saja kita buat, dan teruskan stub ke metode tersebut.

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

7. Cobalah

Jalankan server:

python route_guide_server.py

Dari terminal lain, aktifkan kembali lingkungan virtual (source .venv/bin/activate), lalu jalankan klien:

python route_guide_client.py

Mari kita lihat outputnya.

ListFeatures

Pertama, Anda akan menemukan daftar fitur. Setiap fitur di-streaming dari server (RPC streaming sisi server) saat fitur tersebut ditemukan berada dalam persegi panjang yang diminta:

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

Kedua, RecordRoute menunjukkan daftar titik yang dikunjungi secara acak yang di-streaming dari klien ke server (RPC streaming sisi klien):

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

Setelah klien selesai melakukan streaming semua titik yang dikunjungi, klien akan menerima respons non-streaming (RPC unary) dari server. Respons ini akan berisi ringkasan perhitungan yang dilakukan pada rute lengkap klien.

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

RouteChat

Terakhir, output RouteChat menunjukkan streaming dua arah. Saat "mengunjungi" titik home atau work, klien akan merekam catatan untuk titik tersebut dengan mengirim RouteNote ke server. Jika suatu titik sudah dikunjungi, server akan melakukan streaming kembali semua catatan sebelumnya untuk titik ini.

-------------- 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. Langkah berikutnya

• Pelajari cara kerja gRPC di Pengantar gRPC dan Konsep inti
• Pelajari tutorial Dasar-Dasar.
• Pelajari referensi Python API