gRPC流式服务端方法如何实现客户端超时看门狗机制？

我在开发服务端的长运行gRPC Streaming方法，需求是：当客户端在指定时长（比如5分钟）内未发送请求时，服务端方法停止执行，且每次收到请求时重置该超时时间。当前代码框架如下：

public override async Task StreamingMethodName(IAsyncStreamReader<RequestProtoDTO> requestStream, IServerStreamWriter<ResponseProtoDTO> responseStream, ServerCallContext context)
{
    var cts = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken, WATCHDOG_CTS_GOES_HERE);
    while(await requestStream.MoveNext(cts.Token))
    {
        // 业务处理代码

       WATCHDOG_KICKING_GOES_HERE
    }
}

想问下有没有现成的CancellationToken看门狗实现，还是需要自行创建CancellationTokenSource？另外移动端客户端的HTTP2连接有时会断开但服务端无法察觉，我希望无需反向代理就能释放这些资源。

.NET 生态里没有直接开箱即用的这类看门狗式CancellationToken实现，你需要基于CancellationTokenSource自行封装一个可重置的超时控制逻辑。

实现一个可重置的看门狗CancellationTokenSource #

你可以封装一个简单的WatchdogCts类，核心逻辑是每次"踢"看门狗时，取消原有的CancellationTokenSource并重新创建新的实例：

public class WatchdogCts : IDisposable
{
    private readonly TimeSpan _timeout;
    private CancellationTokenSource _cts;
    private readonly object _lockObj = new object();

    public WatchdogCts(TimeSpan timeout)
    {
        _timeout = timeout;
        _cts = new CancellationTokenSource(timeout);
    }

    public CancellationToken Token => _cts.Token;

    public void Kick()
    {
        lock (_lockObj)
        {
            // 取消旧的超时任务，创建新的
            _cts.Cancel();
            _cts.Dispose();
            _cts = new CancellationTokenSource(_timeout);
        }
    }

    public void Dispose()
    {
        lock (_lockObj)
        {
            _cts.Dispose();
        }
    }
}

在gRPC方法中使用 #

把这个看门狗集成到你的gRPC方法里，替换掉原来的占位符：

public override async Task StreamingMethodName(IAsyncStreamReader<RequestProtoDTO> requestStream, IServerStreamWriter<ResponseProtoDTO> responseStream, ServerCallContext context)
{
    using var watchdog = new WatchdogCts(TimeSpan.FromMinutes(5));
    var cts = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken, watchdog.Token);
    
    try
    {
        while(await requestStream.MoveNext(cts.Token))
        {
            // 业务处理代码

            // 每次收到请求就重置超时
            watchdog.Kick();
        }
    }
    catch (OperationCanceledException)
    {
        // 处理超时或调用取消的逻辑，比如释放资源
    }
    finally
    {
        cts.Dispose();
    }
}

补充说明 #

• 这个实现通过lock保证线程安全，因为gRPC的回调可能在不同线程触发
• 当客户端断开连接但服务端没收到通知时，超时触发后会自动取消cts.Token，终止MoveNext的等待，进而释放服务端资源
• 不需要依赖反向代理，完全在服务端业务代码层面实现超时控制

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