grpcio是 Google 开发的 gRPC 框架的 Python 实现是一个高性能、开源的通用 RPC远程过程调用框架基于 HTTP/2 协议和 Protocol Buffersprotobuf序列化格式。其核心优势包括✅高性能利用 HTTP/2 的多路复用、头部压缩、服务端推送等特性显著降低延迟与连接开销✅强类型接口通过.proto文件定义服务契约自动生成客户端/服务端存根stub保障类型安全与 IDE 支持✅多语言支持跨语言互通Python/Go/Java/C/JS 等天然适合微服务架构✅内置功能丰富支持双向流式通信、拦截器interceptors、认证TLS/mTLS、JWT、负载均衡、健康检查、超时与重试等✅异步支持完善提供async def风格的异步服务端aio模块和客户端适配 asyncio 生态。安装方式pipinstallgrpcio grpcio-tools典型使用流程编写.proto文件定义 service 和 message使用protoc --python_out. --grpc_python_out. xxx.proto生成 Python 代码实现服务端逻辑并启动 gRPC server创建客户端 stub 并发起同步或异步调用。示例简略# hello_pb2_grpc.py自动生成importhello_pb2,hello_pb2_grpcclassGreeterServicer(hello_pb2_grpc.GreeterServicer):defSayHello(self,request,context):returnhello_pb2.HelloReply(messagefHello,{request.name}!)# 启动服务端servergrpc.server(futures.ThreadPoolExecutor(max_workers10))hello_pb2_grpc.add_GreeterServicer_to_server(GreeterServicer(),server)server.add_insecure_port([::]:50051)server.start()server.wait_for_termination()⚠️ 注意事项生产环境建议启用 TLS 加密、合理配置最大消息大小max_message_length、启用 Keepalive 参数以维持长连接稳定性。在 gRPC Python 中实现双向流式通信Bidirectional Streaming需定义.proto文件中服务方法的请求和响应类型均为stream并在服务端和客户端均使用异步迭代器async for或同步生成器foriter进行持续收发。以下是完整、可运行的示例✅ 步骤 1定义.proto文件chat.protosyntax proto3; package chat; service ChatService { // 双向流客户端和服务端均可连续发送/接收消息 rpc Chat(stream ChatMessage) returns (stream ChatMessage); } message ChatMessage { string sender 1; string message 2; int64 timestamp 3; }✅ 步骤 2生成 Python 代码python-mgrpc_tools.protoc -I.--python_out.--grpc_python_out. chat.proto生成chat_pb2.py和chat_pb2_grpc.py。✅ 步骤 3服务端实现支持同步/异步推荐aio异步方式# server.pyimportasyncioimporttimeimportchat_pb2,chat_pb2_grpcclassChatServicer(chat_pb2_grpc.ChatServiceServicer):asyncdefChat(self,request_iterator,context):asyncforrequestinrequest_iterator:# 接收客户端流print(f[Server] Received from{request.sender}:{request.message})# 回复可实时、可延迟、可广播replychat_pb2.ChatMessage(senderserver,messagefEcho:{request.message},timestampint(time.time()))yieldreply# 发送流式响应asyncdefserve():servergrpc.aio.server()chat_pb2_grpc.add_ChatServiceServicer_to_server(ChatServicer(),server)server.add_insecure_port([::]:50051)awaitserver.start()print(gRPC server running on :50051)awaitserver.wait_for_termination()if__name____main__:asyncio.run(serve())✅ 使用grpc.aio需grpcio1.39.0天然支持async def和async for避免线程阻塞适合高并发场景。✅ 步骤 4客户端实现异步双向流# client.pyimportasyncioimporttimeimportchat_pb2,chat_pb2_grpcasyncdefrun():asyncwithgrpc.aio.insecure_channel(localhost:50051)aschannel:stubchat_pb2_grpc.ChatServiceStub(channel)# 创建双向流调用callstub.Chat()# 启动发送协程模拟用户输入asyncdefsend_messages():foriinrange(3):msgchat_pb2.ChatMessage(senderclient,messagefHello #{i1},timestampint(time.time()))awaitcall.write(msg)awaitasyncio.sleep(1)awaitcall.done_writing()# 告知服务端发送结束可选但推荐# 启动接收协程asyncdefrecv_messages():asyncforresponseincall:print(f[Client] Received:{response.message}(from{response.sender}))# 并发执行收发awaitasyncio.gather(send_messages(),recv_messages())if__name____main__:asyncio.run(run())⚠️ 关键点call stub.Chat()返回一个grpc.aio.StreamStreamCall对象call.write()发送单条消息call.done_writing()显式结束写入否则服务端可能永远等待async for response in call持续接收服务端推送的消息必须用asyncio.gather并发处理读写否则会因await call.write()阻塞接收。 补充同步版不推荐高并发仅作兼容参考若必须用同步grpc非aio需在独立线程中运行iter(call)call.send()并手动管理线程/队列复杂度高、易出错故强烈建议优先采用grpc.aio。✅典型应用场景实时聊天、IoT 设备指令与状态双向同步、协同编辑、日志/指标流式上报等。