GRPC服务端流式调用：留存响应流后续写入报错问题排查

我开发了一个gRPC服务端，想要留存不同客户端的响应流，以便后续在事件触发时通过其他方法向客户端写入响应。但事件触发时写入留存的响应流时，抛出异常：{"Can't write the message because the request is complete."}。请问：

1. 请求为何会完成？
2. 如何实现后续向响应流写入数据？

客户端代码 #

static async Task Main(string[] args)
{
        var userId = Guid.NewGuid().ToString();
        using var channel = GrpcChannel.ForAddress("https://localhost:8080");
        Job.JobClient client;
        client = new Job.JobClient(channel);
        
        using var jobServerStreaming = client.GetJobsRT(new JobsRequest() { UserId = userId });
        try
        {                
            var readerTask = Task.Run(async () =>
            {
                await foreach (var res in jobServerStreaming.ResponseStream.ReadAllAsync())
                {
                    Console.WriteLine($"jobServerStreaming Response: {res.Message}");
                }
            });
        }
        catch (Exception)
        {
            throw;
        }
}

服务端代码 #

public class JobService : Job.JobBase
{
public static Dictionary<string, IServerStreamWriter<JobsReply>> _grpcResponseStreams = new();
    //Server streaming
    public override async Task GetJobsRT(
        JobsRequest request,
        IServerStreamWriter<JobsReply> responseStream,
        ServerCallContext context)
    {
        _grpcResponseStreams.Add(request.UserId, responseStream);
        await responseStream.WriteAsync(new JobsReply() { Message = $"Let you know jobs updates and executions on real-time." });
    }
}

事件触发写入方法 #

async ValueTask AfterMessageReceived(Message context)
{
        try
        {
            foreach (var responseStream in JobService._grpcResponseStreams.Values)
                await responseStream.WriteAsync(new JobsReply() { Message = context.Message});
        }
        catch (Exception ex)//** -> here the ex is thrown {"Can't write the message because the request is complete."}**
        {
            Console.WriteLine(ex.ToString());
        }
}

一、请求为何会完成？ #

1. 服务端方法提前结束：gRPC服务端流式RPC中，当GetJobsRT方法执行完毕（返回Task）时，gRPC框架会判定该请求已完成，自动关闭响应流。你的代码中，方法在添加流到字典并发送第一条消息后直接返回，触发了这个逻辑。
2. 客户端连接提前断开：客户端Main方法没有等待readerTask完成，程序会直接退出，导致客户端连接断开，服务端感知后会标记请求完成。

二、如何实现后续向响应流写入数据？ #

需要从服务端、客户端两方面调整，同时保证线程安全：

服务端修改 #

核心是保持GetJobsRT方法活跃，直到客户端断开，同时管理流的生命周期：

public class JobService : Job.JobBase
{
    public static readonly Dictionary<string, IServerStreamWriter<JobsReply>> _grpcResponseStreams = new();
    private static readonly object _streamLock = new object(); // 保证字典操作线程安全

    public override async Task GetJobsRT(
        JobsRequest request,
        IServerStreamWriter<JobsReply> responseStream,
        ServerCallContext context)
    {
        var userId = request.UserId;
        // 线程安全地添加流到字典
        lock (_streamLock)
        {
            if (!_grpcResponseStreams.ContainsKey(userId))
            {
                _grpcResponseStreams.Add(userId, responseStream);
            }
        }

        try
        {
            await responseStream.WriteAsync(new JobsReply() { Message = "实时推送任务更新与执行状态" });
            
            // 等待客户端断开的信号，保持方法活跃
            await context.CancellationToken.WaitHandle.WaitOneAsync();
        }
        finally
        {
            // 客户端断开后，从字典移除无效流
            lock (_streamLock)
            {
                _grpcResponseStreams.Remove(userId);
            }
        }
    }
}

客户端修改 #

保证客户端连接持续活跃，等待读取任务完成：

static async Task Main(string[] args)
{
    var userId = Guid.NewGuid().ToString();
    using var channel = GrpcChannel.ForAddress("https://localhost:8080");
    var client = new Job.JobClient(channel);
    
    using var jobServerStreaming = client.GetJobsRT(new JobsRequest() { UserId = userId });
    try
    {                
        var readerTask = Task.Run(async () =>
        {
            try
            {
                await foreach (var res in jobServerStreaming.ResponseStream.ReadAllAsync())
                {
                    Console.WriteLine($"收到服务端消息: {res.Message}");
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine($"读取消息异常: {ex.Message}");
            }
        });

        // 等待用户输入，防止程序提前退出
        Console.WriteLine("按任意键退出...");
        Console.ReadKey();
        
        // 等待读取任务结束
        await readerTask;
    }
    catch (Exception ex)
    {
        Console.WriteLine($"客户端异常: {ex.Message}");
    }
}

事件触发方法优化 #

添加线程安全保护，避免并发问题：

async ValueTask AfterMessageReceived(Message context)
{
    try
    {
        List<IServerStreamWriter<JobsReply>> streamsToWrite;
        // 先复制流列表，避免迭代时字典被修改
        lock (JobService._streamLock)
        {
            streamsToWrite = JobService._grpcResponseStreams.Values.ToList();
        }

        foreach (var responseStream in streamsToWrite)
        {
            try
            {
                await responseStream.WriteAsync(new JobsReply() { Message = context.Message });
            }
            catch (Exception ex)
            {
                Console.WriteLine($"写入流失败: {ex.Message}");
            }
        }
    }
    catch (Exception ex)
    {
        Console.WriteLine($"处理消息异常: {ex.ToString()}");
    }
}

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