在 Python 中使用 gRPC。

1. 安装必要的包

pip install grpcio grpcio-tools

2. 定义服务接口 (.proto 文件)

创建 calculator.proto 文件：

syntax = "proto3";

package calculator;

service Calculator {
  rpc Add (AddRequest) returns (AddResponse) {}
  rpc Multiply (MultiplyRequest) returns (MultiplyResponse) {}
  rpc GetHistory (HistoryRequest) returns (HistoryResponse) {}
}

message AddRequest {
  double number1 = 1;
  double number2 = 2;
}

message AddResponse {
  double result = 1;
}

message MultiplyRequest {
  double number1 = 1;
  double number2 = 2;
}

message MultiplyResponse {
  double result = 1;
}

message HistoryRequest {}

message HistoryEntry {
  string operation = 1;
  double number1 = 2;
  double number2 = 3;
  double result = 4;
  string timestamp = 5;
}

message HistoryResponse {
  repeated HistoryEntry entries = 1;
}

3. 生成 Python 代码

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

这会生成两个文件：

• calculator_pb2.py - 消息类
• calculator_pb2_grpc.py - 服务类

4. 实现服务端

创建 server.py：

import grpc
from concurrent import futures
import time
import calculator_pb2
import calculator_pb2_grpc
from datetime import datetime

class CalculatorServicer(calculator_pb2_grpc.CalculatorServicer):
    def __init__(self):
        self.history = []
    
    def Add(self, request, context):
        result = request.number1 + request.number2
        
        # 记录历史
        entry = calculator_pb2.HistoryEntry(
            operation="ADD",
            number1=request.number1,
            number2=request.number2,
            result=result,
            timestamp=datetime.now().isoformat()
        )
        self.history.append(entry)
        
        return calculator_pb2.AddResponse(result=result)
    
    def Multiply(self, request, context):
        result = request.number1 * request.number2
        
        # 记录历史
        entry = calculator_pb2.HistoryEntry(
            operation="MULTIPLY",
            number1=request.number1,
            number2=request.number2,
            result=result,
            timestamp=datetime.now().isoformat()
        )
        self.history.append(entry)
        
        return calculator_pb2.MultiplyResponse(result=result)
    
    def GetHistory(self, request, context):
        return calculator_pb2.HistoryResponse(entries=self.history)

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    calculator_pb2_grpc.add_CalculatorServicer_to_server(CalculatorServicer(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    print("服务器启动，监听端口 50051...")
    
    try:
        while True:
            time.sleep(86400)  # 一天
    except KeyboardInterrupt:
        server.stop(0)

if __name__ == '__main__':
    serve()

5. 实现客户端

创建 client.py：

import grpc
import calculator_pb2
import calculator_pb2_grpc

def run():
    channel = grpc.insecure_channel('localhost:50051')
    stub = calculator_pb2_grpc.CalculatorStub(channel)
    
    try:
        # 测试加法
        add_request = calculator_pb2.AddRequest(number1=10.5, number2=20.3)
        add_response = stub.Add(add_request)
        print(f"加法结果: {add_response.result}")
        
        # 测试乘法
        multiply_request = calculator_pb2.MultiplyRequest(number1=5, number2=7)
        multiply_response = stub.Multiply(multiply_request)
        print(f"乘法结果: {multiply_response.result}")
        
        # 获取历史记录
        history_request = calculator_pb2.HistoryRequest()
        history_response = stub.GetHistory(history_request)
        
        print("\n操作历史:")
        for entry in history_response.entries:
            print(f"  {entry.timestamp}: {entry.number1} {entry.operation} {entry.number2} = {entry.result}")
            
    except grpc.RpcError as e:
        print(f"gRPC 错误: {e.code()} - {e.details()}")

if __name__ == '__main__':
    run()

6. 异步版本示例

异步服务端

import asyncio
import grpc
from calculator_pb2 import AddResponse, MultiplyResponse, HistoryResponse, HistoryEntry
from calculator_pb2_grpc import CalculatorServicer, add_CalculatorServicer_to_server
from datetime import datetime

class AsyncCalculatorServicer(CalculatorServicer):
    def __init__(self):
        self.history = []
    
    async def Add(self, request, context):
        result = request.number1 + request.number2
        
        entry = HistoryEntry(
            operation="ADD",
            number1=request.number1,
            number2=request.number2,
            result=result,
            timestamp=datetime.now().isoformat()
        )
        self.history.append(entry)
        
        return AddResponse(result=result)
    
    async def Multiply(self, request, context):
        result = request.number1 * request.number2
        
        entry = HistoryEntry(
            operation="MULTIPLY",
            number1=request.number1,
            number2=request.number2,
            result=result,
            timestamp=datetime.now().isoformat()
        )
        self.history.append(entry)
        
        return MultiplyResponse(result=result)
    
    async def GetHistory(self, request, context):
        return HistoryResponse(entries=self.history)

async def serve():
    server = grpc.aio.server()
    add_CalculatorServicer_to_server(AsyncCalculatorServicer(), server)
    server.add_insecure_port('[::]:50051')
    await server.start()
    print("异步服务器启动，监听端口 50051...")
    
    try:
        await server.wait_for_termination()
    except KeyboardInterrupt:
        await server.stop(0)

if __name__ == '__main__':
    asyncio.run(serve())

异步客户端

import asyncio
import grpc
import calculator_pb2
import calculator_pb2_grpc

async def run():
    async with grpc.aio.insecure_channel('localhost:50051') as channel:
        stub = calculator_pb2_grpc.CalculatorStub(channel)
        
        try:
            # 并发请求
            tasks = [
                stub.Add(calculator_pb2.AddRequest(number1=10.5, number2=20.3)),
                stub.Multiply(calculator_pb2.MultiplyRequest(number1=5, number2=7)),
                stub.GetHistory(calculator_pb2.HistoryRequest())
            ]
            
            add_response, multiply_response, history_response = await asyncio.gather(*tasks)
            
            print(f"加法结果: {add_response.result}")
            print(f"乘法结果: {multiply_response.result}")
            
            print("\n操作历史:")
            for entry in history_response.entries:
                print(f"  {entry.timestamp}: {entry.number1} {entry.operation} {entry.number2} = {entry.result}")
                
        except grpc.RpcError as e:
            print(f"gRPC 错误: {e.code()} - {e.details()}")

if __name__ == '__main__':
    asyncio.run(run())

7. 运行步骤

1. 启动服务端：

python server.py

2. 运行客户端（在另一个终端）：

python client.py

8. 常用功能和配置

添加认证

# 服务端
server_credentials = grpc.ssl_server_credentials()
server.add_secure_port('[::]:50051', server_credentials)

# 客户端
channel_credentials = grpc.ssl_channel_credentials()
channel = grpc.secure_channel('localhost:50051', channel_credentials)

添加拦截器

class LoggingInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        print(f"调用方法: {handler_call_details.method}")
        return continuation(handler_call_details)

server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=10),
    interceptors=[LoggingInterceptor()]
)

设置超时

# 客户端
try:
    response = stub.Add(add_request, timeout=10.0)  # 10秒超时
except grpc.RpcError as e:
    if e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
        print("请求超时")