使用 Python 和 gRPC 实现高性能微服务通信：生产级指南

# server.py
from concurrent import futures
import time
import grpc
import product_pb2
import product_pb2_grpc

# 虚拟数据库
DUMMY_PRODUCTS = [
    product_pb2.Product(id="p001", name="Laptop Pro X", description="High-end laptop", price=1500.00, category="Electronics"),
    product_pb2.Product(id="p002", name="Wireless Mouse", description="Ergonomic mouse", price=75.50, category="Electronics"),
    product_pb2.Product(id="p003", name="Mechanical Keyboard", description="RGB Keyboard", price=120.00, category="Electronics"),
    product_pb2.Product(id="p004", name="The Python Guide", description="A book for Pythonistas", price=45.99, category="Books"),
]

class ProductServiceServicer(product_pb2_grpc.ProductServiceServicer):
    """实现 ProductService 实际逻辑的类"""

    def GetProduct(self, request, context):
        """一元 RPC: 通过商品 ID 查询商品信息"""
        print(f"Received GetProduct request for ID: {request.product_id}")
        for product in DUMMY_PRODUCTS:
            if product.id == request.product_id:
                return product
        
        # 未找到商品
        context.set_code(grpc.StatusCode.NOT_FOUND)
        context.set_details(f"Product with ID '{request.product_id}' not found.")
        return product_pb2.Product()

    def ListProductsByCategory(self, request, context):
        """服务器流式 RPC: 通过类别以流的形式返回商品列表"""
        print(f"Received ListProductsByCategory request for category: {request.category}")
        for product in DUMMY_PRODUCTS:
            if product.category == request.category:
                print(f"Streaming product: {product.name}")
                yield product
                time.sleep(1) # 用于可视化流式传输的延迟

def serve():
    """启动 gRPC 服务器的函数"""
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    product_pb2_grpc.add_ProductServiceServicer_to_server(ProductServiceServicer(), server)
    
    server_address = '[::]:50051'
    server.add_insecure_port(server_address)
    
    print(f"🚀 Server starting on {server_address}")
    server.start()
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

# client.py
import grpc
import product_pb2
import product_pb2_grpc

def run():
    # 创建与服务器的通道
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = product_pb2_grpc.ProductServiceStub(channel)

        # 1. 调用 GetProduct (一元 RPC)
        print("--- Calling GetProduct (p001) ---")
        try:
            product_response = stub.GetProduct(product_pb2.GetProductRequest(product_id="p001"))
            print(f"Product found: {product_response.name}, Price: ${product_response.price}")
        except grpc.RpcError as e:
            print(f"RPC failed: {e.code()} - {e.details()}")

        print("\n--- Calling GetProduct (p999 - not found) ---")
        try:
            stub.GetProduct(product_pb2.GetProductRequest(product_id="p999"))
        except grpc.RpcError as e:
            print(f"RPC failed as expected: {e.code()} - {e.details()}")

        # 2. 调用 ListProductsByCategory (服务器流式 RPC)
        print("\n--- Calling ListProductsByCategory (Electronics) ---")
        try:
            product_stream = stub.ListProductsByCategory(
                product_pb2.ListProductsByCategoryRequest(category="Electronics")
            )
            print("Receiving product stream...")
            for product in product_stream:
                print(f"  - Received: {product.name} (ID: {product.id})")
            print("Stream finished.")
        except grpc.RpcError as e:
            print(f"RPC failed: {e.code()} - {e.details()}")

if __name__ == '__main__':
    run()

# in Servicer class
def SomeRpcMethod(self, request, context):
    if not request.user_id:
        context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
        context.set_details("user_id field is required.")
        return some_pb2.SomeResponse()
    
    # ... logic ...

class AuthInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        metadata = dict(handler_call_details.invocation_metadata)
        auth_token = metadata.get('authorization')

        if not self._is_valid_token(auth_token):
            # 由于不能直接访问 context，因此返回一个抛出错误的特殊处理器
            return self._abort_with_status(grpc.StatusCode.UNAUTHENTICATED, "Invalid token")

        return continuation(handler_call_details)

    # ... helper methods ...

# client.py
try:
    # 如果在 5 秒内没有响应，则抛出 DEADLINE_EXCEEDED 错误
    response = stub.SlowRpcMethod(request, timeout=5) 
except grpc.RpcError as e:
    if e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
        print("Request timed out!")