Python中维持gRPC流连接存活问题求助

问题背景 #

将C#应用转换为Python应用时，gRPC实现遇到问题：需要在应用运行期间维持流连接以接收服务端的广播消息，但调用XchangeBroadcast后连接立即关闭，触发_MultiThreadedRendezvous错误，尝试遍历请求流也无效。

相关文件与代码 #

Proto文件 #

syntax = "proto3";

import "google/protobuf/empty.proto";

option csharp_namespace = "mcGrpcService";

message mcCommandRequest {
  int32 id = 1;
  string content = 2;
}

message mcCommandResponse {
  bool succes = 1;
  string message = 2;
}

service grpcCommsService {
  rpc XchangeCommand (mcCommandRequest) returns (mcCommandResponse);
  rpc XchangeBroadcast (stream google.protobuf.Empty) returns (stream mcCommandRequest);
}

C# 实现代码 #

private readonly Channel? _channel;
private readonly grpcCommsService.grpcCommsServiceClient _client;

public McGrpcClient(string host, int port) {
    _channel = new Channel(host, port, ChannelCredentials.Insecure);
    _client = new grpcCommsService.grpcCommsServiceClient(_channel);
}

public bool Connect(Action<mcCommandRequest> onBroadcastReceived) {
    OnBroadcastReceived = onBroadcastReceived;

    var token = _cancellationTokenSource.Token;
    _broadcastTask = Task.Run(async () => {
        var broadcastCall = _client.XchangeBroadcast(cancellationToken: token);
        while (await broadcastCall.ResponseStream.MoveNext(token)) {
            try {
                var broadcastMessage = broadcastCall.ResponseStream.Current;
                OnBroadcastReceived?.Invoke(broadcastMessage);
            }
            catch {
                // todo implement error handling
                throw;
            }
        }
    }, token);
    return true;
}

当前Python代码 #

async def read_broadcast(request: Iterator[xchange__pb2.mcCommandRequest], stop_event):
    while not stop_event.is_set():
        msg = await request.next()
        print("Broadcast received: ", msg.message)
        await asyncio.sleep(0.1)

def connect(stub, stop_event):
    iterator = None
    request = stub.XchangeBroadcast(iterator)
    asyncio.run(
        read_broadcast(request, stop_event)
        )

if __name__ == '__main__':
    channel = grpc.insecure_channel("localhost:9999")
    stub = grpcCommsServiceStub(channel)
    stop_event = threading.Event()

    connect(stub, stop_event)

    input()
    stop_event.set()
    channel.close()

错误信息 #

Exception has occurred: _MultiThreadedRendezvous
<_MultiThreadedRendezvous of RPC that terminated with:
status = StatusCode.UNKNOWN
details = "Exception iterating requests!"
debug_error_string = "None">

解决方案 #

问题根源 #

XchangeBroadcast是双向流RPC（客户端发stream Empty，服务端回stream mcCommandRequest）。Python代码中传入None作为请求迭代器，服务端会认为客户端请求流已结束，直接关闭连接。而C#客户端生成的方法默认会维持一个空的请求流，不会主动关闭。

修正后的Python代码 #

import asyncio
import grpc
import xchange__pb2 as pb2
import xchange__pb2_grpc as pb2_grpc
from google.protobuf.empty_pb2 import Empty

async def broadcast_request_stream(stop_event):
    # 持续生成Empty消息，保持请求流处于打开状态
    while not stop_event.is_set():
        yield Empty()
        await asyncio.sleep(1)  # 间隔可根据服务端需求调整

async def read_broadcast(stub, stop_event):
    try:
        # 启动双向流，传入持续的请求迭代器
        response_stream = stub.XchangeBroadcast(broadcast_request_stream(stop_event))
        async for msg in response_stream:
            print(f"Broadcast received: ID={msg.id}, Content={msg.content}")
            if stop_event.is_set():
                break
    except grpc.RpcError as e:
        print(f"RPC错误: {e.code()} - {e.details()}")

async def main():
    stop_event = asyncio.Event()
    # 使用gRPC异步通道与Stub，匹配C#异步逻辑
    async with grpc.aio.insecure_channel("localhost:9999") as channel:
        stub = pb2_grpc.grpcCommsServiceStub(channel)
        # 启动广播读取任务
        broadcast_task = asyncio.create_task(read_broadcast(stub, stop_event))
        # 等待用户输入后终止
        await asyncio.get_event_loop().run_in_executor(None, input, "按回车停止...\n")
        stop_event.set()
        await broadcast_task

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

关键修改点 #

1. 异步API适配：改用grpc.aio异步通道与Stub，避免多线程同步冲突，和C#的异步实现逻辑对齐。
2. 维持请求流：实现broadcast_request_stream异步生成器，持续发送Empty消息，确保服务端认为请求流未关闭。
3. 响应流遍历：使用async for遍历响应流，这是gRPC异步API推荐的稳定遍历方式，替代手动调用next()。
4. 优雅终止：用asyncio.Event替代threading.Event，在异步环境中安全控制任务终止。

内容的提问来源于stack exchange，提问作者McNets