Khởi tạo môi trường phát triển

Đảm bảo hệ thống đã cài đặt Python phiên bản 3.6 trở lên. Kiểm tra bằng lệnh:

python --version

Nếu chưa có, tải và cài đặt từ trang chính thức của Python.

Cài đặt thư viện gRPC

Sử dụng công cụ pip để cài đặt các thành phần cần thiết:

pip install grpcio grpcio-tools

Trong đó, grpcio là nền tảng gRPC, còn grpcio-tools hỗ trợ biên dịch file định nghĩa giao diện.

Xây dựng giao diện dịch vụ với .proto

Tạo một file course.proto để định nghĩa giao diện dịch vụ:

syntax = "proto3";

package course_service;

message UploadRequest {
  string course_id = 1;
  bytes data = 2;
}

message UploadResponse {
  string status = 1;
}

message DownloadRequest {
  string course_id = 1;
}

message DownloadResponse {
  bytes content = 1;
}

service CourseManager {
  rpc Upload(UploadRequest) returns (UploadResponse);
  rpc Download(DownloadRequest) returns (DownloadResponse);
}

Biên dịch file .proto thành mã Python

Chạy lệnh sau để sinh mã Python từ file định nghĩa:

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

Sẽ sinh ra hai file: course_service_pb2.py (định nghĩa dữ liệu) và course_service_pb2_grpc.py (giao diện dịch vụ).

Triển khai máy chủ gRPC

Viết mã máy chủ xử lý yêu cầu từ client:

from concurrent import futures
import grpc
import course_service_pb2
import course_service_pb2_grpc

class CourseHandler(course_service_pb2_grpc.CourseManagerServicer):
    def Upload(self, request, context):
        course_id = request.course_id
        file_data = request.data
        # Ghi dữ liệu vào cơ sở dữ liệu hoặc lưu trữ
        save_to_storage(course_id, file_data)
        return course_service_pb2.UploadResponse(status="Uploaded successfully")

    def Download(self, request, context):
        course_id = request.course_id
        data = fetch_from_storage(course_id)
        return course_service_pb2.DownloadResponse(content=data)

def start_server():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=8))
    course_service_pb2_grpc.add_CourseManagerServicer_to_server(CourseHandler(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    print("Server running on port 50051...")
    server.wait_for_termination()

if __name__ == '__main__':
    start_server()

Tạo khách hàng gRPC

Mã phía client kết nối đến máy chủ và thực hiện gọi dịch vụ:

import grpc
import course_service_pb2
import course_service_pb2_grpc

def call_server():
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = course_service_pb2_grpc.CourseManagerStub(channel)
        upload_req = course_service_pb2.UploadRequest(
            course_id="CS101",
            data=b"Sample course material"
        )
        upload_resp = stub.Upload(upload_req)
        print("Upload result:", upload_resp.status)

        download_req = course_service_pb2.DownloadRequest(course_id="CS101")
        download_resp = stub.Download(download_req)
        print("Downloaded content:", download_resp.content.decode())

if __name__ == '__main__':
    call_server()

Xử lý lỗi và kiểm tra hoạt động

Trên máy chủ, có thể ném lỗi khi dữ liệu không hợp lệ:

def Upload(self, request, context):
    if not request.course_id:
        context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
        context.set_details("Course ID is required.")
        raise grpc.RpcError()
    # Xử lý upload...

Trên client, bắt lỗi để phản hồi phù hợp:

try:
    response = stub.Upload(req)
except grpc.RpcError as e:
    print(f"Error: {e.code()} - {e.details()}")

Giao tiếp luồng (Streaming)

gRPC hỗ trợ truyền dữ liệu luồng, hữu ích cho việc gửi/ nhận dữ liệu lớn hoặc thời gian thực.

Luồng từ client

def stream_data():
    for i in range(5):
        yield course_service_pb2.UploadRequest(course_id=f"CS{i}", data=f"Chunk {i}".encode())

def send_stream():
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = course_service_pb2_grpc.CourseManagerStub(channel)
        responses = stub.UploadStream(stream_data())
        for resp in responses:
            print(resp.status)

Luồng từ server

def receive_stream():
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = course_service_pb2_grpc.CourseManagerStub(channel)
        for chunk in stub.DownloadStream(course_service_pb2.DownloadRequest(course_id="CS101")):
            print(chunk.content.decode())

Bảo mật và xác thực

Sử dụng TLS để mã hóa kết nối:

server_credentials = grpc.ssl_server_credentials((
    (open('key.pem').read(), open('cert.pem').read()),
))

server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
server.add_secure_port('[::]:50051', server_credentials)

Xác thực OAuth2 qua interceptor:

token = get_oauth_token(client_id, client_secret, token_url)
metadata = [('authorization', f'Bearer {token}')]

stub.Upload(request, metadata=metadata)

Tối ưu hiệu năng

• Serial hóa nhị phân: Sử dụng Protocol Buffers thay vì JSON giúp giảm kích thước dữ liệu.
• Luồng dữ liệu: Giảm số lần tương tác giữa client và server.
• Compress: Kích hoạt nén trên kênh:

channel = grpc.secure_channel('localhost:50051', grpc.compression_channel_credentials(grpc.ssl_channel_credentials()))