### 使用 gRPC 实现 C# 和 Python 程序双向通信的示例代码 以下是基于 gRPC 的 C# 和 Python 双向通信的具体实现方法。此示例展示了如何设置服务器端（C#）和客户端（Python），并使用双向流式 RPC 进行交互。 --- #### 1. **定义 `.proto` 文件** `.proto` 文件描述了服务接口和消息结构，这是 gRPC 中的核心配置文件。 ```protobuf syntax = "proto3"; service BiDirectionalService { rpc Communicate (stream RequestMessage) returns (stream ResponseMessage); } message RequestMessage { string data = 1; } message ResponseMessage { string message = 1; } ``` --- #### 2. **C# 服务器端实现** 在 C# 中创建一个 gRPC 服务来处理来自 Python 客户端的消息请求。 ##### 创建 `BiDirectionalServer.cs` ```csharp using Grpc.Core; using System.Threading.Tasks; namespace BiDirectionalGrpc.Server { public class BiDirectionalServiceImpl : BiDirectionalService.BiDirectionalServiceBase { public override async Task Communicate( IAsyncStreamReader<RequestMessage> requestStream, IServerStreamWriter<ResponseMessage> responseStream, ServerCallContext context) { while (await requestStream.MoveNext()) { var requestData = requestStream.Current.data; await responseStream.WriteAsync(new ResponseMessage { message = $"Received: {requestData}" }); } } } public class Program { const int Port = 50051; public static void Main(string[] args) { Server server = new Server { Services = { BiDirectionalService.BindService(new BiDirectionalServiceImpl()) }, Ports = { new ServerPort("localhost", Port, ServerCredentials.Insecure) } }; server.Start(); Console.WriteLine($"Listening on port {Port}"); Console.WriteLine("Press any key to stop the server..."); Console.ReadKey(); server.ShutdownAsync().Wait(); } } } ``` --- #### 3. **Python 客户端实现** 在 Python 中编写一个客户端程序，连接到 C# 服务器并通过双向流发送和接收消息。 ##### 创建 `bi_directional_client.py` ```python import grpc from concurrent import futures import time import bi_directional_pb2 as bd_pb2 import bi_directional_pb2_grpc as bd_pb2_grpc class Client: def __init__(self, address='localhost', port=50051): self.channel = grpc.insecure_channel(f'{address}:{port}') self.stub = bd_pb2_grpc.BiDirectionalServiceStub(self.channel) def communicate(self, messages): responses = self.stub.Communicate(messages) for response in responses: yield response.message def generate_messages(): for i in range(5): # Send 5 messages msg = f"Hello from Python! ({i})" yield bd_pb2.RequestMessage(data=msg) time.sleep(1) # Simulate delay between messages if __name__ == "__main__": client = Client() print("Starting communication with C# server...") try: for resp_msg in client.communicate(generate_messages()): print(f"Response received: {resp_msg}") except KeyboardInterrupt: print("\nCommunication stopped by user.") ``` --- #### 4. **编译 `.proto` 文件** 为了使 C# 和 Python 能够解析相同的协议缓冲区定义，需要先编译 `.proto` 文件： - **对于 C#:** 使用 `protoc` 工具生成 C# 类文件： ```bash protoc --csharp_out=./cs_output bi_directional.proto ``` - **对于 Python:** 同样使用 `protoc` 工具生成 Python 类文件： ```bash pip install protobuf grpcio-tools python -m grpc_tools.protoc --python_out=./py_output --grpc_python_out=./py_output bi_directional.proto ``` --- #### 5. **运行流程** 1. 启动 C# 服务器程序。 2. 在另一个终端中启动 Python 客户端程序。 3. 观察双方的通信过程。 --- ### 关键点说明 - **双向流式 RPC:** 此模式允许多个请求和响应在同一会话中交错发生[^2]。这使得它特别适用于实时应用或需要持续更新的数据交换场景。 - **跨语言支持:** gRPC 自带的强大工具链简化了不同语言间的互操作性问题[^1]。只需共享同一份 `.proto` 文件即可确保两端遵循一致的消息格式。 - **性能优化:** 利用 Protocol Buffers 编码方式代替 JSON/XML 等文本序列化技术，大幅提升了传输效率与解析速度[^3]。 ---